Skip to content

michal/mnesia/fix-mnesia-force-load-table-getting-stuck/OTP-20256 #11426

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Mikaka27 wants to merge 3 commits into
base: maint
Choose a base branch
from
Open

michal/mnesia/fix-mnesia-force-load-table-getting-stuck/OTP-20256 #11426

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
9309f7d
Skip not implemented test
Mikaka27 Aug 12, 2026
fffa95e
Add possibility to simulate network outage between nodes
Mikaka27 Aug 13, 2026
7e26a40
Fix mnesia:force_load_table/1 getting stuck
Mikaka27 Aug 13, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
186 changes: 111 additions & 75 deletions lib/mnesia/src/mnesia_controller.erl
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -116,19 +116,19 @@
-define(SERVER_NAME, ?MODULE).

-record(state, {supervisor,
schema_is_merged = false,
early_msgs = [],
loader_pid = [], %% Was Pid is now [{Pid,Work}|..]
loader_queue, %% Was list is now gb_tree
sender_pid = [], %% Was a pid or undef is now [{Pid,Work}|..]
sender_queue = [],
late_loader_queue, %% Was list is now gb_tree
dumper_pid, %% Dumper or schema commit pid
dumper_queue = [], %% Dumper or schema commit queue
others = [], %% Processes that needs the copier_done msg
dump_log_timer_ref,
is_stopping = false
}).
schema_is_merged = false,
early_msgs = [],
loader_pid = [], %% Was Pid is now [{Pid,Work}|..]
loader_queue, %% Was list is now gb_tree
sender_pid = [], %% Was a pid or undef is now [{Pid,Work}|..]
sender_queue = [],
late_loader_queue, %% Was list is now gb_tree
dumper_pid, %% Dumper or schema commit pid
dumper_queue = [], %% Dumper or schema commit queue
others = [], %% {Pid, Ref | undefined} of processes that needs the copier_done msg
dump_log_timer_ref,
is_stopping = false
}).
%% Backwards Comp. Sender_pid is now a list of senders..
get_senders(#state{sender_pid = Pids}) when is_list(Pids) -> Pids.
%% Backwards Comp. loader_pid is now a list of loaders..
Expand Down Expand Up @@ -786,16 +786,39 @@ handle_call({unannounce_add_table_copy, [Tab, Node], From}, ReplyTo, State) ->
noreply(State#state{early_msgs = [{call, Msg, undefined} | Msgs]})
end;

handle_call({add_other_receiver, Who, SenderNode}, _From, State = #state{others=Others0, schema_is_merged=SM}) ->
case SM of
true ->
case lists:member(SenderNode, val({current, db_nodes})) of
true ->
ok;
false ->
%% Additional receiver process was registered after mnesia_down was processed
%% for Node, we must send a message that was missed
Who ! {copier_done, SenderNode}
end,
Ref = erlang:monitor(process, Who),
Others = [{Who,Ref}|Others0],
{reply, ok, State#state{others = Others}};
false ->
{reply, {error, {not_active, schema, node()}}, State}
end;
handle_call({add_other, Who}, _From, State = #state{others=Others0, schema_is_merged=SM}) ->
case SM of
true ->
Others = [Who|Others0],
Others = [{Who,undefined}|Others0],
{reply, ok, State#state{others=Others}};
false ->
{reply, {error, {not_active,schema,node()}}, State}
end;
handle_call({del_other, Who}, _From, State = #state{others=Others0}) ->
Others = lists:delete(Who, Others0),
{value, {Who, Ref}, Others} = lists:keytake(Who, 1, Others0),
case is_reference(Ref) of
true ->
demonitor(Ref, [flush]);
false ->
ok
end,
{reply, ok, State#state{others=Others}};

handle_call(Msg, From, State) when State#state.schema_is_merged /= true ->
Expand Down Expand Up @@ -987,8 +1010,8 @@ handle_cast({mnesia_down, Node}, State) ->
lists:foreach(fun({Pid,_}) -> Pid ! {copier_done, Node} end,
Senders)
end,
lists:foreach(fun(Pid) -> Pid ! {copier_done,Node} end,
State#state.others),
lists:foreach(fun({Pid,_Ref}) -> Pid ! {copier_done,Node} end,
State#state.others),

Remove = fun(ST) ->
node(ST#send_table.receiver_pid) /= Node
Expand Down Expand Up @@ -1184,71 +1207,75 @@ handle_info(#dumper_done{worker_pid=Pid, worker_res=Res}, State) ->
{stop, fatal, State}
end;

handle_info(Done = #loader_done{worker_pid=WPid, table_name=Tab}, State0) ->
handle_info(Done = #loader_done{worker_pid=WPid, table_name=Tab, reply=Reply}, 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
true ->
%% Optional table announcement
if
Done#loader_done.needs_announce == true,
Done#loader_done.needs_reply == true ->
i_have_tab(Tab),
%% Should be {dumper,{add_table_copy, _}} only
reply(Done#loader_done.reply_to,
Done#loader_done.reply);
Done#loader_done.needs_reply == true ->
%% Should be {dumper,{add_table_copy,_}} only
reply(Done#loader_done.reply_to,
Done#loader_done.reply);
Done#loader_done.needs_announce == true, Tab == schema ->
i_have_tab(Tab);
Done#loader_done.needs_announce == true ->
i_have_tab(Tab),
%% Local node needs to perform user_sync_tab/1
Ns = val({current, db_nodes}),
abcast(Ns, {i_have_tab, Tab, node()});
Tab == schema ->
ignore;
true ->
%% Local node needs to perform user_sync_tab/1
Ns = val({current, db_nodes}),
AlreadyKnows = val({Tab, active_replicas}),
abcast(Ns -- AlreadyKnows, {i_have_tab, Tab, node()})
end,
%% Optional user sync
case Done#loader_done.needs_sync of
true -> user_sync_tab(Tab);
false -> ignore
end,
State1#state{late_loader_queue=gb_trees:delete_any(Tab, LateQueue0)};
false ->
%% Either the node went down or table was not
%% loaded remotly yet
case Done#loader_done.needs_reply of
true ->
reply(Done#loader_done.reply_to,
Done#loader_done.reply);
false ->
ignore
end,

case {?catch_val({Tab, storage_type}), val({Tab, active_replicas})} of
{unknown, _} -> %% Should not have a local copy anymore
case Done#loader_done.is_loaded of
true ->
%% Optional table announcement
if
Done#loader_done.needs_announce == true,
Done#loader_done.needs_reply == true ->
i_have_tab(Tab),
%% Should be {dumper,{add_table_copy, _}} only
reply(Done#loader_done.reply_to, Reply);
Done#loader_done.needs_reply == true ->
%% Should be {dumper,{add_table_copy,_}} only
reply(Done#loader_done.reply_to, Reply);
Done#loader_done.needs_announce == true, Tab == schema ->
i_have_tab(Tab);
Done#loader_done.needs_announce == true ->
i_have_tab(Tab),
%% Local node needs to perform user_sync_tab/1
Ns = val({current, db_nodes}),
abcast(Ns, {i_have_tab, Tab, node()});
Tab == schema ->
ignore;
true ->
%% Local node needs to perform user_sync_tab/1
Ns = val({current, db_nodes}),
AlreadyKnows = val({Tab, active_replicas}),
abcast(Ns -- AlreadyKnows, {i_have_tab, Tab, node()})
end,
%% Optional user sync
case Done#loader_done.needs_sync of
true -> user_sync_tab(Tab);
false -> ignore
end,
State1#state{late_loader_queue=gb_trees:delete_any(Tab, LateQueue0)};
false ->
%% Either the node went down or table was not
%% loaded remotly yet
case Done#loader_done.needs_reply of
true ->
reply(Done#loader_done.reply_to, Reply);
false ->
ignore
end,

case {?catch_val({Tab, storage_type}), val({Tab, active_replicas}), ?catch_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
{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)},
{_, [_|_], _} -> % 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;
{_, []} -> %% Table deleted or not loaded anywhere
{_, [], true} when is_record(Worker, net_load), Reply =:= {not_loaded, none_active} ->
%% 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,
end
end,
State3 = opt_start_worker(State2),
noreply(State3);

Expand Down Expand Up @@ -1311,6 +1338,15 @@ handle_info(Msg = {'EXIT', Pid, R}, State) when R /= wait_for_tables_timeout ->
end
end;

handle_info(Msg = {'DOWN', Ref, process, Pid, _Reason}, State = #state{others = Others0}) ->
case lists:keytake(Pid, 1, Others0) of
{value, {Pid, Ref}, Others} ->
%% Monitored receiver has died, remove it from others
noreply(State#state{others = Others});
false ->
error("~p got unexpected info: ~tp~n", [?SERVER_NAME, Msg])
end;

handle_info({From, get_state}, State) ->
From ! {?SERVER_NAME, From, State},
noreply(State);
Expand Down
113 changes: 68 additions & 45 deletions lib/mnesia/src/mnesia_loader.erl
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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]).

Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -339,17 +338,37 @@ 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 receiver pid, we have to
% register with controller to receive {copier_done, Node} if our controller handles
% {mnesia_down, 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_receiver, self(), node(SenderPid)})
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.
Expand All @@ -364,8 +383,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),
Expand All @@ -390,25 +410,28 @@ wait_on_load_complete(Pid) ->
end.

do_init_table(Tab,Storage,Cs,SenderPid,
TabSize,DetsInfo,OrigTabRec,Init) ->
TabSize,DetsInfo,OrigTabRec,Init) ->
case create_table(Tab, TabSize, Storage, Cs) of
{Storage,Tab} ->
%% Debug info
Node = node(SenderPid),
put(mnesia_table_receiver, {Tab, Node, SenderPid}),
mnesia_tm:block_tab(Tab),
case init_table(Tab,Storage,Init,DetsInfo,SenderPid) of
ok ->
tab_receiver(Node,Tab,Storage,Cs,OrigTabRec);
Reason ->
Msg = "[d]ets:init table failed",
verbose("~ts: ~tp: ~tp~n", [Msg, Tab, Reason]),
{Storage,Tab} ->
%% Debug info
Node = node(SenderPid),
put(mnesia_table_receiver, {Tab, Node, SenderPid}),
mnesia_tm:block_tab(Tab),
case init_table(Tab,Storage,Init,DetsInfo,SenderPid) of
ok ->
tab_receiver(Node,Tab,Storage,Cs,OrigTabRec);
Reason ->
%% If there was other receiver and it was aborted by {copier_done, Node},
%% we will also have {copier_done, Node} in our message queue
?flush_msg({copier_done, Node}),
Msg = "[d]ets:init table failed",
verbose("~ts: ~tp: ~tp~n", [Msg, Tab, Reason]),
SenderPid ! {copier_done, node()},
down(Tab, Storage)
end;
Error ->
down(Tab, Storage)
end;
Error ->
SenderPid ! {copier_done, node()},
Error
Error
end.

create_table(Tab, TabSize, Storage, Cs) ->
Expand Down
Loading
Loading