File 5271-mnesia-Improve-dirty-writes-consistency.patch of Package erlang

Overview Repositories Revisions Requests Users Attributes Meta

File 5271-mnesia-Improve-dirty-writes-consistency.patch of Package erlang (Revision 70b2d07037c143e87423048b386444cf)

Currently displaying revision 70b2d07037c143e87423048b386444cf , Show latest

From 1731534c5e97731af691eb1228081b0ca3af3c28 Mon Sep 17 00:00:00 2001
From: Dan Gudmundsson <dgud@erlang.org>
Date: Tue, 24 Jan 2023 15:08:22 +0100
Subject: [PATCH 1/4] mnesia: Improve dirty writes consistency

Improve table copying timing, where dirty writes can cause consistency issues
during mnesia:add_table_copy/3 call.

I.e. don't set where_to_read to early.
Keep subscribers longer (sync with mnesia_tm for writes in the msg queue),
and let reciever wait until copier is done.
---
 lib/mnesia/src/mnesia_loader.erl            | 36 +++++----
 lib/mnesia/src/mnesia_schema.erl            |  1 +
 lib/mnesia/src/mnesia_tm.erl                | 88 ++++++++++++++-------
 lib/mnesia/test/mnesia_consistency_test.erl | 16 ++--
 lib/mnesia/test/mt.erl                      | 22 +++---
 5 files changed, 104 insertions(+), 59 deletions(-)

diff --git a/lib/mnesia/src/mnesia_loader.erl b/lib/mnesia/src/mnesia_loader.erl
index 4dea366964..ed5dba3a16 100644
--- a/lib/mnesia/src/mnesia_loader.erl
+++ b/lib/mnesia/src/mnesia_loader.erl
@@ -222,7 +222,6 @@ do_get_network_copy(Tab, Reason, Ns, Storage, Cs) ->
 		ok ->
 		    set({Tab, load_node}, Node),
 		    set({Tab, load_reason}, Reason),
-		    mnesia_controller:i_have_tab(Tab, Cs),
 		    dbg_out("Table ~tp copied from ~p to ~p~n", [Tab, Node, node()]),
 		    {loaded, ok};
 		Err = {error, _} when element(1, Reason) == dumper ->
@@ -291,7 +290,11 @@ init_receiver(Node, Tab,Storage,Cs,Reason) ->
 	    {atomic, {error,Result}} ->
 		fatal("Cannot create table ~tp: ~tp~n",
 		      [[Tab, Storage], Result]);
