mnesia_recover.erl

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

%% ``The contents of this file are subject to the Erlang Public License,
%% Version 1.1, (the "License"); you may not use this file except in
%% compliance with the License. You should have received a copy of the
%% Erlang Public License along with this software. If not, it can be
%% retrieved via the world wide web at http://www.erlang.org/.
%% 
%% Software distributed under the License is distributed on an "AS IS"
%% basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See
%% the License for the specific language governing rights and limitations
%% under the License.
%% 
%% The Initial Developer of the Original Code is Ericsson Utvecklings AB.
%% Portions created by Ericsson are Copyright 1999, Ericsson Utvecklings
%% AB. All Rights Reserved.''
%% 
%%     $Id$
%%
-module(mnesia_recover).

-behaviour(gen_server).

-export([
	 allow_garb/0,
	 call/1,
	 connect_nodes/1,
	 disconnect/1,
	 dump_decision_tab/0,
	 get_master_node_info/0,
	 get_master_node_tables/0,
	 get_master_nodes/1,
	 get_mnesia_downs/0,
	 has_mnesia_down/1,
	 incr_trans_tid_serial/0,
	 init/0,
	 log_decision/1,
	 log_master_nodes/3,
	 log_mnesia_down/1,
	 log_mnesia_up/1,
	 mnesia_down/1,
	 note_decision/2,
	 note_log_decision/2,
	 outcome/2,
	 start/0,
	 start_garb/0,
	 still_pending/1,
	 sync_trans_tid_serial/1,
	 wait_for_decision/2,
	 what_happened/3
	]).

%% gen_server callbacks
-export([init/1,
	 handle_call/3,
	 handle_cast/2,
	 handle_info/2,
	 terminate/2,
	 code_change/3
	]).


-include("mnesia.hrl").
-import(mnesia_lib, [set/2, verbose/2, error/2, fatal/2]).

-record(state, {supervisor,
		unclear_pid,
		unclear_decision,
		unclear_waitfor,
		tm_queue_len = 0,
		initiated = false,
		early_msgs = []
	       }).

%%-define(DBG(F, A), mnesia:report_event(list_to_atom(lists:flatten(io_lib:format(F, A))))).
%%-define(DBG(F, A), io:format("DBG: " ++ F, A)).

-record(transient_decision, {tid, outcome}).

start() ->
    gen_server:start_link({local, ?MODULE}, ?MODULE, [self()],
			  [{timeout, infinity}
			   %%, {debug, [trace]}
			  ]).

init() ->
    call(init).

start_garb() ->
    Pid = whereis(mnesia_recover),
    {ok, _} = timer:send_interval(timer:minutes(2), Pid, garb_decisions),
    {ok, _} = timer:send_interval(timer:seconds(10), Pid, check_overload).

allow_garb() ->
    cast(allow_garb).


%% The transaction log has either been swiched (latest -> previous) or
%% there is nothing to be dumped. This means that the previous
%% transaction log only may contain commit records which refers to
%% transactions noted in the last two of the 'Prev' tables. All other
%% tables may now be garbed by 'garb_decisions' (after 2 minutes).
%% Max 10 tables are kept. 
do_allow_garb() ->
    %% The order of the following stuff is important!
    Curr = val(latest_transient_decision),
    Old = val(previous_transient_decisions),
    Next = create_transient_decision(),
    {Prev, ReallyOld} = sublist([Curr | Old], 10, []),
    [?ets_delete_table(Tab) || Tab <- ReallyOld],
    set(previous_transient_decisions, Prev),
    set(latest_transient_decision, Next).
    
sublist([H|R], N, Acc) when N > 0 ->
    sublist(R, N-1, [H| Acc]);
sublist(List, _N, Acc) ->
    {lists:reverse(Acc), List}.

do_garb_decisions() ->
    case val(previous_transient_decisions) of
	[First, Second | Rest] ->
	    set(previous_transient_decisions, [First, Second]),
	    [?ets_delete_table(Tab) || Tab <- Rest];
	_ ->
	    ignore
    end.

connect_nodes(Ns) ->
    call({connect_nodes, Ns}).

disconnect(Node) ->
    call({disconnect, Node}).

log_decision(D) ->
    cast({log_decision, D}).

val(Var) ->
    case ?catch_val(Var) of
	{'EXIT', Reason} -> mnesia_lib:other_val(Var, Reason); 
	Value -> Value
    end.

