mnesia_recover.erl

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

wait_for_decision(D, InitBy) when D#decision.outcome == presume_abort ->
    %% asym_trans
    Tid = D#decision.tid,
    Outcome = filter_outcome(outcome(Tid, D#decision.outcome)),
    if 
	Outcome /= unclear ->
	    {Tid, Outcome};

	InitBy /= startup ->
	    %% Wait a while for active transactions
	    %% to end and try again
	    timer:sleep(200), 
	    wait_for_decision(D, InitBy);

	InitBy == startup ->
	    {ok, Res} = call({wait_for_decision, D}),
	    {Tid, Res}
    end.

still_pending([Tid | Pending]) ->
    case filter_outcome(outcome(Tid, unclear)) of
	unclear -> [Tid | still_pending(Pending)];
	_ -> still_pending(Pending)
    end;
still_pending([]) ->
    [].

load_decision_tab() ->
    Cont = mnesia_log:open_decision_tab(),
    load_decision_tab(Cont, load_decision_tab),
    mnesia_log:close_decision_tab().

load_decision_tab(eof, _InitBy) ->
    ok;
load_decision_tab(Cont, InitBy) ->
    case mnesia_log:chunk_decision_tab(Cont) of
	{Cont2, Decisions} ->
	    note_log_decisions(Decisions, InitBy),
	    load_decision_tab(Cont2, InitBy);
	eof ->
	    ok
    end.

%% Dumps DECISION.LOG and PDECISION.LOG and removes them.
%% From now on all decisions are logged in the transaction log file
convert_old() ->
    HasOldStuff = 
	mnesia_lib:exists(mnesia_log:previous_decision_log_file()) or
	mnesia_lib:exists(mnesia_log:decision_log_file()),
    case HasOldStuff of
	true ->
	    mnesia_log:open_decision_log(),
	    dump_decision_log(startup),
	    dump_decision_log(startup),
	    mnesia_log:close_decision_log(),
	    Latest = mnesia_log:decision_log_file(),
	    ok = file:delete(Latest);
	false ->
	    ignore
    end.

dump_decision_log(InitBy) ->
    %% Assumed to be run in transaction log dumper process
    Cont = mnesia_log:prepare_decision_log_dump(),
    perform_dump_decision_log(Cont, InitBy).

perform_dump_decision_log(eof, _InitBy) ->
    confirm_decision_log_dump();
perform_dump_decision_log(Cont, InitBy) when InitBy == startup ->
    case mnesia_log:chunk_decision_log(Cont) of
	{Cont2, Decisions} ->
	    note_log_decisions(Decisions, InitBy),
	    perform_dump_decision_log(Cont2, InitBy);
	eof ->
	    confirm_decision_log_dump()
    end;
perform_dump_decision_log(_Cont, _InitBy) ->
    confirm_decision_log_dump().

confirm_decision_log_dump() ->
    dump_decision_tab(),
    mnesia_log:confirm_decision_log_dump().

dump_decision_tab() ->
    Tab = mnesia_decision,
    All = mnesia_lib:db_match_object(ram_copies,Tab, '_'),
    mnesia_log:save_decision_tab({decision_list, All}).

note_log_decisions([What | Tail], InitBy) ->
    note_log_decision(What, InitBy),
    note_log_decisions(Tail, InitBy);
note_log_decisions([], _InitBy) ->
    ok.

note_log_decision(NewD, InitBy) when NewD#decision.outcome == pre_commit ->
    note_log_decision(NewD#decision{outcome = unclear}, InitBy);

note_log_decision(NewD, _InitBy) when record(NewD, decision) ->
    Tid = NewD#decision.tid,
    sync_trans_tid_serial(Tid),
    note_outcome(NewD);
note_log_decision({trans_tid, serial, _Serial}, startup) ->
    ignore;
note_log_decision({trans_tid, serial, Serial}, _InitBy) ->
    sync_trans_tid_serial(Serial);
note_log_decision({mnesia_up, Node, Date, Time}, _InitBy) ->
    note_up(Node, Date, Time);
note_log_decision({mnesia_down, Node, Date, Time}, _InitBy) ->
    note_down(Node, Date, Time);
note_log_decision({master_nodes, Tab, Nodes}, _InitBy) ->
    note_master_nodes(Tab, Nodes);
note_log_decision(H, _InitBy) when H#log_header.log_kind == decision_log ->
    V = mnesia_log:decision_log_version(),
    if
	H#log_header.log_version == V->
	    ok;
	H#log_header.log_version == "2.0" ->
	    verbose("Accepting an old version format of decision log: ~p~n",
		    [V]),
	    ok;
	true ->
	    fatal("Bad version of decision log: ~p~n", [H])
    end;
note_log_decision(H, _InitBy) when H#log_header.log_kind == decision_tab ->
    V = mnesia_log:decision_tab_version(),
    if
	V == H#log_header.log_version ->
	    ok;
	true ->
	    fatal("Bad version of decision tab: ~p~n", [H])
    end;
note_log_decision({decision_list, ItemList}, InitBy) ->
    note_log_decisions(ItemList, InitBy);
note_log_decision(BadItem, InitBy) ->
    exit({"Bad decision log item", BadItem, InitBy}).

trans_tid_serial() ->
    ?ets_lookup_element(mnesia_decision, serial, 3).

set_trans_tid_serial(Val) ->
    ?ets_insert(mnesia_decision, {trans_tid, serial, Val}).

incr_trans_tid_serial() ->
    ?ets_update_counter(mnesia_decision, serial, 1).

sync_trans_tid_serial(ThatCounter) when integer(ThatCounter) ->
    ThisCounter = trans_tid_serial(),
    if
	ThatCounter > ThisCounter ->
	    set_trans_tid_serial(ThatCounter + 1);
	true ->
	    ignore
    end;
sync_trans_tid_serial(Tid) ->
    sync_trans_tid_serial(Tid#tid.counter).

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
%%% Callback functions from gen_server

%%----------------------------------------------------------------------
%% Func: init/1
%% Returns: {ok, State}          |
%%          {ok, State, Timeout} |
%%          {stop, Reason}
%%----------------------------------------------------------------------
init([Parent]) ->
    process_flag(trap_exit, true),
    mnesia_lib:verbose("~p starting: ~p~n", [?MODULE, self()]),
    set(latest_transient_decision, create_transient_decision()),
    set(previous_transient_decisions, []),
    set(recover_nodes, []),
    State = #state{supervisor = Parent},
    {ok, State}.

create_transient_decision() ->
    ?ets_new_table(mnesia_transient_decision, [{keypos, 2}, set, public]).

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

handle_call(init, From, State) when State#state.initiated == false ->
    Args = [{keypos, 2}, set, public, named_table],
    case mnesia_monitor:use_dir() of
	true ->
	    ?ets_new_table(mnesia_decision, Args),
	    set_trans_tid_serial(0),
	    TabFile = mnesia_log:decision_tab_file(),
	    case mnesia_lib:exists(TabFile) of
		true ->
		    load_decision_tab();
		false ->
		    ignore
	    end,
	    convert_old(),
	    mnesia_dumper:opt_dump_log(scan_decisions);
	false ->
	    ?ets_new_table(mnesia_decision, Args),
	    set_trans_tid_serial(0)
    end,
    handle_early_msgs(State, From);

handle_call(Msg, From, State) when State#state.initiated == false ->
    %% Buffer early messages
    Msgs = State#state.early_msgs,
    {noreply, State#state{early_msgs = [{call, Msg, From} | Msgs]}};

handle_call({disconnect, Node}, _From, State) ->
    mnesia_monitor:disconnect(Node),
    mnesia_lib:del(recover_nodes, Node),
    {reply, ok, State};

handle_call({connect_nodes, Ns}, From, State) ->
    %% Determine which nodes we should try to connect
    AlreadyConnected = val(recover_nodes),
    {_, Nodes} = mnesia_lib:search_delete(node(), Ns),
    Check = Nodes -- AlreadyConnected,
    case mnesia_monitor:negotiate_protocol(Check) of
	busy -> 
	    %% monitor is disconnecting some nodes retry 
	    %% the req (to avoid deadlock).
	    erlang:send_after(2, self(), {connect_nodes,Ns,From}),
	    {noreply, State};
	[] ->
	    %% No good noodes to connect to!
	    %% We can't use reply here because this function can be
	    %% called from handle_info
	    gen_server:reply(From, {[], AlreadyConnected}),
	    {noreply, State};
	GoodNodes ->
	    %% Now we have agreed upon a protocol with some new nodes
	    %% and we may use them when we recover transactions
	    mnesia_lib:add_list(recover_nodes, GoodNodes),
	    cast({announce_all, GoodNodes}),
	    case get_master_nodes(schema) of 
		[] ->
		    Context = starting_partitioned_network,
		    mnesia_monitor:detect_inconcistency(GoodNodes, Context);
		_ -> %% If master_nodes is set ignore old inconsistencies
		    ignore
	    end,
	    gen_server:reply(From, {GoodNodes, AlreadyConnected}),
	    {noreply,State}
    end;

handle_call({what_happened, Default, Tid}, _From, State) ->
    sync_trans_tid_serial(Tid),
    Outcome = outcome(Tid, Default),
    {reply, {ok, Outcome}, State};

handle_call({wait_for_decision, D}, From, State) ->
    Recov = val(recover_nodes),
    AliveRam = (mnesia_lib:intersect(D#decision.ram_nodes, Recov) -- [node()]),
    RemoteDisc = D#decision.disc_nodes -- [node()],
    if
	AliveRam == [], RemoteDisc == [] ->
	    %% No more else to wait for and we may safely abort
	    {reply, {ok, aborted}, State};
	true ->
	    verbose("Transaction ~p is unclear. "
		    "Wait for disc nodes: ~w ram: ~w~n",
		    [D#decision.tid, RemoteDisc, AliveRam]),
	    AliveDisc = mnesia_lib:intersect(RemoteDisc, Recov),
	    Msg = {what_decision, node(), D},
	    abcast(AliveRam, Msg),
	    abcast(AliveDisc, Msg),
	    case val(max_wait_for_decision) of
		infinity ->
		    ignore;
		MaxWait ->
		    ForceMsg =  {force_decision, D#decision.tid},
		    {ok, _} = timer:send_after(MaxWait, ForceMsg)
	    end,
	    State2 = State#state{unclear_pid = From,
				 unclear_decision = D,
				 unclear_waitfor = (RemoteDisc ++ AliveRam)},
	    {noreply, State2}
    end;

handle_call({log_mnesia_up, Node}, _From, State) ->
    do_log_mnesia_up(Node),
    {reply, ok, State};

handle_call({log_mnesia_down, Node}, _From, State) ->
    do_log_mnesia_down(Node),
    {reply, ok, State};

handle_call({log_master_nodes, Tab, Nodes, UseDir, IsRunning}, _From, State) ->
    do_log_master_nodes(Tab, Nodes, UseDir, IsRunning),
    {reply, ok, State};

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

do_log_mnesia_up(Node) ->
    Yoyo = {mnesia_up, Node, Date = date(), Time = time()},
    case mnesia_monitor:use_dir() of
	true ->
	    mnesia_log:append(latest_log, Yoyo),
	    disk_log:sync(latest_log);
	false  ->
	    ignore
    end,
    note_up(Node, Date, Time).

do_log_mnesia_down(Node) ->
    Yoyo = {mnesia_down, Node, Date = date(), Time = time()},
    case mnesia_monitor:use_dir() of
	true ->
	    mnesia_log:append(latest_log, Yoyo),
	    disk_log:sync(latest_log);
	false  ->
	    ignore
    end,
    note_down(Node, Date, Time).

do_log_master_nodes(Tab, Nodes, UseDir, IsRunning) ->
    Master = {master_nodes, Tab, Nodes},
    Res = 
	case UseDir of
	    true ->
		LogRes = mnesia_log:append(latest_log, Master),
		disk_log:sync(latest_log),
		LogRes;
	    false  ->
		ok
	end,
    case IsRunning of
	yes ->
	    note_master_nodes(Tab, Nodes);
	_NotRunning ->
	    ignore
    end,
    Res.

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

handle_cast(Msg, State) when State#state.initiated == false ->
    %% Buffer early messages
    Msgs = State#state.early_msgs,
    {noreply, State#state{early_msgs = [{cast, Msg} | Msgs]}};

handle_cast({im_certain, Node, NewD}, State) ->
    OldD = decision(NewD#decision.tid),
    MergedD = merge_decisions(Node, OldD, NewD),    
    do_log_decision(MergedD, false, undefined),
    {noreply, State};

handle_cast({log_decision, D}, State) ->
    do_log_decision(D),
    {noreply, State};

handle_cast(allow_garb, State) ->
    do_allow_garb(),
    {noreply, State};

handle_cast({decisions, Node, Decisions}, State) ->
    mnesia_lib:add(recover_nodes, Node),
    State2 = add_remote_decisions(Node, Decisions, State),
    {noreply, State2};

handle_cast({what_decision, Node, OtherD}, State) ->
    Tid = OtherD#decision.tid,
    sync_trans_tid_serial(Tid),
    Decision = 
	case decision(Tid) of
	    no_decision -> OtherD;
	    MyD when record(MyD, decision) -> MyD
	end,
    announce([Node], [Decision], [], true),
    {noreply, State};

handle_cast({mnesia_down, Node}, State) ->
    case State#state.unclear_decision of
	undefined ->
	    {noreply, State};
	D ->
	    case lists:member(Node, D#decision.ram_nodes) of