Fix mnesia:force_load_table/1 getting stuck

Author: Mikaka27

lib/mnesia/src/mnesia_controller.erl

@@ -1186,7 +1186,8 @@ handle_info(#dumper_done{worker_pid=Pid, worker_res=Res}, State) ->
 
 handle_info(Done = #loader_done{worker_pid=WPid, table_name=Tab}, State0) ->
     LateQueue0 = State0#state.late_loader_queue,
-    State1 = State0#state{loader_pid = lists:keydelete(WPid,1,get_loaders(State0))},
+    {value, {WPid, Worker}, Loaders} = lists:keytake(WPid, 1, get_loaders(State0)),
+    State1 = State0#state{loader_pid = Loaders},
 
     State2 =
 	case Done#loader_done.is_loaded of
@@ -1235,19 +1236,25 @@ handle_info(Done = #loader_done{worker_pid=WPid, table_name=Tab}, State0) ->
 			ignore
 		end,
 
-                case {?catch_val({Tab, storage_type}), val({Tab, active_replicas})} of
-                    {unknown, _} -> %% Should not have a local copy anymore
-                        State1#state{late_loader_queue=gb_trees:delete_any(Tab, LateQueue0)};
-		    {_, [_|_]} -> % still available elsewhere
-			{value,{_,Worker}} = lists:keysearch(WPid,1,get_loaders(State0)),
-			add_loader(Tab,Worker,State1);
-                    {ram_copies, []} ->
-			DelState = State1#state{late_loader_queue=gb_trees:delete_any(Tab, LateQueue0)},
-                        cast({disc_load, Tab, ram_only}),
-                        DelState;
-                    {_, []} ->  %% Table deleted or not loaded anywhere
-                        State1#state{late_loader_queue=gb_trees:delete_any(Tab, LateQueue0)}
-		end
+            case {?catch_val({Tab, storage_type}), val({Tab, active_replicas}), val({Tab, load_by_force})} of
+                {unknown, _, _} -> %% Should not have a local copy anymore
+                    State1#state{late_loader_queue=gb_trees:delete_any(Tab, LateQueue0)};
+                {_, [_|_], _} -> % still available elsewhere
+                    add_loader(Tab,Worker,State1);
+                {ram_copies, [], _} ->
+                    DelState = State1#state{late_loader_queue=gb_trees:delete_any(Tab, LateQueue0)},
+                    cast({disc_load, Tab, ram_only}),
+                    DelState;
+                {_, [], true} when is_record(Worker, net_load) ->
+                    %% Network load could have been aborted by Sender node going down,
+                    %% if user has forced the load, we retry loading from disc
+                    DelState = State1#state{late_loader_queue=gb_trees:delete_any(Tab, LateQueue0)},
+                    cast({disc_load, Tab, forced_by_user}),
+                    DelState;
+                {_, [], _} ->
+                    %% Table deleted or not loaded anywhere
+                    State1#state{late_loader_queue=gb_trees:delete_any(Tab, LateQueue0)}
+            end
 	end,
     State3 = opt_start_worker(State2),
     noreply(State3);

lib/mnesia/src/mnesia_loader.erl

@@ -33,7 +33,7 @@
 	 net_load_table/4,
 	 send_table/4]).
 
--export([spawned_receiver/8]).    %% Spawned lock taking process
+-export([spawned_receiver/7]).    %% Spawned lock taking process
 
 -import(mnesia_lib, [set/2, fatal/2, verbose/2, dbg_out/2]).
 
@@ -271,24 +271,23 @@ init_receiver(Node, Tab,Storage,Cs,Reason) ->
                         mnesia_locker:rlock(Tid, Ts#tidstore.store, {schema, Tab});
                     _ -> ok
                 end,
-		%% Check that table still exists
-		Active = val({Tab, active_replicas}),
-		%% Check that we haven't loaded it already
-		case val({Tab,where_to_read}) == node() of
-		    true -> ok;
-		    _ ->
-			%% And that sender still got a copy
-			true = lists:member(Node, Active),
-			{SenderPid, TabSize, DetsData} =
-			    start_remote_sender(Node,Tab,Storage,load),
-			Init = table_init_fun(SenderPid, Storage),
-			Args = [self(),Tab,Storage,Cs,SenderPid,
-				TabSize,DetsData,Init],
-			Pid = spawn_link(?MODULE, spawned_receiver, Args),
-			put(mnesia_real_loader, Pid),
-			wait_on_load_complete(Pid)
-		end
-	end,
+            %% Check that table still exists
+            Active = val({Tab, active_replicas}),
+            %% Check that we haven't loaded it already
+            case val({Tab,where_to_read}) == node() of
+                true -> ok;
+                _ ->
+                    %% And that sender still got a copy
+                    true = lists:member(Node, Active),
+                    {SenderPid, TabSize, DetsData} =
+                        start_remote_sender(Node,Tab,Storage,load),
+                    Args = [self(),Tab,Storage,Cs,SenderPid,
+                            TabSize,DetsData],
+                    Pid = spawn_link(?MODULE, spawned_receiver, Args),
+                    put(mnesia_real_loader, Pid),
+                    wait_on_load_complete(Pid)
+            end
+    end,
     Res =
 	case mnesia:transaction(Load, 20) of
 	    {atomic, {error,Result}} when
@@ -339,17 +338,36 @@ start_remote_sender(Node,Tab,Storage, Why) ->
     end.
 
 table_init_fun(SenderPid, Storage) ->
+    Parent = self(),
     fun(read) ->
-	    % We want to store subscribed mnesia table events received during
-	    % table copying for later processing to not let receiver message queue
-	    % to grow too much (which in consequence would slow down the whole copying process)
-	    SubscrCache = ets:new(subscr_cache, [private, ordered_set]),
-	    put(mnesia_table_receiver_subscr_cache, SubscrCache),
-	    Receiver = self(),
-	    SenderPid ! {Receiver, more},
-	    get_data(SenderPid, Receiver, Storage);
+            % If this function is executed by a pid different then the loader pid, we have to
+            % register with controller to receive {copier_done, Node}. This is happening
+            % with disc_only_copies, where this is executed by dets_server.
+            case self() of
+                Parent ->
+                    ok;
+                _ ->
+                    ok = mnesia_controller:call({add_other, self()})
+            end,
+            try
+                % We want to store subscribed mnesia table events received during
+                % table copying for later processing to not let receiver message queue
+                % to grow too much (which in consequence would slow down the whole copying process)
+                SubscrCache = ets:new(subscr_cache, [private, ordered_set]),
+                put(mnesia_table_receiver_subscr_cache, SubscrCache),
+                Receiver = self(),
+                SenderPid ! {Receiver, more},
+                get_data(SenderPid, Receiver, Storage)
+            after
+                case self() of
+                    Parent ->
+                        ok;
+                    _ ->
+                        mnesia_controller:call({del_other, self()})
+                end
+            end;
        (close) ->
-	    ok
+            ok
     end.
 
 %% Add_table_copy gets it's own locks.
@@ -364,8 +382,9 @@ start_receiver(Tab,Storage,Cs,SenderPid,TabSize,DetsData,{dumper,{add_table_copy
 	    Else
     end.
 
-spawned_receiver(ReplyTo,Tab,Storage,Cs, SenderPid,TabSize,DetsData, Init) ->
+spawned_receiver(ReplyTo,Tab,Storage,Cs, SenderPid,TabSize,DetsData) ->
     process_flag(trap_exit, true),
+    Init = table_init_fun(SenderPid, Storage),
     Done = do_init_table(Tab,Storage,Cs,
 			 SenderPid,TabSize,DetsData,
 			 ReplyTo, Init),

lib/mnesia/test/mnesia_durability_test.erl

@@ -49,6 +49,12 @@
          master_on_non_local_tables/1,
          remote_force_load_with_local_master_node/1,
          master_node_with_ram_copy_2/1, master_node_with_ram_copy_3/1,
+         force_load_table_when_loader_aborts_ram/1,
+         force_load_table_when_loader_aborts_disc/1,
+         force_load_table_when_loader_aborts_disc_only/1,
+         force_load_table_when_loader_aborts_ext_ram/1,
+         force_load_table_when_loader_aborts_ext_disc_only/1,
+         force_load_table_when_loader_on_different_pid/1,
          dump_ram_copies/1, dump_disc_copies/1, dump_disc_only/1]).
 
 -include("mnesia_test_lib.hrl").
@@ -71,7 +77,7 @@ all() ->
      durability_of_disc_copies,
      durability_of_disc_only_copies].
 
-groups() -> 
+groups() ->
     [{load_tables, [],
       [load_latest_data, load_local_contents_directly,
        load_directly_when_all_are_ram_copiesA,
@@ -97,8 +103,18 @@ groups() ->
        remote_force_load_with_local_master_node,
        master_node_with_ram_copy_2, master_node_with_ram_copy_3]},
      {durability_of_dump_tables, [],
-      [dump_ram_copies, dump_disc_copies, dump_disc_only]}].
-
+      [dump_ram_copies, dump_disc_copies, dump_disc_only]},
+     {load_tables_with_network_down, [],
+      [force_load_table_when_loader_aborts_ram,
+       force_load_table_when_loader_aborts_disc,
+       force_load_table_when_loader_aborts_disc_only,
+       force_load_table_when_loader_aborts_ext_ram,
+       force_load_table_when_loader_aborts_ext_disc_only,
+       force_load_table_when_loader_on_different_pid]
+     }].
+
+init_per_group(load_tables_with_network_down, Config) ->
+    mnesia_test_lib:skip_if_no_network_blocker(Config);
 init_per_group(_GroupName, Config) ->
     Config.
 
