mnesia_tm.erl

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

    end.

execute_outer(Mod, Fun, Args, Factor, Retries, Type) ->
    case req(start_outer) of
	{error, Reason} -> 
	    {aborted, Reason};
	{new_tid, Tid, Store} ->
	    Ts = #tidstore{store = Store},
	    NewTidTs = {Mod, Tid, Ts},
	    put(mnesia_activity_state, NewTidTs),
	    execute_transaction(Fun, Args, Factor, Retries, Type)
    end.

execute_inner(Mod, Tid, OldMod, Ts, Fun, Args, Factor, Retries, Type) ->
    case req({add_store, Tid}) of
	{error, Reason} ->
	    {aborted, Reason};
	{new_store, Ets} ->
	    copy_ets(Ts#tidstore.store, Ets),
	    Up = [{OldMod,Ts#tidstore.store} | Ts#tidstore.up_stores],
	    NewTs = Ts#tidstore{level = 1 + Ts#tidstore.level,
				store = Ets,
				up_stores = Up},
	    NewTidTs = {Mod, Tid, NewTs},
	    put(mnesia_activity_state, NewTidTs),
	    execute_transaction(Fun, Args, Factor, Retries, Type)
    end.

copy_ets(From, To) ->
    do_copy_ets(?ets_first(From), From, To).
do_copy_ets('$end_of_table', _,_) -> 
    ok;
do_copy_ets(K, From, To) ->
    Objs = ?ets_lookup(From, K),
    insert_objs(Objs, To),
    do_copy_ets(?ets_next(From, K), From, To).

insert_objs([H|T], Tab) ->
    ?ets_insert(Tab, H),
    insert_objs(T, Tab);
insert_objs([], _Tab) ->
    ok.

execute_transaction(Fun, Args, Factor, Retries, Type) ->
    case catch apply_fun(Fun, Args, Type) of
	{'EXIT', Reason} ->
	    check_exit(Fun, Args, Factor, Retries, Reason, Type);
	{atomic, Value} ->
	    mnesia_lib:incr_counter(trans_commits),
	    erase(mnesia_activity_state),
	    %% no need to clear locks, already done by commit ...
	    %% Flush any un processed mnesia_down messages we might have 
	    flush_downs(),
	    catch unlink(whereis(?MODULE)),
	    {atomic, Value};
	{nested_atomic, Value} ->
	    mnesia_lib:incr_counter(trans_commits),
	    {atomic, Value};
	Value -> %% User called throw
	    Reason = {aborted, {throw, Value}},
	    return_abort(Fun, Args, Reason)
    end.

apply_fun(Fun, Args, Type) ->
    Result = apply(Fun, Args),
    case t_commit(Type) of
	do_commit ->
            {atomic, Result};
        do_commit_nested ->
            {nested_atomic, Result};
        {do_abort, {aborted, Reason}} ->
            {'EXIT', {aborted, Reason}};
        {do_abort, Reason} ->
            {'EXIT', {aborted, Reason}}
    end.

check_exit(Fun, Args, Factor, Retries, Reason, Type) ->
    case Reason of
	{aborted, C} when record(C, cyclic) ->
	    maybe_restart(Fun, Args, Factor, Retries, Type, C);
	{aborted, {node_not_running, N}} ->
	    maybe_restart(Fun, Args, Factor, Retries, Type, {node_not_running, N});
	{aborted, {bad_commit, N}} ->
	    maybe_restart(Fun, Args, Factor, Retries, Type, {bad_commit, N});
	_ -> 
	    return_abort(Fun, Args, Reason)
    end.

maybe_restart(Fun, Args, Factor, Retries, Type, Why) ->
    {Mod, Tid, Ts} = get(mnesia_activity_state),
    case try_again(Retries) of
	yes when Ts#tidstore.level == 1 ->
	    restart(Mod, Tid, Ts, Fun, Args, Factor, Retries, Type, Why);
	yes ->
	    return_abort(Fun, Args, Why);
	no ->
	    return_abort(Fun, Args, {aborted, nomore})
    end.

try_again(infinity) -> yes;
try_again(X) when number(X) , X > 1 -> yes;
try_again(_) -> no.

%% We can only restart toplevel transactions.
%% If a deadlock situation occurs in a nested transaction
%% The whole thing including all nested transactions need to be
%% restarted. The stack is thus popped by a consequtive series of
%% exit({aborted, #cyclic{}}) calls

restart(Mod, Tid, Ts, Fun, Args, Factor0, Retries0, Type, Why) ->
    mnesia_lib:incr_counter(trans_restarts),
    Retries = decr(Retries0),
    case Why of
	{bad_commit, _N} ->
	    return_abort(Fun, Args, Why),
	    Factor = 1,
	    SleepTime = mnesia_lib:random_time(Factor, Tid#tid.counter),
	    dbg_out("Restarting transaction ~w: in ~wms ~w~n", [Tid, SleepTime, Why]),
	    timer:sleep(SleepTime),
	    execute_outer(Mod, Fun, Args, Factor, Retries, Type);
	{node_not_running, _N} ->   %% Avoids hanging in receive_release_tid_ack
	    return_abort(Fun, Args, Why),
	    Factor = 1,
	    SleepTime = mnesia_lib:random_time(Factor, Tid#tid.counter),
	    dbg_out("Restarting transaction ~w: in ~wms ~w~n", [Tid, SleepTime, Why]),
	    timer:sleep(SleepTime),
	    execute_outer(Mod, Fun, Args, Factor, Retries, Type);	
	_ ->
	    SleepTime = mnesia_lib:random_time(Factor0, Tid#tid.counter),
	    dbg_out("Restarting transaction ~w: in ~wms ~w~n", [Tid, SleepTime, Why]),
	    
	    if
		Factor0 /= 10 ->
		    ignore;
		true ->
		    %% Our serial may be much larger than other nodes ditto
		    AllNodes = val({current, db_nodes}),
		    verbose("Sync serial ~p~n", [Tid]),
		    rpc:abcast(AllNodes, ?MODULE, {sync_trans_serial, Tid})
	    end,
	    intercept_friends(Tid, Ts),
	    Store = Ts#tidstore.store,
	    Nodes = get_elements(nodes,Store),
	    ?MODULE ! {self(), {restart, Tid, Store}},
	    mnesia_locker:send_release_tid(Nodes, Tid),
	    timer:sleep(SleepTime),
	    mnesia_locker:receive_release_tid_acc(Nodes, Tid),
	    case get_restarted(Tid) of
		{restarted, Tid} ->
		    execute_transaction(Fun, Args, Factor0 + 1, 
					Retries, Type);
		{error, Reason} ->
		    mnesia:abort(Reason)
	    end
    end.

get_restarted(Tid) ->
    case Res = rec() of
	{restarted, Tid} ->
	    Res;
	{error,_} ->
	    Res;
	_ -> %% We could get a couple of aborts to many.
	    get_restarted(Tid)
    end.

decr(infinity) -> infinity;
decr(X) when integer(X), X > 1 -> X - 1;
decr(_X) -> 0.

return_abort(Fun, Args, Reason)  ->
    {_Mod, Tid, Ts} = get(mnesia_activity_state),
    dbg_out("Transaction ~p calling ~p with ~p failed: ~n ~p~n", 
	    [Tid, Fun, Args, Reason]),
    OldStore = Ts#tidstore.store,
    Nodes = get_elements(nodes, OldStore),
    intercept_friends(Tid, Ts),
    catch mnesia_lib:incr_counter(trans_failures),
    Level = Ts#tidstore.level,
    if
	Level == 1 ->
	    mnesia_locker:async_release_tid(Nodes, Tid),
	    ?MODULE ! {delete_transaction, Tid},
	    erase(mnesia_activity_state),	    
	    flush_downs(),
	    catch unlink(whereis(?MODULE)),
	    {aborted, mnesia_lib:fix_error(Reason)};
	true ->
	    %% Nested transaction
	    [{OldMod,NewStore} | Tail] = Ts#tidstore.up_stores,
	    req({del_store, Tid, NewStore, OldStore, true}),
	    Ts2 = Ts#tidstore{store = NewStore,
			      up_stores = Tail,
			      level = Level - 1},
	    NewTidTs = {OldMod, Tid, Ts2},
	    put(mnesia_activity_state, NewTidTs),
	    case Reason of 
		#cyclic{} ->
		    exit({aborted, Reason});
		{node_not_running, _N} -> 
		    exit({aborted, Reason});
		{bad_commit, _N}-> 
		    exit({aborted, Reason});
		_ -> 
		    {aborted, mnesia_lib:fix_error(Reason)}
	    end
    end.

flush_downs() ->
    receive
	{?MODULE, _, _} -> flush_downs(); % Votes
	{mnesia_down, _} -> flush_downs()
    after 0 -> flushed
    end.


put_activity_id(MTT) ->
    put_activity_id(MTT, undefined).
put_activity_id(undefined,_) ->
    erase_activity_id();
put_activity_id({Mod, Tid, Ts},Fun) 
  when record(Tid, tid), record(Ts, tidstore) ->
    flush_downs(),
    Store = Ts#tidstore.store,
    if 
	is_function(Fun) ->
	    ?ets_insert(Store, {friends, {stop,Fun}});
	true ->
	    ?ets_insert(Store, {friends, self()})
    end,
    NewTidTs = {Mod, Tid, Ts},
    put(mnesia_activity_state, NewTidTs);
put_activity_id(SimpleState,_) ->
    put(mnesia_activity_state, SimpleState).

erase_activity_id() ->
    flush_downs(),
    erase(mnesia_activity_state).

get_elements(Type,Store) ->    
    case catch ?ets_lookup(Store, Type) of
	[] -> [];
	[{_,Val}] -> [Val];
	{'EXIT', _} -> [];
	Vals -> [Val|| {_,Val} <- Vals]
    end.
    
opt_propagate_store(_Current, _Obsolete, false) ->
    ok;
opt_propagate_store(Current, Obsolete, true) ->
    propagate_store(Current, nodes, get_elements(nodes,Obsolete)),
    propagate_store(Current, fixtable, get_elements(fixtable,Obsolete)),
    propagate_store(Current, friends, get_elements(friends, Obsolete)).

propagate_store(Store, Var, [Val | Vals]) ->
    ?ets_insert(Store, {Var, Val}),
    propagate_store(Store, Var, Vals);
propagate_store(_Store, _Var, []) ->
    ok.

%% Tell all processes that are cooperating with the current transaction
intercept_friends(_Tid, Ts) ->
    Friends = get_elements(friends,Ts#tidstore.store),
    intercept_best_friend(Friends, false).

intercept_best_friend([],_) ->    ok;
intercept_best_friend([{stop,Fun} | R],Ignore) ->
    catch Fun(),
    intercept_best_friend(R,Ignore);
intercept_best_friend([Pid | R],false) ->    
    Pid ! {activity_ended, undefined, self()}, 
    wait_for_best_friend(Pid, 0),
    intercept_best_friend(R,true);
intercept_best_friend([_|R],true) ->
    intercept_best_friend(R,true).

wait_for_best_friend(Pid, Timeout) ->
    receive
	{'EXIT', Pid, _} -> ok;
	{activity_ended, _, Pid} -> ok
    after Timeout ->
	    case my_process_is_alive(Pid) of
		true -> wait_for_best_friend(Pid, 1000);
		false -> ok
	    end
    end.
    
my_process_is_alive(Pid) ->
    case catch erlang:is_process_alive(Pid) of % New BIF in R5
	true -> 
	    true;
	false -> 
	    false;
	{'EXIT', _} -> % Pre R5 backward compatibility 
	    case process_info(Pid, message_queue_len) of
		undefined -> false;
		_ -> true
	    end 
    end.

dirty(Protocol, Item) ->
    {{Tab, Key}, _Val, _Op} = Item,
    Tid = {dirty, self()},
    Prep = prepare_items(Tid, Tab, Key, [Item], #prep{protocol= Protocol}),
    CR =  Prep#prep.records,
    case Protocol of
	async_dirty ->
	    %% Send commit records to the other involved nodes,
	    %% but do only wait for one node to complete.
	    %% Preferrably, the local node if possible. 
		    
	    ReadNode = val({Tab, where_to_read}),
	    {WaitFor, FirstRes} = async_send_dirty(Tid, CR, Tab, ReadNode),
	    rec_dirty(WaitFor, FirstRes);
		
	sync_dirty ->
	    %% Send commit records to the other involved nodes,
	    %% and wait for all nodes to complete
	    {WaitFor, FirstRes} = sync_send_dirty(Tid, CR, Tab, []),
	    rec_dirty(WaitFor, FirstRes);
	_ ->
	    mnesia:abort({bad_activity, Protocol})
    end.

%% This is the commit function, The first thing it does,
%% is to find out which nodes that have been participating
%% in this particular transaction, all of the mnesia_locker:lock*
%% functions insert the names of the nodes where it aquires locks
%% into the local shadow Store
%% This function exacutes in the context of the user process
t_commit(Type) ->
    {_Mod, Tid, Ts} = get(mnesia_activity_state),
    Store = Ts#tidstore.store,
    if
	Ts#tidstore.level == 1 ->
	    intercept_friends(Tid, Ts),
	    %% N is number of updates 
	    case arrange(Tid, Store, Type) of
		{N, Prep} when N > 0 ->
		    multi_commit(Prep#prep.protocol,
				 Tid, Prep#prep.records, Store);
		{0, Prep} ->
		    multi_commit(read_only, Tid, Prep#prep.records, Store)
	    end;
	true ->
	    %% nested commit
	    Level = Ts#tidstore.level,
	    [{OldMod,Obsolete} | Tail] = Ts#tidstore.up_stores,
	    req({del_store, Tid, Store, Obsolete, false}),
	    NewTs = Ts#tidstore{store = Store,
				up_stores = Tail,
				level = Level - 1},
	    NewTidTs = {OldMod, Tid, NewTs},
	    put(mnesia_activity_state, NewTidTs),
	    do_commit_nested
    end.

%% This function arranges for all objects we shall write in S to be
%% in a list of {Node, CommitRecord}
%% Important function for the performance of mnesia.

arrange(Tid, Store, Type) ->
    %% The local node is always included
    Nodes = get_elements(nodes,Store),
    Recs = prep_recs(Nodes, []),
    Key = ?ets_first(Store),
    N = 0,
    Prep = 
	case Type of 
	    async -> #prep{protocol = sym_trans, records = Recs};
	    sync -> #prep{protocol = sync_sym_trans, records = Recs}
	end,
    case catch do_arrange(Tid, Store, Key, Prep, N) of
	{'EXIT', Reason} ->
	    dbg_out("do_arrange failed ~p ~p~n", [Reason, Tid]),
	    case Reason of
		{aborted, R} ->
		    mnesia:abort(R);
		_ -> 
		    mnesia:abort(Reason)