The R13B03 release.OTP_R13B03

1 files changed, 1196 insertions, 0 deletions

diff --git a/lib/mnesia/src/mnesia_recover.erl b/lib/mnesia/src/mnesia_recover.erl
new file mode 100644
index 0000000000..6c53c2e752
--- /dev/null
+++ b/lib/mnesia/src/mnesia_recover.erl
@@ -0,0 +1,1196 @@
+%%
+%% %CopyrightBegin%
+%% 
+%% Copyright Ericsson AB 1997-2009. All Rights Reserved.
+%% 
+%% 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 online 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.
+%% 
+%% %CopyrightEnd%
+%%
+
+%%
+-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,
+	 sync/0,
+	 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),
+    %% Don't garb small tables, they are created on every 
+    %% dump_log and may be small (empty) for schema transactions
+    %% which are dumped twice
+    case ets:info(Curr, size) > 20 of
+	true ->
+	    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);
+	false ->
+	    ignore
+    end.
+    
+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 ->
+                    Res
+            end
+    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 is_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).
+
+sync() ->
+    call(sync).
+
+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};
+
+wait_for_decision(D, InitBy) when D#decision.outcome == presume_abort ->
+    wait_for_decision(D, InitBy, 0).
+    
+wait_for_decision(D, InitBy, N) -> 
+    %% asym_trans
+    Tid = D#decision.tid,
+    Max = 10,
+    Outcome = outcome(Tid, D#decision.outcome),
+    if 
+	Outcome =:= committed -> {Tid, committed};
+	Outcome =:= aborted   -> {Tid, aborted};
+	Outcome =:= presume_abort -> 
+	    case N > Max of
+		true -> {Tid, aborted};
+		false -> % busy loop for ets decision moving
+		    timer:sleep(10),
+		    wait_for_decision(D, InitBy, N+1)
+	    end;
+	InitBy /= startup ->
+	    %% Wait a while for active transactions
+	    %% to end and try again
+	    timer:sleep(100), 
+	    wait_for_decision(D, InitBy, N);
+	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 is_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 is_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(sync, _From, State) ->
+    {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 is_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
+		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 is_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 is_record(D, decision) ->
+	    D;
+	[C] when is_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;
+	is_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 is_record(D, decision) ->
+    State2 = add_remote_decision(Node, D, State),
+    add_remote_decisions(Node, Tail, State2);
+
+add_remote_decisions(Node, [C | Tail], State)
+        when is_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 is_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]}].
+