call(Msg) ->
    Pid = whereis(?MODULE),
    case Pid of
	undefined ->
	    {error, {node_not_running, node()}};
	Pid ->
	    link(Pid),
	    Res = gen_server:call(Pid, Msg, infinity),
	    unlink(Pid),

            %% We get an exit signal if server dies
            receive
                {'EXIT', Pid, _Reason} ->
                    {error, {node_not_running, node()}}
            after 0 ->
                    ignore
            end,
	    Res
    end.

multicall(Nodes, Msg) ->
    rpc:multicall(Nodes, ?MODULE, call, [Msg]).

cast(Msg) ->
    case whereis(?MODULE) of
	undefined -> ignore;
	Pid ->  gen_server:cast(Pid, Msg)
    end.

abcast(Nodes, Msg) ->
    gen_server:abcast(Nodes, ?MODULE, Msg).

note_decision(Tid, Outcome) ->
    Tab = val(latest_transient_decision),
    ?ets_insert(Tab, #transient_decision{tid = Tid, outcome = Outcome}).

note_up(Node, _Date, _Time) ->
    ?ets_delete(mnesia_decision, Node).
    
note_down(Node, Date, Time) ->
    ?ets_insert(mnesia_decision, {mnesia_down, Node, Date, Time}).
    
note_master_nodes(Tab, []) ->
    ?ets_delete(mnesia_decision, Tab);
note_master_nodes(Tab, Nodes) when list(Nodes) ->
    Master = {master_nodes, Tab, Nodes},
    ?ets_insert(mnesia_decision, Master).

note_outcome(D) when D#decision.disc_nodes == [] ->
%%    ?DBG("~w: note_tmp_decision: ~w~n", [node(), D]),
    note_decision(D#decision.tid, filter_outcome(D#decision.outcome)),
    ?ets_delete(mnesia_decision, D#decision.tid);
note_outcome(D) when D#decision.disc_nodes /= [] ->
%%    ?DBG("~w: note_decision: ~w~n", [node(), D]),
    ?ets_insert(mnesia_decision, D).

do_log_decision(D) when D#decision.outcome /= unclear ->
    OldD = decision(D#decision.tid),
    MergedD = merge_decisions(node(), OldD, D),
    do_log_decision(MergedD, true, D);
do_log_decision(D) ->
    do_log_decision(D, false, undefined).

do_log_decision(D, DoTell, NodeD) ->
    DiscNs = D#decision.disc_nodes -- [node()],
    Outcome = D#decision.outcome,
    D2 =
	case Outcome of
	    aborted -> D#decision{disc_nodes = DiscNs};
	    committed -> D#decision{disc_nodes = DiscNs};
	    _ -> D
	end,
    note_outcome(D2),
    case mnesia_monitor:use_dir() of
	true ->
	    mnesia_log:append(latest_log, D2),
	    if
		DoTell == true, Outcome /= unclear ->
		    tell_im_certain(NodeD#decision.disc_nodes--[node()],D2),
		    tell_im_certain(NodeD#decision.ram_nodes--[node()], D2);
		true ->
		    ignore
	    end;
	false ->
	    ignore
    end.

tell_im_certain([], _D) ->
    ignore;
tell_im_certain(Nodes, D) ->
    Msg = {im_certain, node(), D},
  %%  mnesia_lib:verbose("~w: tell: ~w~n", [Msg, Nodes]), 
    abcast(Nodes, Msg).

log_mnesia_up(Node) ->
    call({log_mnesia_up, Node}).

log_mnesia_down(Node) ->
    call({log_mnesia_down, Node}).

get_mnesia_downs() ->
    Tab = mnesia_decision,
    Pat = {mnesia_down, '_', '_', '_'},
    Downs = ?ets_match_object(Tab, Pat),
    [Node || {mnesia_down, Node, _Date, _Time} <- Downs].

%% Check if we have got a mnesia_down from Node
has_mnesia_down(Node) ->
    case ?ets_lookup(mnesia_decision, Node) of
	[{mnesia_down, Node, _Date, _Time}] ->
	    true;
	[] ->
	    false
    end.
    
mnesia_down(Node) ->
    case ?catch_val(recover_nodes) of
	{'EXIT', _} ->
	    %% Not started yet
	    ignore;
	_ ->
	    mnesia_lib:del(recover_nodes, Node),
	    cast({mnesia_down, Node})
    end.

log_master_nodes(Args, UseDir, IsRunning) ->
    if
	IsRunning == yes ->
	    log_master_nodes2(Args, UseDir, IsRunning, ok);
	UseDir == false ->
	    ok;
	true ->
	    Name = latest_log,
	    Fname = mnesia_log:latest_log_file(),
	    Exists = mnesia_lib:exists(Fname),
	    Repair = mnesia:system_info(auto_repair),
	    OpenArgs = [{file, Fname}, {name, Name}, {repair, Repair}],
	    case disk_log:open(OpenArgs) of
		{ok, Name} ->
		    log_master_nodes2(Args, UseDir, IsRunning, ok);
		{repaired, Name, {recovered,  _R}, {badbytes, _B}}
		  when Exists == true ->
		    log_master_nodes2(Args, UseDir, IsRunning, ok);
		{repaired, Name, {recovered,  _R}, {badbytes, _B}}
		  when Exists == false ->
		    mnesia_log:write_trans_log_header(),
		    log_master_nodes2(Args, UseDir, IsRunning, ok);
		{error, Reason} ->
		    {error, Reason}
	    end
    end.

log_master_nodes2([{Tab, Nodes} | Tail], UseDir, IsRunning, WorstRes) ->
    Res = 
	case IsRunning of
	    yes ->
		R = call({log_master_nodes, Tab, Nodes, UseDir, IsRunning}),
		mnesia_controller:master_nodes_updated(Tab, Nodes),
		R;
	    _ ->
		do_log_master_nodes(Tab, Nodes, UseDir, IsRunning)
	end,
    case Res of
	ok ->
	    log_master_nodes2(Tail, UseDir, IsRunning, WorstRes);
	{error, Reason} ->
	    log_master_nodes2(Tail, UseDir, IsRunning, {error, Reason})
    end;
log_master_nodes2([], _UseDir, IsRunning, WorstRes) ->
    case IsRunning of
	yes ->
	    WorstRes;
	_ ->
	    disk_log:close(latest_log),
	    WorstRes
    end.

get_master_node_info() ->
    Tab = mnesia_decision,
    Pat = {master_nodes, '_', '_'},
    case catch mnesia_lib:db_match_object(ram_copies,Tab, Pat) of
	{'EXIT', _} ->
	    [];
	Masters ->
	    Masters
    end.

get_master_node_tables() ->
    Masters = get_master_node_info(),
    [Tab || {master_nodes, Tab, _Nodes} <- Masters].

get_master_nodes(Tab) ->
    case catch ?ets_lookup_element(mnesia_decision, Tab, 3) of
	{'EXIT', _} -> [];
	Nodes  -> Nodes
    end.

%% Determine what has happened to the transaction
what_happened(Tid, Protocol, Nodes) ->
    Default = 
	case Protocol of
	    asym_trans -> aborted;
	    _ -> unclear  %% sym_trans and sync_sym_trans
	end,
    This = node(),
    case lists:member(This, Nodes) of
	true ->
	    {ok, Outcome} = call({what_happened, Default, Tid}),
	    Others = Nodes -- [This],
	    case filter_outcome(Outcome) of
		unclear -> what_happened_remotely(Tid, Default, Others);
		aborted -> aborted;
		committed -> committed
	    end;
	false ->
	    what_happened_remotely(Tid, Default, Nodes)
    end.

what_happened_remotely(Tid, Default, Nodes) ->
    {Replies, _} = multicall(Nodes, {what_happened, Default, Tid}),
    check_what_happened(Replies, 0, 0).

check_what_happened([H | T], Aborts, Commits) ->
    case H of
	{ok, R} ->
	    case filter_outcome(R) of
		committed ->
		    check_what_happened(T, Aborts, Commits + 1);
		aborted ->
		    check_what_happened(T, Aborts + 1, Commits);
		unclear ->
		    check_what_happened(T, Aborts, Commits)
	    end;
	{error, _} ->
	    check_what_happened(T, Aborts, Commits);
	{badrpc, _} ->
	    check_what_happened(T, Aborts, Commits)
    end;
check_what_happened([], Aborts, Commits) ->
    if
	Aborts == 0, Commits == 0 -> aborted;  % None of the active nodes knows
	Aborts > 0 -> aborted;                 % Someody has aborted
	Aborts == 0, Commits > 0 -> committed  % All has committed
    end.

%% Determine what has happened to the transaction
%% and possibly wait forever for the decision.
wait_for_decision(presume_commit, _InitBy) ->
    %% sym_trans
    {{presume_commit, self()}, committed};