diff options
Diffstat (limited to 'lib/mnesia')
| -rw-r--r-- | lib/mnesia/src/mnesia_checkpoint.erl | 279 | ||||
| -rw-r--r-- | lib/mnesia/test/mnesia_consistency_test.erl | 28 |
2 files changed, 140 insertions, 167 deletions
diff --git a/lib/mnesia/src/mnesia_checkpoint.erl b/lib/mnesia/src/mnesia_checkpoint.erl index eb8fe38908..173e3be2f5 100644 --- a/lib/mnesia/src/mnesia_checkpoint.erl +++ b/lib/mnesia/src/mnesia_checkpoint.erl @@ -1,7 +1,7 @@ %% %% %CopyrightBegin% %% -%% Copyright Ericsson AB 1996-2009. All Rights Reserved. +%% Copyright Ericsson AB 1996-2013 %% %% 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 @@ -31,8 +31,7 @@ tm_retain/5, tm_enter_pending/1, tm_enter_pending/3, - tm_exit_pending/1, - convert_cp_record/1 + tm_exit_pending/1 ]). %% Public interface @@ -88,25 +87,6 @@ pid }). -%% Old record definition --record(checkpoint, {name, - allow_remote, - ram_overrides_dump, - nodes, - node, - now, - min, - max, - pending_tab, - wait_for_old, - is_activated, - ignore_new, - retainers, - iterators, - supervisor, - pid - }). - -record(retainer, {cp_name, tab_name, store, writers = [], really_retain = true}). -record(iter, {tab_name, oid_tab, main_tab, retainer_tab, source, val, pid}). @@ -129,15 +109,6 @@ tm_prepare(Cp) when is_record(Cp, checkpoint_args) -> start_retainer(Cp); true -> {error, {already_exists, Name, node()}} - end; -tm_prepare(Cp) when is_record(Cp, checkpoint) -> - %% Node with old protocol sent an old checkpoint record - %% and we have to convert it - case convert_cp_record(Cp) of - {ok, NewCp} -> - tm_prepare(NewCp); - {error, Reason} -> - {error, Reason} end. tm_mnesia_down(Node) -> @@ -156,6 +127,7 @@ tm_enter_pending(Pending) -> tm_enter_pending([], Pending) -> Pending; tm_enter_pending([Tab | Tabs], Pending) -> + %% io:format("Add ~p ~p ~p~n",[Tab, Pending, hd(tl(element(2, process_info(self(), current_stacktrace))))]), catch ?ets_insert(Tab, Pending), tm_enter_pending(Tabs, Pending). @@ -371,9 +343,7 @@ activate(Args) -> end. args2cp(Args) when is_list(Args)-> - case catch lists:foldl(fun check_arg/2, #checkpoint_args{}, Args) of - {'EXIT', Reason} -> - {error, Reason}; + try lists:foldl(fun check_arg/2, #checkpoint_args{}, Args) of Cp -> case check_tables(Cp) of {error, Reason} -> @@ -381,6 +351,10 @@ args2cp(Args) when is_list(Args)-> {ok, Overriders, AllTabs} -> arrange_retainers(Cp, Overriders, AllTabs) end + catch exit:Reason -> + {error, Reason}; + error:Reason -> + {error, Reason} end; args2cp(Args) -> {error, {badarg, Args}}. @@ -390,10 +364,10 @@ check_arg({name, Name}, Cp) -> true -> exit({already_exists, Name}); false -> - case catch tab2retainer({foo, Name}) of - List when is_list(List) -> - Cp#checkpoint_args{name = Name}; - _ -> + try + [_|_] = tab2retainer({foo, Name}), + Cp#checkpoint_args{name = Name} + catch _:_ -> exit({badarg, Name}) end end; @@ -641,11 +615,7 @@ init(Cp) -> process_flag(priority, high), %% Needed dets files might starve the system Name = Cp#checkpoint_args.name, Props = [set, public, {keypos, 2}], - case catch ?ets_new_table(mnesia_pending_checkpoint, Props) of - {'EXIT', Reason} -> %% system limit - Msg = "Cannot create an ets table for pending transactions", - Error = {error, {system_limit, Name, Msg, Reason}}, - proc_lib:init_ack(Cp#checkpoint_args.supervisor, Error); + try ?ets_new_table(mnesia_pending_checkpoint, Props) of PendingTab -> Rs = [prepare_tab(Cp, R) || R <- Cp#checkpoint_args.retainers], Cp2 = Cp#checkpoint_args{retainers = Rs, @@ -658,6 +628,10 @@ init(Cp) -> dbg_out("Checkpoint ~p (~p) started~n", [Name, self()]), proc_lib:init_ack(Cp2#checkpoint_args.supervisor, {ok, self()}), retainer_loop(Cp2) + catch error:Reason -> %% system limit + Msg = "Cannot create an ets table for pending transactions", + Error = {error, {system_limit, Name, Msg, Reason}}, + proc_lib:init_ack(Cp#checkpoint_args.supervisor, Error) end. prepare_tab(Cp, R) -> @@ -798,104 +772,142 @@ retainer_delete({dets, Store}) -> Fname = tab2retainer(Store), file:delete(Fname). -retainer_loop(Cp) -> - Name = Cp#checkpoint_args.name, +retainer_loop(Cp = #checkpoint_args{is_activated=false, name=Name}) -> receive - {_From, {retain, Tid, Tab, Key, OldRecs}} - when Cp#checkpoint_args.wait_for_old == [] -> - R = val({Tab, {retainer, Name}}), - PendingTab = Cp#checkpoint_args.pending_tab, - case R#retainer.really_retain of - true when PendingTab =:= undefined -> - Store = R#retainer.store, - case retainer_get(Store, Key) of - [] -> retainer_put(Store, {Tab, Key, OldRecs}); - _ -> already_retained - end; - true -> - case ets:member(PendingTab, Tid) of - true -> ignore; - false -> - Store = R#retainer.store, - case retainer_get(Store, Key) of - [] -> retainer_put(Store, {Tab, Key, OldRecs}); - _ -> already_retained - end - end; - false -> - ignore - end, - retainer_loop(Cp); - %% Adm + {From, {activate, Pending}} -> + StillPending = mnesia_recover:still_pending(Pending), + enter_still_pending(StillPending, Cp#checkpoint_args.pending_tab), + Local = [Tid || #tid{pid=Pid} = Tid <- StillPending, node(Pid) =/= node()], + Cp2 = maybe_activate(Cp#checkpoint_args{wait_for_old = Local}), + + reply(From, Name, activated), + retainer_loop(Cp2); + + {_From, {exit_pending, Tid}} when is_list(Cp#checkpoint_args.wait_for_old) -> + StillPending = lists:delete(Tid, Cp#checkpoint_args.wait_for_old), + Cp2 = Cp#checkpoint_args{wait_for_old = StillPending}, + Cp3 = maybe_activate(Cp2), + retainer_loop(Cp3); + {From, deactivate} -> do_stop(Cp), reply(From, Name, deactivated), unlink(From), exit(shutdown); + {From, get_checkpoint} -> + reply(From, Name, Cp), + retainer_loop(Cp); + {_From, {add_retainer, R, Node}} -> + Cp2 = do_add_retainer(Cp, R, Node), + retainer_loop(Cp2); + + {From, collect_pending} -> + PendingTab = Cp#checkpoint_args.pending_tab, + del(pending_checkpoints, PendingTab), + Pending = ?ets_match_object(PendingTab, '_'), + reply(From, Name, {ok, Pending}), + retainer_loop(Cp); + + {_From, {mnesia_down, Node}} -> + Cp2 = do_del_retainers(Cp, Node), + retainer_loop(Cp2); + {'EXIT', Parent, _} when Parent == Cp#checkpoint_args.supervisor -> %% do_stop(Cp), %% assume that entire Mnesia is terminating exit(shutdown); - {_From, {mnesia_down, Node}} -> - Cp2 = do_del_retainers(Cp, Node), - retainer_loop(Cp2); + {'EXIT', From, _Reason} -> + Iters = [Iter || Iter <- Cp#checkpoint_args.iterators, + check_iter(From, Iter)], + retainer_loop(Cp#checkpoint_args{iterators = Iters}); + + {system, From, Msg} -> + dbg_out("~p got {system, ~p, ~p}~n", [?MODULE, From, Msg]), + sys:handle_system_msg(Msg, From, Cp#checkpoint_args.supervisor, + ?MODULE, [], Cp) + end; + +retainer_loop(Cp = #checkpoint_args{name=Name}) -> + receive + {_From, {retain, Tid, Tab, Key, OldRecs}} -> + R = val({Tab, {retainer, Name}}), + PendingTab = Cp#checkpoint_args.pending_tab, + case R#retainer.really_retain of + true -> + Store = R#retainer.store, + try true = ets:member(PendingTab, Tid), + %% io:format("CP: ~p ~p ~p ~p~n",[true, Tab, Key, Tid]), + case retainer_get(Store, Key) of + [] -> ignore; + _ -> ets:delete(element(2,Store), Key) + end + catch _:_ -> + %% io:format("CP: ~p ~p ~p ~p~n",[false, Tab, Key, Tid]), + case retainer_get(Store, Key) of + [] -> retainer_put(Store, {Tab, Key, OldRecs}); + _ -> already_retained + end + end; + false -> + ignore + end, + retainer_loop(Cp); + + %% Adm {From, get_checkpoint} -> reply(From, Name, Cp), retainer_loop(Cp); - {From, {add_copy, Tab, Node}} when Cp#checkpoint_args.wait_for_old == [] -> + {From, {add_copy, Tab, Node}} -> {Res, Cp2} = do_add_copy(Cp, Tab, Node), reply(From, Name, Res), retainer_loop(Cp2); - {From, {del_copy, Tab, Node}} when Cp#checkpoint_args.wait_for_old == [] -> + {From, {del_copy, Tab, Node}} -> Cp2 = do_del_copy(Cp, Tab, Node), reply(From, Name, ok), retainer_loop(Cp2); - {From, {change_copy, Tab, From, To}} when Cp#checkpoint_args.wait_for_old == [] -> + {From, {change_copy, Tab, From, To}} -> Cp2 = do_change_copy(Cp, Tab, From, To), reply(From, Name, ok), retainer_loop(Cp2); {_From, {add_retainer, R, Node}} -> Cp2 = do_add_retainer(Cp, R, Node), retainer_loop(Cp2); - {_From, {del_retainer, R, Node}} when Cp#checkpoint_args.wait_for_old == [] -> + {_From, {del_retainer, R, Node}} -> Cp2 = do_del_retainer(Cp, R, Node), retainer_loop(Cp2); %% Iteration - {From, {iter_begin, Iter}} when Cp#checkpoint_args.wait_for_old == [] -> + {From, {iter_begin, Iter}} -> Cp2 = iter_begin(Cp, From, Iter), retainer_loop(Cp2); - {From, {iter_end, Iter}} when Cp#checkpoint_args.wait_for_old == [] -> + {From, {iter_end, Iter}} -> retainer_fixtable(Iter#iter.oid_tab, false), Iters = Cp#checkpoint_args.iterators -- [Iter], reply(From, Name, ok), - retainer_loop(Cp#checkpoint_args{iterators = Iters}); - - {_From, {exit_pending, Tid}} - when is_list(Cp#checkpoint_args.wait_for_old) -> - StillPending = lists:delete(Tid, Cp#checkpoint_args.wait_for_old), - Cp2 = Cp#checkpoint_args{wait_for_old = StillPending}, - Cp3 = maybe_activate(Cp2), - retainer_loop(Cp3); + retainer_loop(Cp#checkpoint_args{iterators = Iters}); - {From, collect_pending} -> - PendingTab = Cp#checkpoint_args.pending_tab, - del(pending_checkpoints, PendingTab), - Pending = ?ets_match_object(PendingTab, '_'), - reply(From, Name, {ok, Pending}), + {_From, {exit_pending, _Tid}} -> retainer_loop(Cp); - {From, {activate, Pending}} -> - StillPending = mnesia_recover:still_pending(Pending), - enter_still_pending(StillPending, Cp#checkpoint_args.pending_tab), - Cp2 = maybe_activate(Cp#checkpoint_args{wait_for_old = StillPending}), - reply(From, Name, activated), + {From, deactivate} -> + do_stop(Cp), + reply(From, Name, deactivated), + unlink(From), + exit(shutdown); + + {_From, {mnesia_down, Node}} -> + Cp2 = do_del_retainers(Cp, Node), retainer_loop(Cp2); + {'EXIT', Parent, _} when Parent == Cp#checkpoint_args.supervisor -> + %% do_stop(Cp), + %% assume that entire Mnesia is terminating + exit(shutdown); + {'EXIT', From, _Reason} -> Iters = [Iter || Iter <- Cp#checkpoint_args.iterators, check_iter(From, Iter)], @@ -903,13 +915,17 @@ retainer_loop(Cp) -> {system, From, Msg} -> dbg_out("~p got {system, ~p, ~p}~n", [?MODULE, From, Msg]), - sys:handle_system_msg(Msg, From, no_parent, ?MODULE, [], Cp) + sys:handle_system_msg(Msg, From, Cp#checkpoint_args.supervisor, + ?MODULE, [], Cp); + Msg -> + dbg_out("~p got ~p~n", [?MODULE, Msg]) end. maybe_activate(Cp) - when Cp#checkpoint_args.wait_for_old == [], - Cp#checkpoint_args.is_activated == false -> - Cp#checkpoint_args{pending_tab = undefined, is_activated = true}; + when Cp#checkpoint_args.wait_for_old == [], + Cp#checkpoint_args.is_activated == false -> + Cp#checkpoint_args{%% pending_tab = undefined, + is_activated = true}; maybe_activate(Cp) -> Cp. @@ -1226,65 +1242,6 @@ system_terminate(_Reason, _Parent,_Debug, Cp) -> system_code_change(Cp, _Module, _OldVsn, _Extra) -> {ok, Cp}. -convert_cp_record(Cp) when is_record(Cp, checkpoint) -> - ROD = - case Cp#checkpoint.ram_overrides_dump of - true -> Cp#checkpoint.min ++ Cp#checkpoint.max; - false -> [] - end, - - {ok, #checkpoint_args{name = Cp#checkpoint.name, - allow_remote = Cp#checkpoint.name, - ram_overrides_dump = ROD, - nodes = Cp#checkpoint.nodes, - node = Cp#checkpoint.node, - now = Cp#checkpoint.now, - cookie = ?unique_cookie, - min = Cp#checkpoint.min, - max = Cp#checkpoint.max, - pending_tab = Cp#checkpoint.pending_tab, - wait_for_old = Cp#checkpoint.wait_for_old, - is_activated = Cp#checkpoint.is_activated, - ignore_new = Cp#checkpoint.ignore_new, - retainers = Cp#checkpoint.retainers, - iterators = Cp#checkpoint.iterators, - supervisor = Cp#checkpoint.supervisor, - pid = Cp#checkpoint.pid - }}; -convert_cp_record(Cp) when is_record(Cp, checkpoint_args) -> - AllTabs = Cp#checkpoint_args.min ++ Cp#checkpoint_args.max, - ROD = case Cp#checkpoint_args.ram_overrides_dump of - [] -> - false; - AllTabs -> - true; - _ -> - error - end, - if - ROD == error -> - {error, {"Old node cannot handle new checkpoint protocol", - ram_overrides_dump}}; - true -> - {ok, #checkpoint{name = Cp#checkpoint_args.name, - allow_remote = Cp#checkpoint_args.name, - ram_overrides_dump = ROD, - nodes = Cp#checkpoint_args.nodes, - node = Cp#checkpoint_args.node, - now = Cp#checkpoint_args.now, - min = Cp#checkpoint_args.min, - max = Cp#checkpoint_args.max, - pending_tab = Cp#checkpoint_args.pending_tab, - wait_for_old = Cp#checkpoint_args.wait_for_old, - is_activated = Cp#checkpoint_args.is_activated, - ignore_new = Cp#checkpoint_args.ignore_new, - retainers = Cp#checkpoint_args.retainers, - iterators = Cp#checkpoint_args.iterators, - supervisor = Cp#checkpoint_args.supervisor, - pid = Cp#checkpoint_args.pid - }} - end. - %%%%%%%%%%%%%%%%%%%%%%%%%% val(Var) -> diff --git a/lib/mnesia/test/mnesia_consistency_test.erl b/lib/mnesia/test/mnesia_consistency_test.erl index 922b89ec2b..a76a0116d5 100644 --- a/lib/mnesia/test/mnesia_consistency_test.erl +++ b/lib/mnesia/test/mnesia_consistency_test.erl @@ -1,7 +1,7 @@ %% %% %CopyrightBegin% %% -%% Copyright Ericsson AB 1997-2012. All Rights Reserved. +%% Copyright Ericsson AB 1997-2013. 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 @@ -39,7 +39,6 @@ all() -> {group, consistency_after_move_replica}, {group, consistency_after_transform_table}, consistency_after_change_table_copy_type, - {group, consistency_after_fallback}, {group, consistency_after_restore}, consistency_after_rename_of_node, {group, checkpoint_retainer_consistency}, @@ -99,10 +98,14 @@ groups() -> [{group, updates_during_checkpoint_activation}, {group, updates_during_checkpoint_iteration}, {group, load_table_with_activated_checkpoint}, - {group, - add_table_copy_to_table_checkpoint}]}, + {group, add_table_copy_to_table_checkpoint}, + {group, consistency_after_fallback} + ]}, {updates_during_checkpoint_activation, [], - [updates_during_checkpoint_activation_2_ram, + [updates_during_checkpoint_activation_1_ram, + updates_during_checkpoint_activation_1_disc, + updates_during_checkpoint_activation_1_disc_only, + updates_during_checkpoint_activation_2_ram, updates_during_checkpoint_activation_2_disc, updates_during_checkpoint_activation_2_disc_only, updates_during_checkpoint_activation_3_ram, @@ -730,6 +733,18 @@ consistency_after_rename_of_node(doc) -> %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% +updates_during_checkpoint_activation_1_ram(suite) -> []; +updates_during_checkpoint_activation_1_ram(Config) when is_list(Config) -> + updates_during_checkpoint_activation(ram_copies, 1, Config). + +updates_during_checkpoint_activation_1_disc(suite) -> []; +updates_during_checkpoint_activation_1_disc(Config) when is_list(Config) -> + updates_during_checkpoint_activation(disc_copies, 1, Config). + +updates_during_checkpoint_activation_1_disc_only(suite) -> []; +updates_during_checkpoint_activation_1_disc_only(Config) when is_list(Config) -> + updates_during_checkpoint_activation(disc_only_copies, 1, Config). + updates_during_checkpoint_activation_2_ram(suite) -> []; updates_during_checkpoint_activation_2_ram(Config) when is_list(Config) -> updates_during_checkpoint_activation(ram_copies, 2, Config). @@ -771,7 +786,8 @@ updates_during_checkpoint_activation(ReplicaType,NodeConfig,Config) -> timer:sleep(timer:seconds(Delay)), {ok, CPName, _NodeList} = - mnesia:activate_checkpoint([{max, mnesia:system_info(tables)}]), + mnesia:activate_checkpoint([{max, mnesia:system_info(tables)}, + {ram_overrides_dump, true}]), timer:sleep(timer:seconds(Delay)), %% Stop tpcb |