@@ -1270,6 +1286,75 @@ master_node_with_ram_copy_3(Config) when is_list(Config) ->
 
     ?verify_mnesia(Nodes, []).
 
+force_load_table_when_loader_aborts_ram(Config) ->
+    force_load_table_when_loader_aborts(Config, ram_copies).
+
+force_load_table_when_loader_aborts_disc(Config) ->
+    force_load_table_when_loader_aborts(Config, disc_copies).
+
+force_load_table_when_loader_aborts_disc_only(Config) ->
+    force_load_table_when_loader_aborts(Config, disc_only_copies).
+
+force_load_table_when_loader_aborts_ext_ram(Config) ->
+    force_load_table_when_loader_aborts(Config, ext_ram_copies).
+
+force_load_table_when_loader_aborts_ext_disc_only(Config) ->
+    force_load_table_when_loader_aborts(Config, ext_disc_only_copies).
+
+force_load_table_when_loader_aborts(Config, Storage) ->
+    [Node1, Node2] = Nodes = ?acquire_nodes(2, Config),
+    Table = ?FUNCTION_NAME,
+    Populate = fun(F, N) when N > 0 ->
+                       mnesia:write({Table, N, []}),
+                       F(F, N - 1);
+                  (_F, _N) ->
+                       ok
+               end,
+    ?match({atomic, ok}, mnesia:create_table(Table, [{Storage, Nodes}])),
+    ?match({atomic, ok}, mnesia:transaction(fun() -> Populate(Populate, 1_000_000) end)),
+    ?match(ok, mnesia:set_master_nodes([Node2])),
+    ?match(stopped, mnesia:stop()),
+    ?match(ok, mnesia:start()),
+    ?match(ok, mnesia_test_lib:block_peer(Node1, Node2)),
+    ?match({timeout, [Table]}, mnesia:wait_for_tables([Table], 0)),
+    ?match(pang, net_adm:ping(Node2)),
+    {Pid, Ref} = spawn_monitor(fun() -> yes = mnesia:force_load_table(Table) end),
+    receive
+        {'DOWN', Ref, process, Pid, _} ->
+            ok
+    after timer:seconds(15) ->
+            ct:fail("Pid: ~p stuck in:~n~p~n",
+                    [Pid, process_info(Pid, current_stacktrace)])
+    end.
+
+force_load_table_when_loader_on_different_pid(Config) ->
+    [Node1, Node2] = Nodes = ?acquire_nodes(2, Config),
+    Table = ?FUNCTION_NAME,
+    Populate = fun(F, N) when N > 0 ->
+                       mnesia:write({Table, N, []}),
+                       F(F, N - 1);
+                  (_F, _N) ->
+                       ok
+               end,
+    ?match({atomic, ok}, mnesia:create_table(Table, [{disc_only_copies, Nodes}])),
+    ?match({atomic, ok}, mnesia:transaction(fun() -> Populate(Populate, 500_000) end)),
+    ?match(ok, mnesia:set_master_nodes([Node2])),
+    ?match(stopped, mnesia:stop()),
+    ?match(ok, mnesia:start()),
+    ?match(ok, mnesia_test_lib:block_peer(Node1, Node2)),
+    ?match({timeout, [Table]}, mnesia:wait_for_tables([Table], 0)),
+    ?match(pang, net_adm:ping(Node2)),
+    ?match(ok, timer:sleep(timer:seconds(1))),
+    {Pid, Ref} = spawn_monitor(fun() -> yes = mnesia:force_load_table(Table) end),
+    receive
+        {'DOWN', Ref, process, Pid, _} ->
+            ok
+    after timer:seconds(15) ->
+            ct:fail("Pid: ~p stuck in:~n~p~n",
+                    [Pid, [process_info(dets_server:get_pid(Table), K) ||
+                              K <- [monitored_by, current_stacktrace]]])
+    end.
+
 %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%