mnesia_recover.erl

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

		false ->
		    {noreply, State};
		true ->
		    State2 = add_remote_decision(Node, D, State),
		    {noreply, State2}
		end
    end;

handle_cast({announce_all, Nodes}, State) ->
    announce_all(Nodes),
    {noreply, State};

handle_cast(Msg, State) ->
    error("~p got unexpected cast: ~p~n", [?MODULE, Msg]),
    {noreply, State}.

%%----------------------------------------------------------------------
%% Func: handle_info/2
%% Returns: {noreply, State}          |
%%          {noreply, State, Timeout} |
%%          {stop, Reason, State}            (terminate/2 is called)
%%----------------------------------------------------------------------

%% No need for buffering
%% handle_info(Msg, State) when State#state.initiated == false ->
%%     %% Buffer early messages
%%     Msgs = State#state.early_msgs,
%%     {noreply, State#state{early_msgs = [{info, Msg} | Msgs]}};

handle_info({connect_nodes, Ns, From}, State) ->
    handle_call({connect_nodes,Ns},From,State);

handle_info(check_overload, S) ->
    %% Time to check if mnesia_tm is overloaded
    case whereis(mnesia_tm) of
	Pid when pid(Pid) ->
	    
	    Threshold = 100,
	    Prev = S#state.tm_queue_len,
	    {message_queue_len, Len} =
		process_info(Pid, message_queue_len),
	    if
		Len > Threshold, Prev > Threshold ->
		    What = {mnesia_tm, message_queue_len, [Prev, Len]},
		    mnesia_lib:report_system_event({mnesia_overload, What}),
		    {noreply, S#state{tm_queue_len = 0}};
		
		Len > Threshold ->
		    {noreply, S#state{tm_queue_len = Len}};
		
		true ->
		    {noreply, S#state{tm_queue_len = 0}}
	    end;
	undefined ->
	    {noreply, S}
    end;

handle_info(garb_decisions, State) ->
    do_garb_decisions(),
    {noreply, State};

handle_info({force_decision, Tid}, State) ->
    %% Enforce a transaction recovery decision,
    %% if we still are waiting for the outcome
    
    case State#state.unclear_decision of
	U when U#decision.tid == Tid ->
	    verbose("Decided to abort transaction ~p since "
		    "max_wait_for_decision has been exceeded~n",
		    [Tid]),
	    D = U#decision{outcome = aborted},
	    State2 = add_remote_decision(node(), D, State),
	    {noreply, State2};
	_ ->
	    {noreply, State}
    end;

handle_info({'EXIT', Pid, R}, State) when Pid == State#state.supervisor ->
    mnesia_lib:dbg_out("~p was ~p~n",[?MODULE, R]),
    {stop, shutdown, State};

handle_info(Msg, State) ->
    error("~p got unexpected info: ~p~n", [?MODULE, Msg]),
    {noreply, State}.

%%----------------------------------------------------------------------
%% Func: terminate/2
%% Purpose: Shutdown the server
%% Returns: any (ignored by gen_server)
%%----------------------------------------------------------------------

terminate(Reason, State) ->
    mnesia_monitor:terminate_proc(?MODULE, Reason, State).

%%----------------------------------------------------------------------
%% Func: code_change/3
%% Purpose: Upgrade process when its code is to be changed
%% Returns: {ok, NewState}
%%----------------------------------------------------------------------
code_change(_OldVsn, State, _Extra) ->
    {ok, State}.

%%%----------------------------------------------------------------------
%%% Internal functions
%%%----------------------------------------------------------------------

handle_early_msgs(State, From) ->
    Res = do_handle_early_msgs(State#state.early_msgs,
			       State#state{early_msgs = [],
					   initiated = true}),
    gen_server:reply(From, ok),
    Res.

do_handle_early_msgs([Msg | Msgs], State) ->
    %% The messages are in reverted order
    case do_handle_early_msgs(Msgs, State) of
%%         {stop, Reason, Reply, State2} ->
%% 	    {stop, Reason, Reply, State2};
        {stop, Reason, State2} ->
	    {stop, Reason, State2};
	{noreply, State2} ->
	    handle_early_msg(Msg, State2)
    end;

do_handle_early_msgs([], State) ->
    {noreply, State}.
    
handle_early_msg({call, Msg, From}, State) ->
    case handle_call(Msg, From, State) of
	{reply, R, S} ->
	    gen_server:reply(From, R),
	    {noreply, S};
	Other ->
	    Other
    end;
handle_early_msg({cast, Msg}, State) ->
    handle_cast(Msg, State);
handle_early_msg({info, Msg}, State) ->
    handle_info(Msg, State).

tabs() ->
    Curr = val(latest_transient_decision),    % Do not miss any trans even
    Prev = val(previous_transient_decisions), % if the tabs are switched
    [Curr, mnesia_decision | Prev].           % Ordered by hit probability

decision(Tid) ->
    decision(Tid, tabs()).

decision(Tid, [Tab | Tabs]) ->
    case catch ?ets_lookup(Tab, Tid) of
	[D] when record(D, decision) ->
	    D;
	[C] when record(C, transient_decision) ->
	    #decision{tid = C#transient_decision.tid,
		      outcome =  C#transient_decision.outcome,
		      disc_nodes = [],
		      ram_nodes = []
		     };
	[] ->
	    decision(Tid, Tabs);
	{'EXIT', _} ->
	    %% Recently switched transient decision table
	    decision(Tid, Tabs)
    end;
decision(_Tid, []) ->
    no_decision.

outcome(Tid, Default) ->
    outcome(Tid, Default, tabs()).

outcome(Tid, Default, [Tab | Tabs]) ->
    case catch ?ets_lookup_element(Tab, Tid, 3) of
	{'EXIT', _} ->
	    outcome(Tid, Default, Tabs);
	Val ->
	    Val
    end;
outcome(_Tid, Default, []) ->
    Default.

filter_outcome(Val) ->
    case Val of
	unclear -> unclear;
	aborted -> aborted;
	presume_abort -> aborted;
	committed -> committed;
	pre_commit -> unclear
    end.

filter_aborted(D) when D#decision.outcome == presume_abort ->
    D#decision{outcome = aborted};
filter_aborted(D) ->
    D.
  
%% Merge old decision D with new (probably remote) decision
merge_decisions(Node, D, NewD0) ->
    NewD = filter_aborted(NewD0),
    if
	D == no_decision, node() /= Node ->
	    %% We did not know anything about this txn
	    NewD#decision{disc_nodes = []};
	D == no_decision ->
	    NewD;
	record(D, decision) ->
	    DiscNs = D#decision.disc_nodes -- ([node(), Node]),
	    OldD = filter_aborted(D#decision{disc_nodes = DiscNs}),
%%	    mnesia_lib:dbg_out("merge ~w: NewD = ~w~n D = ~w~n OldD = ~w~n", 
%%			       [Node, NewD, D, OldD]),
	    if
		OldD#decision.outcome == unclear,
		NewD#decision.outcome == unclear ->
		    D;

		OldD#decision.outcome == NewD#decision.outcome ->
		    %% We have come to the same decision
		    OldD;

		OldD#decision.outcome == committed,
		NewD#decision.outcome == aborted ->
		    %% Interesting! We have already committed,
		    %% but someone else has aborted. Now we
		    %% have a nice little inconcistency. The
		    %% other guy (or some one else) has 
		    %% enforced a recovery decision when
		    %% max_wait_for_decision was exceeded.
		    %% We will pretend that we have obeyed
		    %% the forced recovery decision, but we
		    %% will also generate an event in case the
		    %% application wants to do something clever.
		    Msg = {inconsistent_database, bad_decision, Node},
		    mnesia_lib:report_system_event(Msg),
		    OldD#decision{outcome = aborted};

		OldD#decision.outcome == aborted ->
		    %% aborted overrrides anything
		    OldD#decision{outcome = aborted};

		NewD#decision.outcome == aborted ->
		    %% aborted overrrides anything
		    OldD#decision{outcome = aborted};

		OldD#decision.outcome == committed,
		NewD#decision.outcome == unclear ->
		    %% committed overrides unclear
		    OldD#decision{outcome = committed};

		OldD#decision.outcome == unclear,
		NewD#decision.outcome == committed ->
		    %% committed overrides unclear
		    OldD#decision{outcome = committed}
	    end
    end.

add_remote_decisions(Node, [D | Tail], State) when record(D, decision) ->
    State2 = add_remote_decision(Node, D, State),
    add_remote_decisions(Node, Tail, State2);

add_remote_decisions(Node, [C | Tail], State)
        when record(C, transient_decision) ->
    D = #decision{tid = C#transient_decision.tid,
		  outcome = C#transient_decision.outcome,
		  disc_nodes = [],
		  ram_nodes = []},
    State2 = add_remote_decision(Node, D, State),
    add_remote_decisions(Node, Tail, State2);

add_remote_decisions(Node, [{mnesia_down, _, _, _} | Tail], State) ->
    add_remote_decisions(Node, Tail, State);

add_remote_decisions(Node, [{trans_tid, serial, Serial} | Tail], State) ->
    sync_trans_tid_serial(Serial),
    case State#state.unclear_decision of
	undefined ->
	    ignored;
	D ->
	    case lists:member(Node, D#decision.ram_nodes) of
		true ->
		    ignore;
		false ->
		    abcast([Node], {what_decision, node(), D})
	    end
    end,
    add_remote_decisions(Node, Tail, State);

add_remote_decisions(_Node, [], State) ->
    State.

add_remote_decision(Node, NewD, State) ->
    Tid = NewD#decision.tid,
    OldD = decision(Tid),
    D = merge_decisions(Node, OldD, NewD),
    do_log_decision(D, false, undefined),
    Outcome = D#decision.outcome,
    if
	OldD == no_decision ->
	    ignore;
	Outcome == unclear ->
	    ignore;
	true ->
	    case lists:member(node(), NewD#decision.disc_nodes) or
		 lists:member(node(), NewD#decision.ram_nodes) of
		true ->
		    tell_im_certain([Node], D);
		false ->
		    ignore
	    end
    end,
    case State#state.unclear_decision of
	U when U#decision.tid == Tid ->
	    WaitFor = State#state.unclear_waitfor -- [Node],
	    if
		Outcome == unclear, WaitFor == [] ->
		    %% Everybody are uncertain, lets abort
		    NewOutcome = aborted,
		    CertainD = D#decision{outcome = NewOutcome, 
					  disc_nodes = [],
					  ram_nodes = []},
		    tell_im_certain(D#decision.disc_nodes, CertainD),
		    tell_im_certain(D#decision.ram_nodes, CertainD),
		    do_log_decision(CertainD, false, undefined),
		    verbose("Decided to abort transaction ~p "
			    "since everybody are uncertain ~p~n",
			    [Tid, CertainD]),
		    gen_server:reply(State#state.unclear_pid, {ok, NewOutcome}),
		    State#state{unclear_pid = undefined,
				unclear_decision = undefined,
				unclear_waitfor = undefined};
		Outcome /= unclear ->
		    verbose("~p told us that transaction ~p was ~p~n",
			    [Node, Tid, Outcome]),
		    gen_server:reply(State#state.unclear_pid, {ok, Outcome}),
		    State#state{unclear_pid = undefined,
				unclear_decision = undefined,
				unclear_waitfor = undefined};
		Outcome == unclear ->
		    State#state{unclear_waitfor = WaitFor}
	    end;
	_ ->
	    State
    end.

announce_all([]) ->
    ok;
announce_all(ToNodes) ->
    Tid = trans_tid_serial(),
    announce(ToNodes, [{trans_tid,serial,Tid}], [], false).
    
announce(ToNodes, [Head | Tail], Acc, ForceSend) ->
    Acc2 = arrange(ToNodes, Head, Acc, ForceSend),
    announce(ToNodes, Tail, Acc2, ForceSend);

announce(_ToNodes, [], Acc, _ForceSend) ->
    send_decisions(Acc).

send_decisions([{Node, Decisions} | Tail]) ->
    abcast([Node], {decisions, node(), Decisions}),
    send_decisions(Tail);
send_decisions([]) ->
    ok.

arrange([To | ToNodes], D, Acc, ForceSend) when record(D, decision) ->
    NeedsAdd = (ForceSend or
		lists:member(To, D#decision.disc_nodes) or
		lists:member(To, D#decision.ram_nodes)),
    case NeedsAdd of
	true ->
	    Acc2 = add_decision(To, D, Acc),
	    arrange(ToNodes, D, Acc2, ForceSend);
	false ->
	    arrange(ToNodes, D, Acc, ForceSend)
    end;

arrange([To | ToNodes], {trans_tid, serial, Serial}, Acc, ForceSend) ->
    %% Do the lamport thing plus release the others
    %% from uncertainity.
    Acc2 = add_decision(To, {trans_tid, serial, Serial}, Acc),
    arrange(ToNodes, {trans_tid, serial, Serial}, Acc2, ForceSend);

arrange([], _Decision, Acc, _ForceSend) ->
    Acc.

add_decision(Node, Decision, [{Node, Decisions} | Tail]) ->
    [{Node, [Decision | Decisions]} | Tail];
add_decision(Node, Decision, [Head | Tail]) ->
    [Head | add_decision(Node, Decision, Tail)];
add_decision(Node, Decision, []) ->
    [{Node, [Decision]}].