-	    {atomic,  Result} -> Result;
+            {atomic, ok} ->
+                mnesia_controller:i_have_tab(Tab, Cs),
+                ok;
+	    {atomic, Result} ->
+                Result;
 	    {aborted, nomore} -> restart;
 	    {aborted, _Reas} ->
 		verbose("Receiver failed on ~tp from ~p:~nReason: ~tp~n",
@@ -555,14 +558,17 @@ init_table(Tab, _, Fun, _DetsInfo,_) ->
 
 finish_copy(Storage,Tab,Cs,SenderPid,DatBin,OrigTabRec) ->
     TabRef = {Storage, Tab},
-    subscr_postprocess(TabRef, Cs#cstruct.record_name),
     case handle_last(TabRef, Cs#cstruct.type, DatBin) of
 	ok ->
-	    mnesia_index:init_index(Tab, Storage),
-	    snmpify(Tab, Storage),
+            subscr_postprocess(TabRef, Cs#cstruct.record_name),
 	    %% OrigTabRec must not be the spawned tab-receiver
 	    %% due to old protocol.
 	    SenderPid ! {OrigTabRec, no_more},
+            Ref = monitor(process, SenderPid),
+            %% and all remaining events
+            subscr_receiver(TabRef, Cs#cstruct.record_name, Ref),
+	    mnesia_index:init_index(Tab, Storage),
+	    snmpify(Tab, Storage),
 	    mnesia_tm:unblock_tab(Tab),
 	    ok;
 	{error, Reason} ->
@@ -582,22 +588,21 @@ subscr_postprocess(TabRef, RecName) ->
 		    handle_subscr_event(Event, TabRef, RecName)
 		end, ok, SubscrCache),
 	    ets:delete(SubscrCache)
-    end,
-    % and all remaining events
-    subscr_receiver(TabRef, RecName).
+    end.
 
-subscr_receiver(TabRef = {_, Tab}, RecName) ->
+subscr_receiver(TabRef = {_, Tab}, RecName, Ref) ->
     receive
 	{mnesia_table_event, {_Op, Val, _Tid}} = Event
 	  when element(1, Val) =:= Tab; element(1, Val) =:= schema ->
 	    handle_subscr_event(Event, TabRef, RecName),
-	    subscr_receiver(TabRef, RecName);
+	    subscr_receiver(TabRef, RecName, Ref);
 
 	{'EXIT', Pid, Reason} ->
 	    handle_exit(Pid, Reason),
-	    subscr_receiver(TabRef, RecName)
-    after 0 ->
-	    ok
+	    subscr_receiver(TabRef, RecName, Ref);
+
+        {'DOWN', Ref, process, _, _} ->
+            ok
     end.
 
 handle_subscr_event(Event, TabRef = {_, Tab}, RecName) ->
@@ -1014,12 +1019,15 @@ finish_copy(Pid, Tab, Storage, RemoteS, NeedLock) ->
                         mnesia_checkpoint:tm_add_copy(Tab, RecNode),
                         DatBin = dat2bin(Tab, ?catch_val({Tab, storage_type}), RemoteS),
                         Pid ! {self(), {no_more, DatBin}},
-                        cleanup_tab_copier(Pid, Storage, Tab),
                         receive
                             {Pid, no_more} -> % Dont bother about the spurious 'more' message
+                                %% Sync mnesia_tm (before unsubscribing)
+                                mnesia_tm:sync(),
+                                cleanup_tab_copier(Pid, Storage, Tab),
                                 no_more;
                             {copier_done, Node} ->
                                 verbose("Tab receiver ~tp crashed (more): ~p~n", [Tab, Node]),
+                                cleanup_tab_copier(Pid, Storage, Tab),
                                 receiver_died
                         end
                 end
diff --git a/lib/mnesia/src/mnesia_schema.erl b/lib/mnesia/src/mnesia_schema.erl
index 42c717425e..ea77e5ec6a 100644
--- a/lib/mnesia/src/mnesia_schema.erl
+++ b/lib/mnesia/src/mnesia_schema.erl
@@ -2472,6 +2472,7 @@ prepare_op(Tid, {op, add_table_copy, Storage, Node, TabDef}, _WaitFor) ->
 		{loaded, ok} ->
                     %% Tables are created by mnesia_loader get_network code
                     insert_cstruct(Tid, Cs, true),
+                    mnesia_controller:i_have_tab(Tab, Cs),
 		    {true, optional};
                 {not_loaded, {not_active, schema, Node}} ->
                     mnesia:abort({node_not_running, Node});
diff --git a/lib/mnesia/src/mnesia_tm.erl b/lib/mnesia/src/mnesia_tm.erl
index 847cd0074b..3a563db0be 100644
--- a/lib/mnesia/src/mnesia_tm.erl
+++ b/lib/mnesia/src/mnesia_tm.erl
@@ -42,6 +42,7 @@
 	 put_activity_id/2,
 	 block_tab/1,
 	 unblock_tab/1,
+         sync/0,
 	 fixtable/3,
 	 new_cr_format/1
 	]).
@@ -205,6 +206,17 @@ block_tab(Tab) ->
 unblock_tab(Tab) ->
     req({unblock_tab, Tab}).
 
+fixtable(Tab, Lock, Me) ->
+    case req({fixtable, [Tab,Lock,Me]}) of
+	error ->
+	    exit({no_exists, Tab});
+	Else ->
+	    Else
+    end.
+
+sync() ->
+    req(sync).
+
 doit_loop(#state{coordinators=Coordinators,participants=Participants,supervisor=Sup}=State) ->
     receive
 	{_From, {async_dirty, Tid, Commit, Tab}} ->
@@ -250,25 +262,30 @@ doit_loop(#state{coordinators=Coordinators,participants=Participants,supervisor=
 			    [{tid, Tid}, {prot, Protocol}]),
 	    mnesia_checkpoint:tm_enter_pending(Tid, DiscNs, RamNs),
 	    Commit = new_cr_format(Commit0),
-	    Pid =
-                if
-                    node(Tid#tid.pid) =:= node() ->
-                        error({internal_error, local_node});
-                    Protocol =:= asym_trans orelse Protocol =:= sync_asym_trans ->
-			Args = [Protocol, tmpid(From), Tid, Commit, DiscNs, RamNs],
-			spawn_link(?MODULE, commit_participant, Args);
-                    true -> %% *_sym_trans
-			reply(From, {vote_yes, Tid}),
-			nopid
-		end,
-	    P = #participant{tid = Tid,
-			     pid = Pid,
-			     commit = Commit,
-			     disc_nodes = DiscNs,
-			     ram_nodes = RamNs,
-			     protocol = Protocol},
-	    State2 = State#state{participants = gb_trees:insert(Tid,P,Participants)},
-	    doit_loop(State2);
+            case is_blocked(State#state.blocked_tabs, Commit) of
+                false ->
+                    Pid =
+                        if
+                            node(Tid#tid.pid) =:= node() ->
+                                error({internal_error, local_node});
+                            Protocol =:= asym_trans orelse Protocol =:= sync_asym_trans ->
+                                Args = [Protocol, tmpid(From), Tid, Commit, DiscNs, RamNs],
+                                spawn_link(?MODULE, commit_participant, Args);
+                            true -> %% *_sym_trans
+                                reply(From, {vote_yes, Tid}),
+                                nopid
+                        end,
+                    P = #participant{tid = Tid,
+                                     pid = Pid,
+                                     commit = Commit,
+                                     disc_nodes = DiscNs,
+                                     ram_nodes = RamNs,
+                                     protocol = Protocol},
+                    State2 = State#state{participants = gb_trees:insert(Tid,P,Participants)},
+                    doit_loop(State2);
+                true ->
+                    reply(From, {vote_no, Tid, {bad_commit, node()}}, State)
+            end;
 
 	{Tid, do_commit} ->
 	    case gb_trees:lookup(Tid, Participants) of
@@ -449,6 +466,9 @@ doit_loop(#state{coordinators=Coordinators,participants=Participants,supervisor=
 		    reply(From, ok, State2)
 	    end;
 
+        {From, sync} ->
+            reply(From, ok, State);
+
 	{From, {prepare_checkpoint, Cp}} ->
 	    Res = mnesia_checkpoint:tm_prepare(Cp),
 	    case Res of
@@ -478,6 +498,28 @@ doit_loop(#state{coordinators=Coordinators,participants=Participants,supervisor=
 	    doit_loop(State)
     end.
 
+is_blocked([], _Commit) ->
+    false;
+is_blocked([Tab|Tabs], #commit{ram_copies=RCs, disc_copies=DCs,
+                               disc_only_copies=DOs, ext=Exts} = Commit) ->
+    is_blocked_tab(RCs, Tab) orelse
+        is_blocked_tab(DCs, Tab) orelse
+        is_blocked_tab(DOs, Tab) orelse
+        is_blocked_ext_tab(Exts, Tab) orelse
+        is_blocked(Tabs, Commit).
+
+is_blocked_tab([{{Tab,_},_,_}|_Ops], Tab) -> true;
+is_blocked_tab([_|Ops], Tab) -> is_blocked_tab(Ops, Tab);
+is_blocked_tab([],_) -> false.
+
+is_blocked_ext_tab([], _Tab) ->
+    false;
+is_blocked_ext_tab(Exts, Tab) ->
+    case lists:keyfind(ext_copies, 1, Exts) of
+        false -> false;
+        {_, ExtOps} -> is_blocked_tab([Op || {_, Op} <- ExtOps], Tab)
+    end.
+
 do_sync_dirty(From, Tid, Commit, _Tab) ->
     ?eval_debug_fun({?MODULE, sync_dirty, pre}, [{tid, Tid}]),
     Res = do_dirty(Tid, Commit),
@@ -2326,14 +2368,6 @@ do_stop(#state{coordinators = Coordinators}) ->
     mnesia_log:stop(),
     exit(shutdown).
 
-fixtable(Tab, Lock, Me) ->
-    case req({fixtable, [Tab,Lock,Me]}) of
-	error ->
-	    exit({no_exists, Tab});
-	Else ->
-	    Else
-    end.
-
 %%%%%%%%%%%%%%%%%%%%%%%%%%%
 %% System upgrade
 
diff --git a/lib/mnesia/test/mnesia_consistency_test.erl b/lib/mnesia/test/mnesia_consistency_test.erl
index 2be17e69c2..fa729aa714 100644
--- a/lib/mnesia/test/mnesia_consistency_test.erl
+++ b/lib/mnesia/test/mnesia_consistency_test.erl
@@ -381,7 +381,7 @@ consistency_after_restart(ReplicaType, NodeConfig, Config) ->
     TpcbConfig = tpcb_config(ReplicaType, NodeConfig, Nodes, [Node1]),
     mnesia_tpcb:init(TpcbConfig),
     A ! fun () -> mnesia_tpcb:run(TpcbConfig) end,
-    timer:sleep(timer:seconds(10)),
+    timer:sleep(timer:seconds(3)),
     mnesia_test_lib:kill_mnesia([Node1]),
     %% Start and wait for tables to be loaded on all nodes
     timer:sleep(timer:seconds(3)),
@@ -408,7 +408,7 @@ consistency_after_dump_tables(ReplicaType, NodeConfig, Config) ->
     TpcbConfig = tpcb_config(ReplicaType, NodeConfig, Nodes, []),
     mnesia_tpcb:init(TpcbConfig),
     A ! fun() -> mnesia_tpcb:run(TpcbConfig) end,
-    timer:sleep(timer:seconds(10)),
+    timer:sleep(timer:seconds(3)),
     ?match({atomic, ok}, rpc:call(Node1, mnesia, dump_tables,
                         [[branch, teller, account, history]])),
     mnesia_tpcb:stop(),
@@ -459,7 +459,7 @@ consistency_after_add_replica(ReplicaType, NodeConfig, Config) ->
     TpcbConfig = tpcb_config(ReplicaType, NodeConfig, Nodes, []),
     mnesia_tpcb:init(TpcbConfig),
     A ! fun () -> mnesia_tpcb:run(TpcbConfig) end,
-    timer:sleep(timer:seconds(10)),
+    timer:sleep(timer:seconds(2)),
     ?match({atomic, ok}, mnesia:add_table_copy(account, AddNode, ReplicaType)),
     mnesia_tpcb:stop(),
     ?match(ok, mnesia_tpcb:verify_tabs()),
@@ -501,7 +501,7 @@ consistency_after_del_replica(ReplicaType, NodeConfig, Config) ->
     TpcbConfig = tpcb_config(ReplicaType, NodeConfig, Nodes, []),
     mnesia_tpcb:init(TpcbConfig),
     A ! fun () -> mnesia_tpcb:run(TpcbConfig) end,
-    timer:sleep(timer:seconds(10)),
+    timer:sleep(timer:seconds(3)),
     ?match({atomic, ok}, mnesia:del_table_copy(account, Node2)),
     mnesia_tpcb:stop(),
     ?match(ok, mnesia_tpcb:verify_tabs()),
@@ -543,7 +543,7 @@ consistency_after_move_replica(ReplicaType, NodeConfig, Config) ->
     TpcbConfig = tpcb_config(ReplicaType, NodeConfig, Nodes -- [Node2], []),
     mnesia_tpcb:init(TpcbConfig),
     A ! fun () -> mnesia_tpcb:run(TpcbConfig) end,
-    timer:sleep(timer:seconds(10)),
+    timer:sleep(timer:seconds(3)),
     ?match({atomic, ok}, mnesia:move_table_copy(account, Node1, Node2)),    
     ?log("First move completed from node ~p to ~p ~n", [Node1, Node2]),
     ?match({atomic, ok}, mnesia:move_table_copy(account, Node2, Node1)),
@@ -638,7 +638,7 @@ consistency_after_fallback_3_disc_only(Config) when is_list(Config) ->
 consistency_after_fallback(ReplicaType, NodeConfig, Config) ->
     put(mnesia_test_verbose, true),
     %%?verbose("Starting consistency_after_fallback2 at ~p~n", [self()]),
-    Delay = 5,
+    Delay = 3,
     Nodes = ?acquire_nodes(NodeConfig, [{tc_timeout, timer:minutes(10)} | Config]),
     Node1 = hd(Nodes),
     %%?verbose("Mnesia info: ~p~n", [mnesia:info()]),
@@ -867,7 +867,7 @@ updates_during_checkpoint_activation_3_disc_only(Config) when is_list(Config) ->
 
 updates_during_checkpoint_activation(ReplicaType,NodeConfig,Config) ->
     %%?verbose("updates_during_checkpoint_activation2 at ~p~n", [self()]),
-    Delay = 5,
+    Delay = 2,
     Nodes = ?acquire_nodes(NodeConfig, Config),
     Node1 = hd(Nodes),
     %%?verbose("Mnesia info: ~p~n", [mnesia:info()]),
@@ -922,7 +922,7 @@ updates_during_checkpoint_iteration_2_disc_only(Config) when is_list(Config) ->
 
 updates_during_checkpoint_iteration(ReplicaType,NodeConfig,Config) ->
    %?verbose("updates_during_checkpoint_iteration2 at ~p~n", [self()]),
-    Delay = 5,
+    Delay = 2,
     Nodes = ?acquire_nodes(NodeConfig, Config),
     Node1 = hd(Nodes),
    %?verbose("Mnesia info: ~p~n", [mnesia:info()]),
diff --git a/lib/mnesia/test/mt.erl b/lib/mnesia/test/mt.erl
index c1859bef3f..0bb6aadc4e 100644
--- a/lib/mnesia/test/mt.erl
+++ b/lib/mnesia/test/mt.erl
@@ -245,30 +245,32 @@ start_nodes() ->
 %% loop one testcase /suite until it fails
 
 loop(Case) ->
-    loop_1(Case,-1,read_config()).
+    loop_1(Case,1,infinity,read_config()).
 
 loop(M,F) when is_atom(F) ->
-    loop_1({M,F},-1,read_config());
+    loop_1({M,F},1, infinity, read_config());
 loop(Case,N) when is_integer(N) ->
-    loop_1(Case, N,read_config()).
+    loop_1(Case, 1, N,read_config()).
 
 loop(M,F,N) when is_integer(N) ->
-    loop_1({M,F},N,read_config()).
+    loop_1({M,F},1, N,read_config()).
+
+loop_1(Case,N,Max,Config) when N < Max ->
+    io:format("~nLoop test ~p ~n", [abs(N)]),
 
-loop_1(Case,N,Config) when N /= 0 ->
-    io:format("Loop test ~p ~n", [abs(N)]),
     case ok_result(Res = t(Case,Config)) of
 	true ->
-	    loop_1(Case,N-1,Config);
+	    loop_1(Case,N+1,Max,Config);
 	error ->
+            io:format("Failed after ~p~n", [N]),
 	    Res
     end;
-loop_1(_,_,_) ->
+loop_1(_,_,_,_) ->
     ok.
-	    
+
 ok_result([{_T,{ok,_,_}}|R]) -> 
     ok_result(R);
-ok_result([{_T,{TC,List}}|R]) when is_tuple(TC), is_list(List) -> 
+ok_result([{_T,{TC,List}}|R]) when is_tuple(TC), is_list(List) ->
     ok_result(List) andalso ok_result(R);
 ok_result([]) -> true;
 ok_result(_) -> error.
-- 
2.35.3

Places

File 5271-mnesia-Improve-dirty-writes-consistency.patch of Package erlang (Revision 70b2d07037c143e87423048b386444cf)

Places