mnesia_tm.erl

来自「OTP是开放电信平台的简称」· ERL 代码 · 共 1,898 行 · 第 1/5 页

	    end;
	{New, Prepared} ->
	    {New, Prepared#prep{records = reverse(Prepared#prep.records)}}
    end.

reverse([]) ->
    [];
reverse([H|R]) when record(H, commit) ->
    [
     H#commit{
       ram_copies       =  lists:reverse(H#commit.ram_copies),
       disc_copies      =  lists:reverse(H#commit.disc_copies),
       disc_only_copies =  lists:reverse(H#commit.disc_only_copies),
       snmp             = lists:reverse(H#commit.snmp)
      }  
     | reverse(R)].

prep_recs([N | Nodes], Recs) ->
    prep_recs(Nodes, [#commit{decision = presume_commit, node = N} | Recs]);
prep_recs([], Recs) ->
    Recs.

%% storage_types is a list of {Node, Storage} tuples
%% where each tuple represents an active replica
do_arrange(Tid, Store, {Tab, Key}, Prep, N) ->
    Oid = {Tab, Key},
    Items = ?ets_lookup(Store, Oid), %% Store is a bag
    P2 = prepare_items(Tid, Tab, Key, Items, Prep),
    do_arrange(Tid, Store, ?ets_next(Store, Oid), P2, N + 1);
do_arrange(Tid, Store, SchemaKey, Prep, N) when SchemaKey == op ->
    Items = ?ets_lookup(Store, SchemaKey), %% Store is a bag
    P2 = prepare_schema_items(Tid, Items, Prep),
    do_arrange(Tid, Store, ?ets_next(Store, SchemaKey), P2, N + 1);
do_arrange(Tid, Store, RestoreKey, Prep, N) when RestoreKey == restore_op ->
    [{restore_op, R}] = ?ets_lookup(Store, RestoreKey),
    Fun = fun({Tab, Key}, CommitRecs, _RecName, Where, Snmp) ->
		  Item = [{{Tab, Key}, {Tab, Key}, delete}],
		  do_prepare_items(Tid, Tab, Key, Where, Snmp, Item, CommitRecs);
	     (BupRec, CommitRecs, RecName, Where, Snmp) ->
		  Tab = element(1, BupRec),
		  Key = element(2, BupRec),
		  Item = 
		      if
			  Tab == RecName ->
			      [{{Tab, Key}, BupRec, write}];
			  true ->
			      BupRec2 = setelement(1, BupRec, RecName),
			      [{{Tab, Key}, BupRec2, write}]
		      end,
		  do_prepare_items(Tid, Tab, Key, Where, Snmp, Item, CommitRecs)
	  end,	  
    Recs2 = mnesia_schema:arrange_restore(R, Fun, Prep#prep.records),
    P2 = Prep#prep{protocol = asym_trans, records = Recs2},
    do_arrange(Tid, Store, ?ets_next(Store, RestoreKey), P2, N + 1);
do_arrange(_Tid, _Store, '$end_of_table', Prep, N) ->
    {N, Prep};
do_arrange(Tid, Store, IgnoredKey, Prep, N) -> %% locks, nodes ... local atoms...
    do_arrange(Tid, Store, ?ets_next(Store, IgnoredKey), Prep, N).

%% Returns a prep record  with all items in reverse order
prepare_schema_items(Tid, Items, Prep) ->
    Types = [{N, schema_ops} || N <- val({current, db_nodes})],
    Recs = prepare_nodes(Tid, Types, Items, Prep#prep.records, schema),
    Prep#prep{protocol = asym_trans, records = Recs}.

%% Returns a prep record with all items in reverse order
prepare_items(Tid, Tab, Key, Items, Prep) when Prep#prep.prev_tab == Tab ->
    Types = Prep#prep.prev_types,
    Snmp = Prep#prep.prev_snmp,
    Recs = Prep#prep.records,
    Recs2 = do_prepare_items(Tid, Tab, Key, Types, Snmp, Items, Recs),
    Prep#prep{records = Recs2};
    
prepare_items(Tid, Tab, Key, Items, Prep) ->
    Types = val({Tab, where_to_commit}),
    case Types of
	[] -> mnesia:abort({no_exists, Tab});
	{blocked, _} -> 
	    unblocked = req({unblock_me, Tab}),
	    prepare_items(Tid, Tab, Key, Items, Prep);
	_ ->     
	    Snmp = val({Tab, snmp}),
	    Recs2 = do_prepare_items(Tid, Tab, Key, Types, 
				     Snmp, Items, Prep#prep.records),
	    Prep2 = Prep#prep{records = Recs2, prev_tab = Tab, 
			      prev_types = Types, prev_snmp = Snmp},
	    check_prep(Prep2, Types)
    end.

do_prepare_items(Tid, Tab, Key, Types, Snmp, Items, Recs) ->
    Recs2 = prepare_snmp(Tid, Tab, Key, Types, Snmp, Items, Recs), % May exit
    prepare_nodes(Tid, Types, Items, Recs2, normal).

prepare_snmp(Tab, Key, Items) ->
    case val({Tab, snmp}) of 
	[] ->
	    [];
	Ustruct when Key /= '_' ->
	    {_Oid, _Val, Op} = hd(Items),
	    %% Still making snmp oid (not used) because we want to catch errors here
	    %% And also it keeps backwards comp. with old nodes.
	    SnmpOid = mnesia_snmp_hook:key_to_oid(Tab, Key, Ustruct), % May exit
	    [{Op, Tab, Key, SnmpOid}];
	_ ->
	    [{clear_table, Tab}]
    end.

prepare_snmp(_Tid, _Tab, _Key, _Types, [], _Items, Recs) -> 
    Recs;

prepare_snmp(Tid, Tab, Key, Types, Us, Items, Recs) -> 
    if Key /= '_' ->
	    {_Oid, _Val, Op} = hd(Items),
	    SnmpOid = mnesia_snmp_hook:key_to_oid(Tab, Key, Us), % May exit
	    prepare_nodes(Tid, Types, [{Op, Tab, Key, SnmpOid}], Recs, snmp);
       Key == '_' ->
	    prepare_nodes(Tid, Types, [{clear_table, Tab}], Recs, snmp)
    end.

check_prep(Prep, Types) when Prep#prep.types == Types ->
    Prep;
check_prep(Prep, Types) when Prep#prep.types == undefined ->
    Prep#prep{types = Types};
check_prep(Prep, _Types) ->
    Prep#prep{protocol = asym_trans}.

%% Returns a list of commit records
prepare_nodes(Tid, [{Node, Storage} | Rest], Items, C, Kind) ->
    {Rec, C2} = pick_node(Tid, Node, C, []),
    Rec2 = prepare_node(Node, Storage, Items, Rec, Kind),
    [Rec2 | prepare_nodes(Tid, Rest, Items, C2, Kind)];
prepare_nodes(_Tid, [], _Items, CommitRecords, _Kind) ->
    CommitRecords.

pick_node(Tid, Node, [Rec | Rest], Done) ->
    if
	Rec#commit.node == Node ->
	    {Rec, Done ++ Rest};
	true ->
	    pick_node(Tid, Node, Rest, [Rec | Done])
    end;
pick_node({dirty,_}, Node, [], Done) ->
    {#commit{decision = presume_commit, node = Node}, Done};
pick_node(_Tid, Node, [], _Done) ->
    mnesia:abort({bad_commit, {missing_lock, Node}}).

prepare_node(Node, Storage, [Item | Items], Rec, Kind) when Kind == snmp ->
    Rec2 = Rec#commit{snmp = [Item | Rec#commit.snmp]},
    prepare_node(Node, Storage, Items, Rec2, Kind);
prepare_node(Node, Storage, [Item | Items], Rec, Kind) when Kind /= schema ->
    Rec2 = 
	case Storage of
	    ram_copies ->
		Rec#commit{ram_copies = [Item | Rec#commit.ram_copies]};
	    disc_copies ->
		Rec#commit{disc_copies = [Item | Rec#commit.disc_copies]};
	    disc_only_copies ->
		Rec#commit{disc_only_copies =
			   [Item | Rec#commit.disc_only_copies]}
	end,
    prepare_node(Node, Storage, Items, Rec2, Kind);
prepare_node(_Node, _Storage, Items, Rec, Kind) 
  when Kind == schema, Rec#commit.schema_ops == []  ->
    Rec#commit{schema_ops = Items};
prepare_node(_Node, _Storage, [], Rec, _Kind) ->
    Rec.

%% multi_commit((Protocol, Tid, CommitRecords, Store)
%% Local work is always performed in users process
multi_commit(read_only, Tid, CR, _Store) ->
    %% This featherweight commit protocol is used when no 
    %% updates has been performed in the transaction.

    {DiscNs, RamNs} = commit_nodes(CR, [], []),
    Msg = {Tid, simple_commit},
    rpc:abcast(DiscNs -- [node()], ?MODULE, Msg),
    rpc:abcast(RamNs -- [node()], ?MODULE, Msg),
    mnesia_recover:note_decision(Tid, committed),
    mnesia_locker:release_tid(Tid),
    ?MODULE ! {delete_transaction, Tid},
    do_commit;

multi_commit(sym_trans, Tid, CR, Store) ->
    %% This lightweight commit protocol is used when all
    %% the involved tables are replicated symetrically.
    %% Their storage types must match on each node.
    %%
    %% 1  Ask the other involved nodes if they want to commit
    %%    All involved nodes votes yes if they are up
    %% 2a Somebody has voted no
    %%    Tell all yes voters to do_abort
    %% 2b Everybody has voted yes
    %%    Tell everybody to do_commit. I.e. that they should
    %%    prepare the commit, log the commit record and
    %%    perform the updates.
    %%
    %%    The outcome is kept 3 minutes in the transient decision table.
    %%    
    %% Recovery:
    %%    If somebody dies before the coordinator has
    %%    broadcasted do_commit, the transaction is aborted.
    %%    
    %%    If a participant dies, the table load algorithm
    %%    ensures that the contents of the involved tables
    %%    are picked from another node.
    %%
    %%    If the coordinator dies, each participants checks
    %%    the outcome with all the others. If all are uncertain
    %%    about the outcome, the transaction is aborted. If
    %%    somebody knows the outcome the others will follow.
    
    {DiscNs, RamNs} = commit_nodes(CR, [], []),
    Pending = mnesia_checkpoint:tm_enter_pending(Tid, DiscNs, RamNs),
    ?ets_insert(Store, Pending),

    {WaitFor, Local} = ask_commit(sym_trans, Tid, CR, DiscNs, RamNs),
    {Outcome, []} = rec_all(WaitFor, Tid, do_commit, []), 
    ?eval_debug_fun({?MODULE, multi_commit_sym}, 
		    [{tid, Tid}, {outcome, Outcome}]), 
    rpc:abcast(DiscNs -- [node()], ?MODULE, {Tid, Outcome}),
    rpc:abcast(RamNs -- [node()], ?MODULE, {Tid, Outcome}),
    case Outcome of
	do_commit ->
	    mnesia_recover:note_decision(Tid, committed),
	    do_dirty(Tid, Local),
	    mnesia_locker:release_tid(Tid),
	    ?MODULE ! {delete_transaction, Tid};
	{do_abort, _Reason} ->
	    mnesia_recover:note_decision(Tid, aborted)
    end,
    ?eval_debug_fun({?MODULE, multi_commit_sym, post},
		    [{tid, Tid}, {outcome, Outcome}]),
    Outcome;

multi_commit(sync_sym_trans, Tid, CR, Store) ->
    %%   This protocol is the same as sym_trans except that it
    %%   uses syncronized calls to disk_log and syncronized commits
    %%   when several nodes are involved.
    
    {DiscNs, RamNs} = commit_nodes(CR, [], []),
    Pending = mnesia_checkpoint:tm_enter_pending(Tid, DiscNs, RamNs),
    ?ets_insert(Store, Pending),

    {WaitFor, Local} = ask_commit(sync_sym_trans, Tid, CR, DiscNs, RamNs),
    {Outcome, []} = rec_all(WaitFor, Tid, do_commit, []), 
    ?eval_debug_fun({?MODULE, multi_commit_sym_sync}, 
		    [{tid, Tid}, {outcome, Outcome}]), 
    rpc:abcast(DiscNs -- [node()], ?MODULE, {Tid, Outcome}),
    rpc:abcast(RamNs -- [node()], ?MODULE, {Tid, Outcome}),
    case Outcome of
	do_commit ->
	    mnesia_recover:note_decision(Tid, committed),
	    mnesia_log:slog(Local),
	    do_commit(Tid, Local),
	    %% Just wait for completion result is ignore.
	    rec_all(WaitFor, Tid, ignore, []),
	    mnesia_locker:release_tid(Tid),
	    ?MODULE ! {delete_transaction, Tid};
	{do_abort, _Reason} ->
	    mnesia_recover:note_decision(Tid, aborted)
    end,
    ?eval_debug_fun({?MODULE, multi_commit_sym, post},
		    [{tid, Tid}, {outcome, Outcome}]),
    Outcome;

multi_commit(asym_trans, Tid, CR, Store) ->
    %% This more expensive commit protocol is used when 
    %% table definitions are changed (schema transactions).
    %% It is also used when the involved tables are
    %% replicated asymetrically. If the storage type differs
    %% on at least one node this protocol is used.
    %%
    %% 1 Ask the other involved nodes if they want to commit.
    %%   All involved nodes prepares the commit, logs a presume_abort
    %%   commit record and votes yes or no depending of the
    %%   outcome of the prepare. The preparation is also performed
    %%   by the coordinator.
    %%   
    %% 2a Somebody has died or voted no
    %%    Tell all yes voters to do_abort
    %% 2b Everybody has voted yes
    %%    Put a unclear marker in the log.
    %%    Tell the others to pre_commit. I.e. that they should
    %%    put a unclear marker in the log and reply
    %%    acc_pre_commit when they are done. 
    %%
    %% 3a Somebody died
    %%    Tell the remaining participants to do_abort
    %% 3b Everybody has replied acc_pre_commit
    %%    Tell everybody to committed. I.e that they should
    %%    put a committed marker in the log, perform the updates
    %%    and reply done_commit when they are done. The coordinator
    %%    must wait with putting his committed marker inte the log
    %%    until the committed has been sent to all the others.
    %%    Then he performs local commit before collecting replies.
    %%
    %% 4  Everybody has either died or replied done_commit
    %%    Return to the caller.
    %%
    %% Recovery:
    %%    If the coordinator dies, the participants (and
    %%    the coordinator when he starts again) must do
    %%    the following:
    %%
    %%    If we have no unclear marker in the log we may
    %%    safely abort, since we know that nobody may have
    %%    decided to commit yet.
    %%    
    %%    If we have a committed marker in the log we may
    %%    safely commit since we know that everybody else
    %%    also will come to this conclusion.
    %%
    %%    If we have a unclear marker but no committed
    %%    in the log we are uncertain about the real outcome
    %%    of the transaction and must ask the others before
    %%    we can decide what to do. If someone knows the
    %%    outcome we will do the same. If nobody knows, we
    %%    will wait for the remaining involved nodes to come
    %%    up. When all involved nodes are up and uncertain,
    %%    we decide to commit (first put a committed marker
    %%    in the log, then do the updates).
    
    D = #decision{tid = Tid, outcome = presume_abort},
    {D2, CR2} = commit_decision(D, CR, [], []),
    DiscNs = D2#decision.disc_nodes,
    RamNs = D2#decision.ram_nodes,
    Pending = mnesia_checkpoint:tm_enter_pending(Tid, DiscNs, RamNs),
    ?ets_insert(Store, Pending),
    {WaitFor, Local} = ask_commit(asym_trans, Tid, CR2, DiscNs, RamNs),
    SchemaPrep = (catch mnesia_schema:prepare_commit(Tid, Local, {coord, WaitFor})), 
    {Votes, Pids} = rec_all(WaitFor, Tid, do_commit, []), 
    
    ?eval_debug_fun({?MODULE, multi_commit_asym_got_votes}, 
		    [{tid, Tid}, {votes, Votes}]),
    case Votes of
	do_commit ->
	    case SchemaPrep of
		{_Modified, C, DumperMode} when record(C, commit) ->
		    mnesia_log:log(C), % C is not a binary
		    ?eval_debug_fun({?MODULE, multi_commit_asym_log_commit_rec},
				    [{tid, Tid}]),
		    
		    D3 = C#commit.decision,
		    D4 = D3#decision{outcome = unclear},		
		    mnesia_recover:log_decision(D4),			
		    ?eval_debug_fun({?MODULE, multi_commit_asym_log_commit_dec},
				    [{tid, Tid}]),
		    tell_participants(Pids, {Tid, pre_commit}),
		    %% Now we are uncertain and we do not know
		    %% if all participants have logged that
		    %% they are uncertain or not
		    rec_acc_pre_commit(Pids, Tid, Store, {C,Local}, 
				       do_commit, DumperMode, [], []);
		{'EXIT', Reason} ->
		    %% The others have logged the commit 
		    %% record but they are not uncertain
		    mnesia_recover:note_decision(Tid, aborted),
		    ?eval_debug_fun({?MODULE, multi_commit_asym_prepare_exit},
				    [{tid, Tid}]),
		    tell_participants(Pids, {Tid, {do_abort, Reason}}),
		    do_abort(Tid, Local),
		    {do_abort, Reason}
	    end;

	{do_abort, Reason} ->
	    %% The others have logged the commit
	    %% record but they are not uncertain
	    mnesia_recover:note_decision(Tid, aborted),
	    ?eval_debug_fun({?MODULE, multi_commit_asym_do_abort}, [{tid, Tid}]),
	    tell_participants(Pids, {Tid, {do_abort, Reason}}),
	    do_abort(Tid, Local),
	    {do_abort, Reason}
    end.

%% Returns do_commit or {do_abort, Reason}
rec_acc_pre_commit([Pid | Tail], Tid, Store, Commit, Res, DumperMode, 
		   GoodPids, SchemaAckPids) ->
    receive
	{?MODULE, _, {acc_pre_commit, Tid, Pid, true}} ->
	    rec_acc_pre_commit(Tail, Tid, Store, Commit, Res, DumperMode,