mnesia_tm.erl

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

		    State2 = State#state{participants=gb_trees:delete(Tid,Participants)},
		    doit_loop(State2)
	    end;
	
	{sync_trans_serial, Tid} ->
	    %% Do the Lamport thing here
	    mnesia_recover:sync_trans_tid_serial(Tid),
	    doit_loop(State);
	    
	{From, info} ->
	    reply(From, {info, gb_trees:values(Participants), 
			 gb_trees:to_list(Coordinators)}, State);
	
	{mnesia_down, N} ->
	    verbose("Got mnesia_down from ~p, reconfiguring...~n", [N]),
	    reconfigure_coordinators(N, gb_trees:to_list(Coordinators)),
	    
	    Tids = gb_trees:keys(Participants),
	    reconfigure_participants(N, gb_trees:values(Participants)),
	    NewState = clear_fixtable(N, State),
	    mnesia_monitor:mnesia_down(?MODULE, {N, Tids}),
	    doit_loop(NewState);

	{From, {unblock_me, Tab}} ->
	    case lists:member(Tab, State#state.blocked_tabs) of
		false ->
		    verbose("Wrong dirty Op blocked on ~p ~p ~p", 
			    [node(), Tab, From]),
		    reply(From, unblocked),
		    doit_loop(State);
		true ->
		    Item = {Tab, unblock_me, From}, 
		    State2 = State#state{dirty_queue = [Item | State#state.dirty_queue]},
		    doit_loop(State2)
	    end;	    
	
	{From, {block_tab, Tab}} ->
	    State2 = State#state{blocked_tabs = [Tab | State#state.blocked_tabs]},
	    reply(From, ok, State2);
	
	{From, {unblock_tab, Tab}} ->
	    BlockedTabs2 = State#state.blocked_tabs -- [Tab],
	    case lists:member(Tab, BlockedTabs2) of
		false ->
		    mnesia_controller:unblock_table(Tab),
		    Queue = process_dirty_queue(Tab, State#state.dirty_queue),
		    State2 = State#state{blocked_tabs = BlockedTabs2, 
					 dirty_queue = Queue},
		    reply(From, ok, State2);
		true ->
		    State2 = State#state{blocked_tabs = BlockedTabs2},
		    reply(From, ok, State2)
	    end;
	
	{From, {prepare_checkpoint, Cp}} ->
	    Res = mnesia_checkpoint:tm_prepare(Cp),
	    case Res of
		{ok, _Name, IgnoreNew, _Node} ->
		    prepare_pending_coordinators(gb_trees:to_list(Coordinators), IgnoreNew),
		    prepare_pending_participants(gb_trees:values(Participants), IgnoreNew);
		{error, _Reason} ->
		    ignore
	    end,
	    reply(From, Res, State);
	{From, {fixtable, [Tab,Lock,Requester]}} ->
	    case ?catch_val({Tab, storage_type}) of
		{'EXIT', _} -> 
		    reply(From, error, State);
		Storage ->
		    mnesia_lib:db_fixtable(Storage,Tab,Lock),
		    NewState = manage_fixtable(Tab,Lock,Requester,State),
		    reply(From, node(), NewState)
	    end;
	
	{system, From, Msg} ->
	    dbg_out("~p got {system, ~p, ~p}~n", [?MODULE, From, Msg]),
	    sys:handle_system_msg(Msg, From, Sup, ?MODULE, [], State);
	
	Msg ->
	    verbose("** ERROR ** ~p got unexpected message: ~p~n", [?MODULE, Msg]),
	    doit_loop(State)
    end.

do_sync_dirty(From, Tid, Commit, _Tab) ->
    ?eval_debug_fun({?MODULE, sync_dirty, pre}, [{tid, Tid}]),
    Res = (catch do_dirty(Tid, Commit)),
    ?eval_debug_fun({?MODULE, sync_dirty, post}, [{tid, Tid}]),
    From ! {?MODULE, node(), {dirty_res, Res}}.

do_async_dirty(Tid, Commit, _Tab) ->
    ?eval_debug_fun({?MODULE, async_dirty, pre}, [{tid, Tid}]),
    catch do_dirty(Tid, Commit),
    ?eval_debug_fun({?MODULE, async_dirty, post}, [{tid, Tid}]).


%% Process items in fifo order
process_dirty_queue(Tab, [Item | Queue]) ->
    Queue2 = process_dirty_queue(Tab, Queue),
    case Item of
	{async_dirty, Tid, Commit, Tab} ->
	    do_async_dirty(Tid, Commit, Tab),
	    Queue2;
	{sync_dirty, From, Tid, Commit, Tab} ->
	    do_sync_dirty(From, Tid, Commit, Tab),
	    Queue2;
	{Tab, unblock_me, From} ->
	    reply(From, unblocked),
	    Queue2;
	_ ->
	    [Item | Queue2]
    end;
process_dirty_queue(_Tab, []) ->
    [].

prepare_pending_coordinators([{Tid, [Store | _Etabs]} | Coords], IgnoreNew) ->
    case catch ?ets_lookup(Store, pending) of
	[] ->
	    prepare_pending_coordinators(Coords, IgnoreNew);
	[Pending] ->
	    case lists:member(Tid, IgnoreNew) of
		false ->
		    mnesia_checkpoint:tm_enter_pending(Pending);
		true ->
		    ignore
	    end,
	    prepare_pending_coordinators(Coords, IgnoreNew);
	{'EXIT', _} -> 
	    prepare_pending_coordinators(Coords, IgnoreNew)
    end;
prepare_pending_coordinators([], _IgnoreNew) ->
    ok.

prepare_pending_participants([Part | Parts], IgnoreNew) ->
    Tid = Part#participant.tid,
    D = Part#participant.disc_nodes,
    R = Part#participant.ram_nodes,
    case lists:member(Tid, IgnoreNew) of
	false ->
	    mnesia_checkpoint:tm_enter_pending(Tid, D, R);
	true ->
	    ignore
    end,
    prepare_pending_participants(Parts, IgnoreNew);
prepare_pending_participants([], _IgnoreNew) ->
    ok.

handle_exit(Pid, _Reason, State) when node(Pid) /= node() ->
    %% We got exit from a remote fool
    doit_loop(State);

handle_exit(Pid, _Reason, State) when Pid == State#state.supervisor ->
    %% Our supervisor has died, time to stop
    do_stop(State);

handle_exit(Pid, Reason, State) ->
    %% Check if it is a coordinator
    case pid_search_delete(Pid, gb_trees:to_list(State#state.coordinators)) of 
	{none, _} ->
	    %% Check if it is a participant
	    Ps = gb_trees:values(State#state.participants),
	    case mnesia_lib:key_search_delete(Pid,#participant.pid,Ps) of
		{none, _} ->
		    %% We got exit from a local fool
		    doit_loop(State);
		{P, _RestP} when record(P, participant) ->
		    fatal("Participant ~p in transaction ~p died ~p~n",
			  [P#participant.pid, P#participant.tid, Reason]),
		    NewPs = gb_trees:delete(P#participant.tid,State#state.participants),
		    doit_loop(State#state{participants = NewPs})
	    end;
	
	{{Tid, Etabs}, RestC} ->
	    %% A local coordinator has died and 
	    %% we must determine the outcome of the
	    %% transaction and tell mnesia_tm on the
	    %% other nodes about it and then recover
	    %% locally.
	    recover_coordinator(Tid, Etabs),
	    doit_loop(State#state{coordinators = RestC})
    end.

recover_coordinator(Tid, Etabs) ->
    verbose("Coordinator ~p in transaction ~p died.~n", [Tid#tid.pid, Tid]),

    Store = hd(Etabs),
    CheckNodes = get_elements(nodes,Store),
    TellNodes = CheckNodes -- [node()],
    case catch arrange(Tid, Store, async) of
	{'EXIT', Reason} ->
	    dbg_out("Recovery of coordinator ~p failed:~n", [Tid, Reason]),
	    Protocol = asym_trans,
	    tell_outcome(Tid, Protocol, node(), CheckNodes, TellNodes);
	{_N, Prep} ->
	    %% Tell the participants about the outcome
	    Protocol = Prep#prep.protocol,
	    Outcome = tell_outcome(Tid, Protocol, node(), CheckNodes, TellNodes),
	    
	    %% Recover locally
	    CR = Prep#prep.records,
	    {DiscNs, RamNs} = commit_nodes(CR, [], []),
	    case lists:keysearch(node(), #commit.node, CR) of
		{value, Local} ->
		    ?eval_debug_fun({?MODULE, recover_coordinator, pre},
				    [{tid, Tid}, {outcome, Outcome}, {prot, Protocol}]),
		    recover_coordinator(Tid, Protocol, Outcome, Local, DiscNs, RamNs),
		    ?eval_debug_fun({?MODULE, recover_coordinator, post},
				    [{tid, Tid}, {outcome, Outcome}, {prot, Protocol}]);
		false ->  %% When killed before store havn't been copied to 
		    ok    %% to the new nested trans store.
	    end
    end,
    erase_ets_tabs(Etabs),
    transaction_terminated(Tid),
    mnesia_locker:release_tid(Tid).

recover_coordinator(Tid, sym_trans, committed, Local, _, _) ->
    mnesia_recover:note_decision(Tid, committed),
    do_dirty(Tid, Local);
recover_coordinator(Tid, sym_trans, aborted, _Local, _, _) ->
    mnesia_recover:note_decision(Tid, aborted);
recover_coordinator(Tid, sync_sym_trans, committed, Local, _, _) ->
    mnesia_recover:note_decision(Tid, committed),
    do_dirty(Tid, Local);
recover_coordinator(Tid, sync_sym_trans, aborted, _Local, _, _) ->
    mnesia_recover:note_decision(Tid, aborted);

recover_coordinator(Tid, asym_trans, committed, Local, DiscNs, RamNs) ->
    D = #decision{tid = Tid, outcome = committed,
		  disc_nodes = DiscNs, ram_nodes = RamNs},		
    mnesia_recover:log_decision(D),
    do_commit(Tid, Local);
recover_coordinator(Tid, asym_trans, aborted, Local, DiscNs, RamNs) ->
    D = #decision{tid = Tid, outcome = aborted,
 		  disc_nodes = DiscNs, ram_nodes = RamNs},		
    mnesia_recover:log_decision(D),
    do_abort(Tid, Local).

restore_stores(Coords, Tid, Store) ->
    Etstabs = gb_trees:get(Tid,Coords),
    Remaining  = lists:delete(Store, Etstabs),
    erase_ets_tabs(Remaining),
    gb_trees:update(Tid,[Store],Coords).

add_coord_store(Coords, Tid, Etab) ->
    Stores = gb_trees:get(Tid, Coords),
    gb_trees:update(Tid, [Etab|Stores], Coords).

del_coord_store(Coords, Tid, Current, Obsolete) ->
    Stores = gb_trees:get(Tid, Coords),
    Rest = 
    	case Stores of
    	    [Obsolete, Current | Tail] -> Tail;
    	    [Current, Obsolete | Tail] -> Tail
    	end,
    ?ets_delete_table(Obsolete),
    gb_trees:update(Tid, [Current|Rest], Coords).

erase_ets_tabs([H | T]) ->
    ?ets_delete_table(H),
    erase_ets_tabs(T);
erase_ets_tabs([]) -> 
    ok.

%% Clear one transactions all fixtables
clear_fixtable([Store|_]) ->
    Fixed = get_elements(fixtable, Store),
    lists:foreach(fun({Tab,Node}) ->
			  rpc:cast(Node, ?MODULE, fixtable, [Tab,false,self()]) 
		  end, Fixed).

%% Clear all fixtable Node have done
clear_fixtable(Node, State=#state{fixed_tabs = FT0}) ->
    case mnesia_lib:key_search_delete(Node, 1, FT0) of
	{none, _Ft} ->
	    State;
	{{Node,Tabs},FT} ->
	    lists:foreach(
	      fun(Tab) ->
		      case ?catch_val({Tab, storage_type}) of
			  {'EXIT', _} -> 
			      ignore;
			  Storage ->
			      mnesia_lib:db_fixtable(Storage,Tab,false)
		      end
	      end, Tabs),
	    State#state{fixed_tabs=FT}
    end.

manage_fixtable(Tab,true,Requester,State=#state{fixed_tabs = FT0}) ->
    Node = node(Requester),
    case mnesia_lib:key_search_delete(Node, 1, FT0) of
	{none, FT}->
	    State#state{fixed_tabs=[{Node, [Tab]}|FT]};
	{{Node,Tabs},FT} ->
	    State#state{fixed_tabs=[{Node, [Tab|Tabs]}|FT]}
    end;
manage_fixtable(Tab,false,Requester,State = #state{fixed_tabs = FT0}) ->
    Node = node(Requester),
    case mnesia_lib:key_search_delete(Node, 1, FT0) of 
	{none,_FT} -> State; % Hmm? Safeguard
	{{Node, Tabs0},FT} -> 
	    case lists:delete(Tab, Tabs0) of
		[] -> State#state{fixed_tabs=FT};
		Tabs -> State#state{fixed_tabs=[{Node,Tabs}|FT]}
	    end
    end.

%% Deletes a pid from a list of participants
%% or from a gb_trees of coordinators
%% {none, All} or {Tr, Rest} 
pid_search_delete(Pid, Trs) ->
    pid_search_delete(Pid, Trs, none, []).
pid_search_delete(Pid, [Tr = {Tid, _Ts} | Trs], _Val, Ack) when Tid#tid.pid == Pid ->
    pid_search_delete(Pid, Trs, Tr, Ack);
pid_search_delete(Pid, [Tr | Trs], Val, Ack) ->
    pid_search_delete(Pid, Trs, Val, [Tr | Ack]);

pid_search_delete(_Pid, [], Val, Ack) ->
    {Val, gb_trees:from_orddict(lists:reverse(Ack))}.
 
transaction_terminated(Tid)  ->
    mnesia_checkpoint:tm_exit_pending(Tid),
    Pid = Tid#tid.pid,
    if
	node(Pid) == node() ->
	    unlink(Pid);
	true ->  %% Do the Lamport thing here
	    mnesia_recover:sync_trans_tid_serial(Tid)
    end.

%% If there are an surrounding transaction, we inherit it's context
non_transaction(OldState={_,_,Trans}, Fun, Args, ActivityKind, Mod) 
  when Trans /= non_transaction ->
    Kind = case ActivityKind of 
	       sync_dirty -> sync;
	       _ -> async
	   end,
    case transaction(OldState, Fun, Args, infinity, Mod, Kind) of
	{atomic, Res} -> 
	    Res;
	{aborted,Res} ->
	    exit(Res)
    end;
non_transaction(OldState, Fun, Args, ActivityKind, Mod) ->
    Id = {ActivityKind, self()},
    NewState = {Mod, Id, non_transaction},
    put(mnesia_activity_state, NewState),
    %% I Want something uniqe here, references are expensive
    Ref = mNeSia_nOn_TrAnSacTioN,
    RefRes = (catch {Ref, apply(Fun, Args)}),
    case OldState of
	undefined -> erase(mnesia_activity_state);
	_ -> put(mnesia_activity_state, OldState)
    end,
    case RefRes of
	{Ref, Res} ->
	    case Res of
		{'EXIT', Reason} -> exit(Reason);
		{aborted, Reason} -> mnesia:abort(Reason);
		_ -> Res
	    end;
	{'EXIT', Reason} ->
	    exit(Reason);
	Throw ->
	    throw(Throw)
    end.

transaction(OldTidTs, Fun, Args, Retries, Mod, Type) ->
    Factor = 1,
    case OldTidTs of
	undefined -> % Outer
	    execute_outer(Mod, Fun, Args, Factor, Retries, Type);
	{_, _, non_transaction} -> % Transaction inside ?sync_dirty
	    Res = execute_outer(Mod, Fun, Args, Factor, Retries, Type),
	    put(mnesia_activity_state, OldTidTs),
	    Res;
	{OldMod, Tid, Ts} ->  % Nested
	    execute_inner(Mod, Tid, OldMod, Ts, Fun, Args, Factor, Retries, Type);
	_ -> % Bad nesting
	    {aborted, nested_transaction}