diff --git a/lib/mnesia/src/mnesia_controller.erl b/lib/mnesia/src/mnesia_controller.erl index c44c4122cbef..3873159c25c4 100644 --- a/lib/mnesia/src/mnesia_controller.erl +++ b/lib/mnesia/src/mnesia_controller.erl @@ -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.. @@ -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 -> @@ -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 @@ -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); @@ -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); diff --git a/lib/mnesia/src/mnesia_loader.erl b/lib/mnesia/src/mnesia_loader.erl index 2fe1a0d85578..8127ef43a02e 100644 --- a/lib/mnesia/src/mnesia_loader.erl +++ b/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,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. @@ -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), @@ -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) -> diff --git a/lib/mnesia/test/Makefile b/lib/mnesia/test/Makefile index 64921c495216..37b5ec1dd634 100644 --- a/lib/mnesia/test/Makefile +++ b/lib/mnesia/test/Makefile @@ -57,7 +57,8 @@ MODULES= \ ext_test \ ext_test_server \ mnesia_index_plugin_test \ - mnesia_external_backend_test + mnesia_external_backend_test \ + gen_tcp_blocking_dist DocExamplesDir := ../doc/src/ @@ -78,7 +79,7 @@ ExampleModules = \ ERL_FILES= $(MODULES:%=%.erl) $(DocExampleModules:%=$(DocExamplesDir)/%.erl) $(ExampleModules:%=$(ExamplesDir)/%.erl) -HRL_FILES= mnesia_test_lib.hrl ext_test_server.hrl $(DocExamplesHrl:%=$(DocExamplesDir)/%) +HRL_FILES= mnesia_test_lib.hrl ext_test_server.hrl gen_tcp_blocking_dist.hrl $(DocExamplesHrl:%=$(DocExamplesDir)/%) TARGET_FILES= \ $(MODULES:%=$(EBIN)/%.$(EMULATOR)) $(DocExampleModules:%=$(EBIN)/%.$(EMULATOR)) $(ExampleModules:%=$(EBIN)/%.$(EMULATOR)) diff --git a/lib/mnesia/test/gen_tcp_blocking_dist.erl b/lib/mnesia/test/gen_tcp_blocking_dist.erl new file mode 100644 index 000000000000..26f596bb895b --- /dev/null +++ b/lib/mnesia/test/gen_tcp_blocking_dist.erl @@ -0,0 +1,928 @@ +%% +%% %CopyrightBegin% +%% +%% SPDX-License-Identifier: Apache-2.0 +%% +%% Copyright Ericsson AB 2026. All Rights Reserved. +%% +%% Licensed under the Apache License, Version 2.0 (the "License"); +%% you may not use this file except in compliance with the License. +%% You may obtain a copy of the License at +%% +%% http://www.apache.org/licenses/LICENSE-2.0 +%% +%% Unless required by applicable law or agreed to in writing, software +%% distributed under the License is distributed on an "AS IS" BASIS, +%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +%% See the License for the specific language governing permissions and +%% limitations under the License. +%% +%% %CopyrightEnd% +%% +-module(gen_tcp_blocking_dist). + +%% +%% This is a modification of gen_tcp_dist that adds possibility to simulate +%% network outage between nodes. +%% This currently behaves in similar way to inet_tcp_proxy: https://github.com/rabbitmq/inet_tcp_proxy +%% When on Node1 we block Node2, we: +%% - stop consuming distribution messages (we allow them to be queued, but not sent, they will be +%% sent in bulk when node is unblocked later) +%% - stop sending ticks +%% - raise exit exception from setup/5 if node we want to connect to is blocked +%% - raise exit exception after handshake with blocked node completes +%% + +%% Public interface for blocking network communication +-export([block_peer/2, unblock_peer/2, + block_pair/2, unblock_pair/2, + unblock_pair_ignore_node_down/2, has_network_blocker/0, + set_peer_ref/2, get_peer_ref/1, get_peer_ref/2]). + +%% gen_tcp_dist interface +-export([listen/1, accept/1, accept_connection/5, + setup/5, close/1, select/1, is_node_name/1, + address/0]). + +%% Optional +-export([setopts/2, getopts/2]). + +%% internal exports + +-export([dist_cntrlr_setup/1, dist_cntrlr_input_setup/4, + dist_cntrlr_tick_handler/2]). + +-export([accept_loop/2,do_accept/6,do_setup/6]). + +%% Internal exports for rpc calls +-export([block_internal/1, unblock_internal/1]). + +-import(error_logger,[error_msg/2]). + +-include("gen_tcp_blocking_dist.hrl"). + +-include_lib("kernel/include/net_address.hrl"). + +-include_lib("kernel/include/dist.hrl"). +-include_lib("kernel/include/dist_util.hrl"). + +%% ------------------------------------------------------------ +%% Block outgoing communication (and ticks) from node +%% 'From' to node 'To'. This is a one way block. +%% ------------------------------------------------------------ +block_peer(From, To) when From =:= node() -> + ok = block_internal(To); +block_peer(From, To) -> + ok = remote_call(From, ?MODULE, block_internal, [To]). + +%% ------------------------------------------------------------ +%% Unblock outgoing communication (and ticks) from node +%% 'From' to node 'To'. This is a one way unblock. +%% ------------------------------------------------------------ +unblock_peer(From, To) when From =:= node() -> + ok = unblock_internal(To); +unblock_peer(From, To) -> + ok = remote_call(From, ?MODULE, unblock_internal, [To]). + +%% ------------------------------------------------------------ +%% Block outgoing communication (and ticks) between 2 nodes. +%% This is a two way block. +%% ------------------------------------------------------------ +block_pair(Node1, Node2) when Node1 =:= node() -> + ok = block_peer(Node2, Node1), + %% Block local outgoing communication last + ok = block_peer(Node1, Node2); +block_pair(Node1, Node2) -> + ok = block_peer(Node1, Node2), + ok = block_peer(Node2, Node1). + +%% ------------------------------------------------------------ +%% Unblock outgoing communication (and ticks) between 2 nodes. +%% This is a two way unblock. +%% ------------------------------------------------------------ +unblock_pair(Node1, Node2) when Node2 =:= node() -> + %% Unblock local outgoing communication first + ok = unblock_peer(Node2, Node1), + ok = unblock_peer(Node1, Node2); +unblock_pair(Node1, Node2) -> + ok = unblock_peer(Node1, Node2), + ok = unblock_peer(Node2, Node1). + +%% ------------------------------------------------------------ +%% Unblock outgoing communication (and ticks) between 2 nodes. +%% This is a two way unblock. +%% ------------------------------------------------------------ +unblock_pair_ignore_node_down(Node1, Node2) when Node1 =:= node() -> + %% Unblock local outgoing communication first + ok = unblock_peer(Node1, Node2), + try unblock_peer(Node2, Node1) catch error : {badmatch, {badrpc, nodedown}} -> ok end; +unblock_pair_ignore_node_down(Node1, Node2) when Node2 =:= node() -> + %% Unblock local outgoing communication first + ok = unblock_peer(Node2, Node1), + try unblock_peer(Node1, Node2) catch error : {badmatch, {badrpc, nodedown}} -> ok end; +unblock_pair_ignore_node_down(Node1, Node2) -> + try unblock_peer(Node1, Node2) catch error : {badmatch, {badrpc, nodedown}} -> ok end, + try unblock_peer(Node2, Node1) catch error : {badmatch, {badrpc, nodedown}} -> ok end. + +%% ------------------------------------------------------------ +%% Internal stuff for network blocking +%% ------------------------------------------------------------ +block_internal(Node) -> + persistent_term:put({?MODULE, block, Node}, true). + +unblock_internal(Node) -> + case persistent_term:erase({?MODULE, block, Node}) of + true -> + case lists:keysearch(Node, 1, erlang:system_info(dist_ctrl)) of + {value, {Node, Pid}} -> + %% Notify output handler to flush data + Pid ! allow, + ok; + _ -> + ok + end; + false -> + ok + end. + +is_blocked(Node) -> + persistent_term:get({?MODULE, block, Node}, false). + +remote_call(Node, M, F, A) -> + case get_peer_ref(Node, undefined) of + undefined -> + rpc:call(Node, M, F, A); + Peer -> + peer:call(Peer, M, F, A) + end. + +%% ------------------------------------------------------------ +%% Check if network blocker is enabled on current node +%% ------------------------------------------------------------ +has_network_blocker() -> + Args = init:get_arguments(), + lists:all(fun(Elem) -> lists:member(Elem, Args) end, ?NETWORK_BLOCKER_DIST_OPTS_PARSED). + +set_peer_ref(Node, Ref) -> + persistent_term:put({?MODULE, peer, Node}, Ref). + +get_peer_ref(Node) -> + persistent_term:get({?MODULE, peer, Node}). + +get_peer_ref(Node, Default) -> + persistent_term:get({?MODULE, peer, Node}, Default). + +%% gen_tcp_dist interface starts here: +%% ------------------------------------------------------------ +%% Select this protocol based on node name +%% select(Node) => Bool +%% ------------------------------------------------------------ + +select(Node) -> + case split_node(atom_to_list(Node), $@, []) of + [_, Host] -> + case inet:getaddr(Host, inet) of + {ok,_} -> true; + _ -> false + end; + _ -> false + end. + +%% ------------------------------------------------------------ +%% Get the address family that this distribution uses +%% ------------------------------------------------------------ +address() -> + get_tcp_address(). + +%% ------------------------------------------------------------ +%% Create the listen socket, i.e. the port that this erlang +%% node is accessible through. +%% ------------------------------------------------------------ + +listen(Name) -> + case do_listen([binary, {active, false}, {packet,2}, {reuseaddr, true}]) of + {ok, Socket} -> + TcpAddress = get_tcp_address(Socket), + {_,Port} = TcpAddress#net_address.address, + ErlEpmd = net_kernel:epmd_module(), + case ErlEpmd:register_node(Name, Port) of + {ok, Creation} -> + {ok, {Socket, TcpAddress, Creation}}; + Error -> + Error + end; + Error -> + Error + end. + +do_listen(Options) -> + {First,Last} = case application:get_env(kernel,inet_dist_listen_min) of + {ok,N} when is_integer(N) -> + case application:get_env(kernel, + inet_dist_listen_max) of + {ok,M} when is_integer(M) -> + {N,M}; + _ -> + {N,N} + end; + _ -> + {0,0} + end, + do_listen(First, Last, listen_options([{backlog,128}|Options])). + +do_listen(First,Last,_) when First > Last -> + {error,eaddrinuse}; +do_listen(First,Last,Options) -> + case gen_tcp:listen(First, Options) of + {error, eaddrinuse} -> + do_listen(First+1,Last,Options); + Other -> + Other + end. + +listen_options(Opts0) -> + Opts1 = + case application:get_env(kernel, inet_dist_use_interface) of + {ok, Ip} -> + [{ip, Ip} | Opts0]; + _ -> + Opts0 + end, + case application:get_env(kernel, inet_dist_listen_options) of + {ok,ListenOpts} -> + ListenOpts ++ Opts1; + _ -> + Opts1 + end. + + +%% ------------------------------------------------------------ +%% Accepts new connection attempts from other Erlang nodes. +%% ------------------------------------------------------------ + +accept(Listen) -> + spawn_opt(?MODULE, accept_loop, [self(), Listen], [link, {priority, max}]). + +accept_loop(Kernel, Listen) -> + ?trace("~p~n",[{?MODULE, accept_loop, self()}]), + case gen_tcp:accept(Listen) of + {ok, Socket} -> + DistCtrl = spawn_dist_cntrlr(Socket), + ?trace("~p~n",[{?MODULE, accept_loop, accepted, Socket, DistCtrl, self()}]), + flush_controller(DistCtrl, Socket), + gen_tcp:controlling_process(Socket, DistCtrl), + flush_controller(DistCtrl, Socket), + Kernel ! {accept,self(),DistCtrl,inet,tcp}, + receive + {Kernel, controller, Pid} -> + call_ctrlr(DistCtrl, {supervisor, Pid}), + Pid ! {self(), controller}; + {Kernel, unsupported_protocol} -> + exit(unsupported_protocol) + end, + accept_loop(Kernel, Listen); + Error -> + exit(Error) + end. + +flush_controller(Pid, Socket) -> + receive + {tcp, Socket, Data} -> + Pid ! {tcp, Socket, Data}, + flush_controller(Pid, Socket); + {tcp_closed, Socket} -> + Pid ! {tcp_closed, Socket}, + flush_controller(Pid, Socket) + after 0 -> + ok + end. + +%% ------------------------------------------------------------ +%% Accepts a new connection attempt from another Erlang node. +%% Performs the handshake with the other side. +%% ------------------------------------------------------------ + +accept_connection(AcceptPid, DistCtrl, MyNode, Allowed, SetupTime) -> + spawn_opt(?MODULE, do_accept, + [self(), AcceptPid, DistCtrl, MyNode, Allowed, SetupTime], + dist_util:net_ticker_spawn_options()). + +do_accept(Kernel, AcceptPid, DistCtrl, MyNode, Allowed, SetupTime) -> + ?trace("~p~n",[{?MODULE, do_accept, self(), MyNode}]), + receive + {AcceptPid, controller} -> + Timer = dist_util:start_timer(SetupTime), + case check_ip(DistCtrl) of + true -> + HSData0 = hs_data_common(DistCtrl), + HSData = HSData0#hs_data{kernel_pid = Kernel, + this_node = MyNode, + socket = DistCtrl, + timer = Timer, + this_flags = 0, + allowed = Allowed}, + dist_util:handshake_other_started(HSData); + {false,IP} -> + error_msg("** Connection attempt from " + "disallowed IP ~w ** ~n", [IP]), + ?shutdown(no_node) + end + end. + +%% we may not always want the nodelay behaviour +%% for performance reasons + +nodelay() -> + case application:get_env(kernel, dist_nodelay) of + undefined -> + {nodelay, true}; + {ok, true} -> + {nodelay, true}; + {ok, false} -> + {nodelay, false}; + _ -> + {nodelay, true} + end. + +%% ------------------------------------------------------------ +%% Setup a new connection to another Erlang node. +%% Performs the handshake with the other side. +%% ------------------------------------------------------------ + +setup(Node, Type, MyNode, LongOrShortNames,SetupTime) -> + spawn_opt(?MODULE, do_setup, + [self(), Node, Type, MyNode, LongOrShortNames, SetupTime], + dist_util:net_ticker_spawn_options()). + +do_setup(Kernel, Node, Type, MyNode, LongOrShortNames, SetupTime) -> + ?trace("~p~n",[{?MODULE, do_setup, self(), Node}]), + case is_blocked(Node) of + true -> + ?trace("(~p): Node: ~p blocked, aborting setup/5~n", [node(), Node]), + ?shutdown(Node); + false -> + [Name, Address] = splitnode(Node, LongOrShortNames), + case inet:getaddr(Address, inet) of + {ok, Ip} -> + Timer = dist_util:start_timer(SetupTime), + ErlEpmd = net_kernel:epmd_module(), + case ErlEpmd:port_please(Name, Ip) of + {port, TcpPort, Version} -> + ?trace("port_please(~p) -> version ~p~n", + [Node,Version]), + dist_util:reset_timer(Timer), + case + gen_tcp:connect( + Ip, TcpPort, + connect_options([binary, {active, false}, {packet, 2}])) + of + {ok, Socket} -> + DistCtrl = spawn_dist_cntrlr(Socket), + call_ctrlr(DistCtrl, {supervisor, self()}), + flush_controller(DistCtrl, Socket), + gen_tcp:controlling_process(Socket, DistCtrl), + flush_controller(DistCtrl, Socket), + HSData0 = hs_data_common(DistCtrl), + HSData = HSData0#hs_data{kernel_pid = Kernel, + other_node = Node, + this_node = MyNode, + socket = DistCtrl, + timer = Timer, + this_flags = 0, + other_version = Version, + request_type = Type}, + dist_util:handshake_we_started(HSData); + _ -> + %% Other Node may have closed since + %% port_please ! + ?trace("other node (~p) " + "closed since port_please.~n", + [Node]), + ?shutdown(Node) + end; + _ -> + ?trace("port_please (~p) " + "failed.~n", [Node]), + ?shutdown(Node) + end; + _Other -> + ?trace("inet_getaddr(~p) " + "failed (~p).~n", [Node,_Other]), + ?shutdown(Node) + end + end. + +connect_options(Opts) -> + case application:get_env(kernel, inet_dist_connect_options) of + {ok,ConnectOpts} -> + ConnectOpts ++ Opts; + _ -> + Opts + end. + +%% +%% Close a socket. +%% +close(Listen) -> + gen_tcp:close(Listen). + + +%% If Node is illegal terminate the connection setup!! +splitnode(Node, LongOrShortNames) -> + case split_node(atom_to_list(Node), $@, []) of + [Name|Tail] when Tail =/= [] -> + Host = lists:append(Tail), + case split_node(Host, $., []) of + [_] when LongOrShortNames =:= longnames -> + case inet:parse_address(Host) of + {ok, _} -> + [Name, Host]; + _ -> + error_msg("** System running to use " + "fully qualified " + "hostnames **~n" + "** Hostname ~ts is illegal **~n", + [Host]), + ?shutdown(Node) + end; + L when length(L) > 1, LongOrShortNames =:= shortnames -> + error_msg("** System NOT running to use fully qualified " + "hostnames **~n" + "** Hostname ~ts is illegal **~n", + [Host]), + ?shutdown(Node); + _ -> + [Name, Host] + end; + [_] -> + error_msg("** Nodename ~p illegal, no '@' character **~n", + [Node]), + ?shutdown(Node); + _ -> + error_msg("** Nodename ~p illegal **~n", [Node]), + ?shutdown(Node) + end. + +split_node([Chr|T], Chr, Ack) -> [lists:reverse(Ack)|split_node(T, Chr, [])]; +split_node([H|T], Chr, Ack) -> split_node(T, Chr, [H|Ack]); +split_node([], _, Ack) -> [lists:reverse(Ack)]. + +%% ------------------------------------------------------------ +%% Fetch local information about a Socket. +%% ------------------------------------------------------------ +get_tcp_address(Socket) -> + {ok, Address} = inet:sockname(Socket), + NetAddr = get_tcp_address(), + NetAddr#net_address{address = Address}. + +get_tcp_address() -> + {ok, Host} = inet:gethostname(), + #net_address { + host = Host, + protocol = tcp, + family = inet + }. + +%% ------------------------------------------------------------ +%% Do only accept new connection attempts from nodes at our +%% own LAN, if the check_ip environment parameter is true. +%% ------------------------------------------------------------ +check_ip(DistCtrl) -> + case application:get_env(check_ip) of + {ok, true} -> + case get_ifs(DistCtrl) of + {ok, IFs, IP} -> + check_ip(IFs, IP); + _ -> + ?shutdown(no_node) + end; + _ -> + true + end. + +get_ifs(DistCtrl) -> + Socket = call_ctrlr(DistCtrl, socket), + case inet:peername(Socket) of + {ok, {IP, _}} -> + case inet:getif(Socket) of + {ok, IFs} -> {ok, IFs, IP}; + Error -> Error + end; + Error -> + Error + end. + +check_ip([{OwnIP, _, Netmask}|IFs], PeerIP) -> + case {inet_tcp:mask(Netmask, PeerIP), inet_tcp:mask(Netmask, OwnIP)} of + {M, M} -> true; + _ -> check_ip(IFs, PeerIP) + end; +check_ip([], PeerIP) -> + {false, PeerIP}. + +is_node_name(Node) when is_atom(Node) -> + case split_node(atom_to_list(Node), $@, []) of + [_, _Host] -> true; + _ -> false + end; +is_node_name(_Node) -> + false. + +hs_data_common(DistCtrl) -> + TickHandler = call_ctrlr(DistCtrl, tick_handler), + Socket = call_ctrlr(DistCtrl, socket), + RejectFlags = case init:get_argument(gen_tcp_dist_reject_flags) of + {ok,[[Flags]]} -> list_to_integer(Flags); + _ -> #hs_data{}#hs_data.reject_flags + end, + #hs_data{f_send = send_fun(), + f_recv = recv_fun(), + f_setopts_pre_nodeup = setopts_pre_nodeup_fun(), + f_setopts_post_nodeup = setopts_post_nodeup_fun(), + f_getll = getll_fun(), + f_handshake_complete = handshake_complete_fun(), + f_address = address_fun(), + mf_setopts = setopts_fun(DistCtrl, Socket), + mf_getopts = getopts_fun(DistCtrl, Socket), + mf_getstat = getstat_fun(DistCtrl, Socket), + mf_tick = tick_fun(DistCtrl, TickHandler), + reject_flags = RejectFlags}. + +%%% ------------------------------------------------------------ +%%% Distribution controller processes +%%% ------------------------------------------------------------ + +%% +%% There will be five parties working together when the +%% connection is up: +%% - The gen_tcp socket. Providing a tcp/ip connection +%% to the other node. +%% - The output handler. It will dispatch all outgoing +%% traffic from the VM to the gen_tcp socket. This +%% process is registered as distribution controller +%% for this channel with the VM. +%% - The input handler. It will dispatch all incoming +%% traffic from the gen_tcp socket to the VM. This +%% process is also the socket owner and receives +%% incoming traffic using active-N. +%% - The tick handler. Dispatches asynchronous tick +%% requests to the socket. It executes on max priority +%% since it is important to get ticks through to the +%% other end. +%% - The channel supervisor (provided by dist_util). It +%% monitors traffic. Issue tick requests to the tick +%% handler when no outgoing traffic is seen and bring +%% the connection down if no incoming traffic is seen. +%% This process also executes on max priority. +%% +%% These parties are linked together so should one +%% of them fail, all of them are terminated and the +%% connection is taken down. +%% + +%% In order to avoid issues with lingering signal binaries +%% we enable off-heap message queue data as well as fullsweep +%% after 0. The fullsweeps will be cheap since we have more +%% or less no live data. +-define(DIST_CNTRL_COMMON_SPAWN_OPTS, + [{message_queue_data, off_heap}, + {fullsweep_after, 0}]). + +tick_fun(DistCtrl, TickHandler) -> + fun (Ctrl) when Ctrl == DistCtrl -> + TickHandler ! tick + end. + +getstat_fun(DistCtrl, Socket) -> + fun (Ctrl) when Ctrl == DistCtrl -> + case inet:getstat(Socket, [recv_cnt, send_cnt, send_pend]) of + {ok, Stat} -> + split_stat(Stat,0,0,0); + Error -> + Error + end + end. + +split_stat([{recv_cnt, R}|Stat], _, W, P) -> + split_stat(Stat, R, W, P); +split_stat([{send_cnt, W}|Stat], R, _, P) -> + split_stat(Stat, R, W, P); +split_stat([{send_pend, P}|Stat], R, W, _) -> + split_stat(Stat, R, W, P); +split_stat([], R, W, P) -> + {ok, R, W, P}. + +setopts_fun(DistCtrl, Socket) -> + fun (Ctrl, Opts) when Ctrl == DistCtrl -> + setopts(Socket, Opts) + end. + +getopts_fun(DistCtrl, Socket) -> + fun (Ctrl, Opts) when Ctrl == DistCtrl -> + getopts(Socket, Opts) + end. + +setopts(S, Opts) -> + case [Opt || {K,_}=Opt <- Opts, + K =:= active orelse K =:= deliver orelse K =:= packet] of + [] -> inet:setopts(S,Opts); + Opts1 -> {error, {badopts,Opts1}} + end. + +getopts(S, Opts) -> + inet:getopts(S, Opts). + +send_fun() -> + fun (Ctrlr, Packet) -> + call_ctrlr(Ctrlr, {send, Packet}) + end. + +recv_fun() -> + fun (Ctrlr, Length, Timeout) -> + case call_ctrlr(Ctrlr, {recv, Length, Timeout}) of + {ok, Bin} when is_binary(Bin) -> + {ok, binary_to_list(Bin)}; + Other -> + Other + end + end. + +getll_fun() -> + fun (Ctrlr) -> + call_ctrlr(Ctrlr, getll) + end. + +address_fun() -> + fun (Ctrlr, Node) -> + case call_ctrlr(Ctrlr, {address, Node}) of + {error, no_node} -> %% No '@' or more than one '@' in node name. + ?shutdown(no_node); + Res -> + Res + end + end. + +setopts_pre_nodeup_fun() -> + fun (Ctrlr) -> + call_ctrlr(Ctrlr, pre_nodeup) + end. + +setopts_post_nodeup_fun() -> + fun (Ctrlr) -> + call_ctrlr(Ctrlr, post_nodeup) + end. + +handshake_complete_fun() -> + fun (Ctrlr, Node, DHandle) -> + call_ctrlr(Ctrlr, {handshake_complete, Node, DHandle}) + end. + +call_ctrlr(Ctrlr, Msg) -> + Ref = erlang:monitor(process, Ctrlr), + Ctrlr ! {Ref, self(), Msg}, + receive + {Ref, Res} -> + erlang:demonitor(Ref, [flush]), + Res; + {'DOWN', Ref, process, Ctrlr, Reason} -> + exit({dist_controller_exit, Reason}) + end. + +%% +%% The tick handler process writes a tick to the +%% socket when it receives a 'tick' message from +%% the connection supervisor. +%% +%% We are not allowed to block the connection +%% superviser when writing a tick and we also want +%% the tick to go through even during a heavily +%% loaded system. gen_tcp does not have a +%% non-blocking send operation exposed in its API +%% and we don't want to run the distribution +%% controller under high priority. Therefore this +%% separate process with max prio that dispatches +%% ticks. +%% +dist_cntrlr_tick_handler(Socket, Node0) -> + receive + tick -> + case is_blocked(Node0) of + true -> + %% Don't send tick... + ok; + false -> + %% May block due to busy port... + sock_send(Socket, "") + end; + {set_node, Node} -> + dist_cntrlr_tick_handler(Socket, Node); + _ -> + ok + end, + dist_cntrlr_tick_handler(Socket, Node0). + +spawn_dist_cntrlr(Socket) -> + spawn_opt(?MODULE, dist_cntrlr_setup, [Socket], + [{priority, max}] ++ ?DIST_CNTRL_COMMON_SPAWN_OPTS). + +dist_cntrlr_setup(Socket) -> + TickHandler = spawn_opt(?MODULE, dist_cntrlr_tick_handler, + [Socket, undefined], + [link, {priority, max}] + ++ ?DIST_CNTRL_COMMON_SPAWN_OPTS), + dist_cntrlr_setup_loop(Socket, TickHandler, undefined). + +%% +%% During the handshake phase we loop in dist_cntrlr_setup(). +%% When the connection is up we spawn an input handler and +%% continue as output handler. +%% +dist_cntrlr_setup_loop(Socket, TickHandler, Sup) -> + receive + {tcp_closed, Socket} -> + exit(connection_closed); + + {Ref, From, {supervisor, Pid}} -> + Res = link(Pid), + From ! {Ref, Res}, + dist_cntrlr_setup_loop(Socket, TickHandler, Pid); + + {Ref, From, tick_handler} -> + From ! {Ref, TickHandler}, + dist_cntrlr_setup_loop(Socket, TickHandler, Sup); + + {Ref, From, socket} -> + From ! {Ref, Socket}, + dist_cntrlr_setup_loop(Socket, TickHandler, Sup); + + {Ref, From, {send, Packet}} -> + Res = gen_tcp:send(Socket, Packet), + From ! {Ref, Res}, + dist_cntrlr_setup_loop(Socket, TickHandler, Sup); + + {Ref, From, {recv, Length, Timeout}} -> + Res = gen_tcp:recv(Socket, Length, Timeout), + From ! {Ref, Res}, + dist_cntrlr_setup_loop(Socket, TickHandler, Sup); + + {Ref, From, getll} -> + From ! {Ref, {ok, self()}}, + dist_cntrlr_setup_loop(Socket, TickHandler, Sup); + + {Ref, From, {address, Node}} -> + Res = case inet:peername(Socket) of + {ok, Address} -> + case split_node(atom_to_list(Node), $@, []) of + [_,Host] -> + #net_address{address=Address,host=Host, + protocol=tcp, family=inet}; + _ -> + {error, no_node} + end + end, + From ! {Ref, Res}, + dist_cntrlr_setup_loop(Socket, TickHandler, Sup); + + {Ref, From, pre_nodeup} -> + Res = inet:setopts(Socket, + [{active, false}, + {packet, 4}, + nodelay()]), + From ! {Ref, Res}, + dist_cntrlr_setup_loop(Socket, TickHandler, Sup); + + {Ref, From, post_nodeup} -> + Res = inet:setopts(Socket, + [{active, false}, + {packet, 4}, + nodelay()]), + From ! {Ref, Res}, + dist_cntrlr_setup_loop(Socket, TickHandler, Sup); + + {Ref, From, {handshake_complete, Node, DHandle}} -> + case is_blocked(Node) of + true -> + ?trace("(~p): Node: ~p blocked, aborting after handshake_complete~n", [node(), Node]), + exit({shutdown, blocked}); + false -> + From ! {Ref, ok}, + TickHandler ! {set_node, Node}, + %% Handshake complete! Begin dispatching traffic... + + %% We use separate process for dispatching input. This + %% is not necessary, but it enables parallel execution + %% of independent work loads at the same time as it + %% simplifies the the implementation... + InputHandler = spawn_opt(?MODULE, dist_cntrlr_input_setup, + [DHandle, Socket, Node, Sup], + [link] ++ ?DIST_CNTRL_COMMON_SPAWN_OPTS), + + flush_controller(InputHandler, Socket), + gen_tcp:controlling_process(Socket, InputHandler), + flush_controller(InputHandler, Socket), + + ok = erlang:dist_ctrl_input_handler(DHandle, InputHandler), + + InputHandler ! DHandle, + + %% From now on we execute on normal priority + process_flag(priority, normal), + erlang:dist_ctrl_get_data_notification(DHandle), + dist_cntrlr_output_loop(DHandle, Socket, Node) + end + end. + +%% We use active 10 for good throughput while still +%% maintaining back-pressure if the input controller +%% isn't able to handle all incoming messages... +-define(ACTIVE_INPUT, 10). + +dist_cntrlr_input_setup(DHandle, Socket, Node, Sup) -> + link(Sup), + %% Ensure we don't try to put data before we are registered + %% as input handler... + receive + DHandle -> + dist_cntrlr_input_loop(DHandle, Socket, Node, 0) + end. + +dist_cntrlr_input_loop(DHandle, Socket, Node, N) when N =< ?ACTIVE_INPUT/2 -> + inet:setopts(Socket, [{active, ?ACTIVE_INPUT - N}]), + dist_cntrlr_input_loop(DHandle, Socket, Node, ?ACTIVE_INPUT); +dist_cntrlr_input_loop(DHandle, Socket, Node, N) -> + receive + {tcp_closed, Socket} -> + %% Connection to remote node terminated... + exit(connection_closed); + + {tcp, Socket, Data} -> + %% Incoming data from remote node... + try erlang:dist_ctrl_put_data(DHandle, Data) + catch _ : _ -> death_row() + end, + dist_cntrlr_input_loop(DHandle, Socket, Node, N - 1); + + _ -> + %% Ignore... + dist_cntrlr_input_loop(DHandle, Socket, Node, N) + end. + +dist_cntrlr_send_data(DHandle, Socket) -> + case erlang:dist_ctrl_get_data(DHandle) of + none -> + erlang:dist_ctrl_get_data_notification(DHandle); + Data -> + sock_send(Socket, Data), + dist_cntrlr_send_data(DHandle, Socket) + end. + + +dist_cntrlr_output_loop(DHandle, Socket, Node) -> + receive + Msg when Msg =:= dist_data; Msg =:= allow -> + %% Outgoing data from this node... + case is_blocked(Node) of + true -> + %% Don't flush message from buffer if peer is blocked + dist_cntrlr_output_loop(DHandle, Socket, Node); + false -> + try dist_cntrlr_send_data(DHandle, Socket) + catch _ : _ -> death_row() + end, + dist_cntrlr_output_loop(DHandle, Socket, Node) + end; + + _ -> + %% Drop garbage message... + dist_cntrlr_output_loop(DHandle, Socket, Node) + + end. + +sock_send(Socket, Data) -> + try gen_tcp:send(Socket, Data) of + ok -> ok; + {error, Reason} -> death_row({send_error, Reason}) + catch + Type : Reason -> death_row({send_error, {Type, Reason}}) + end. + +death_row() -> + death_row(connection_closed). + +death_row(normal) -> + %% We do not want to exit with normal + %% exit reason since it won't bring down + %% linked processes... + death_row(); +death_row(Reason) -> + %% When the connection is on its way down operations + %% begin to fail. We catch the failures and call + %% this function waiting for termination. We should + %% be terminated by one of our links to the other + %% involved parties that began bringing the + %% connection down. By waiting for termination we + %% avoid altering the exit reason for the connection + %% teardown. We however limit the wait to 5 seconds + %% and bring down the connection ourselves if not + %% terminated... + receive after 5000 -> exit(Reason) end. diff --git a/lib/mnesia/test/gen_tcp_blocking_dist.hrl b/lib/mnesia/test/gen_tcp_blocking_dist.hrl new file mode 100644 index 000000000000..4343451d128a --- /dev/null +++ b/lib/mnesia/test/gen_tcp_blocking_dist.hrl @@ -0,0 +1,26 @@ +%% +%% %CopyrightBegin% +%% +%% SPDX-License-Identifier: Apache-2.0 +%% +%% Copyright Ericsson AB 2026. All Rights Reserved. +%% +%% Licensed under the Apache License, Version 2.0 (the "License"); +%% you may not use this file except in compliance with the License. +%% You may obtain a copy of the License at +%% +%% http://www.apache.org/licenses/LICENSE-2.0 +%% +%% Unless required by applicable law or agreed to in writing, software +%% distributed under the License is distributed on an "AS IS" BASIS, +%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +%% See the License for the specific language governing permissions and +%% limitations under the License. +%% +%% %CopyrightEnd% +%% + +%% + +-define(NETWORK_BLOCKER_DIST_OPTS_PARSED, [{proto_dist, ["gen_tcp_blocking"]}]). +-define(NETWORK_BLOCKER_DIST_OPTS, ["-proto_dist gen_tcp_blocking"]). diff --git a/lib/mnesia/test/mnesia_durability_test.erl b/lib/mnesia/test/mnesia_durability_test.erl index e03225388f6f..7c3f5987f8da 100644 --- a/lib/mnesia/test/mnesia_durability_test.erl +++ b/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. + %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% diff --git a/lib/mnesia/test/mnesia_external_backend_test.erl b/lib/mnesia/test/mnesia_external_backend_test.erl index 17a3719a355a..0789f8ef3494 100644 --- a/lib/mnesia/test/mnesia_external_backend_test.erl +++ b/lib/mnesia/test/mnesia_external_backend_test.erl @@ -47,6 +47,8 @@ all() -> [ groups() -> []. +init_per_testcase(backup_and_restore_should_work_with_external_backend, _Conf) -> + {skip, "Not implemented"}; init_per_testcase(Func, Conf) -> file:delete("bup0.BUP"), file:delete("bup1.BUP"), diff --git a/lib/mnesia/test/mnesia_test_lib.erl b/lib/mnesia/test/mnesia_test_lib.erl index 220dce759b43..3db0e68ebd56 100644 --- a/lib/mnesia/test/mnesia_test_lib.erl +++ b/lib/mnesia/test/mnesia_test_lib.erl @@ -135,10 +135,20 @@ init_per_testcase/2, end_per_testcase/2, kill_tc/2, - get_ext_test_server_name/0 + get_ext_test_server_name/0, + set_peer_ref/2, + get_peer_ref/1, + get_peer_ref/2, + has_network_blocker/0, + skip_if_no_network_blocker/1, + block_peer/2, + unblock_peer/2, + block_pair/2, + unblock_pair/2 ]). -include("mnesia_test_lib.hrl"). +-include("gen_tcp_blocking_dist.hrl"). -compile([{nowarn_possibly_unsafe_function, {erlang, list_to_atom, 1}}]). @@ -284,9 +294,15 @@ node_start_link(Host, Name) -> node_start_link(Host, Name, Retries) -> Debug = atom_to_list(mnesia:system_info(debug)), - Args = ["-mnesia", "debug", Debug, - "-pa", filename:dirname(code:which(?MODULE)), - "-pa", filename:dirname(code:which(mnesia))], + Args0 = ["-mnesia", "debug", Debug, + "-pa", filename:dirname(code:which(?MODULE)), + "-pa", filename:dirname(code:which(mnesia))], + Args = case has_network_blocker() of + true -> + Args0 ++ ?NETWORK_BLOCKER_DIST_OPTS; + false -> + Args0 + end, case starter(Host, Name, Args) of {ok, NewNode} -> ?match(pong, net_adm:ping(NewNode)), @@ -308,7 +324,8 @@ node_start_link(Host, Name, Retries) -> end. starter(Host, Name, Args) -> - {ok, _, Node} = peer:start(#{host => Host, name => Name, args => Args}), + {ok, Peer, Node} = peer:start(#{host => Host, name => Name, args => Args, connection => 0}), + ok = set_peer_ref(Node, Peer), {ok, Node}. node_sup() -> @@ -666,6 +683,18 @@ prepare_test_case(Actions, N, Config, File, Line) -> NodeList3 = append_unique(NodeList1, NodeList2), This = node(), All = [This | lists:delete(This, NodeList3)], + case has_network_blocker() of + true -> + Pairs0 = [{A, B} || A <- All, B <- All, A < B], + {Local, Remote} = lists:partition(fun({A, B}) -> + A =:= node() orelse B =:= node() + end, Pairs0), + Pairs = Local ++ Remote, + %% Unlock local pairs first, so it works if we don't have peer refs + [gen_tcp_blocking_dist:unblock_pair_ignore_node_down(Node1, Node2) || {Node1, Node2} <- Pairs]; + false -> + ok + end, Selected = pick_nodes(N, All, File, Line), case diskless(Config) of true -> @@ -1137,3 +1166,36 @@ sort(W) -> get_ext_test_server_name() -> list_to_atom("ext_test_server_" ++ atom_to_list(node())). + +set_peer_ref(Node, Ref) -> + gen_tcp_blocking_dist:set_peer_ref(Node, Ref). + +get_peer_ref(Node) -> + gen_tcp_blocking_dist:get_peer_ref(Node). + +get_peer_ref(Node, Default) -> + gen_tcp_blocking_dist:get_peer_ref(Node, Default). + +%% Simulate network outage, use this to block/unblock communication between 2 nodes +has_network_blocker() -> + gen_tcp_blocking_dist:has_network_blocker(). + +skip_if_no_network_blocker(Config) -> + case has_network_blocker() of + true -> + Config; + false -> + {skip, "Network blocker required"} + end. + +block_peer(From, To) -> + gen_tcp_blocking_dist:block_peer(From, To). + +unblock_peer(From, To) -> + gen_tcp_blocking_dist:unblock_peer(From, To). + +block_pair(Node1, Node2) -> + gen_tcp_blocking_dist:block_pair(Node1, Node2). + +unblock_pair(Node1, Node2) -> + gen_tcp_blocking_dist:unblock_pair(Node1, Node2). diff --git a/lib/mnesia/test/mt b/lib/mnesia/test/mt index d05255805c28..c0a62b147798 100755 --- a/lib/mnesia/test/mt +++ b/lib/mnesia/test/mt @@ -24,18 +24,29 @@ # Author: Hakan Mattsson # Purpose: Simplified execution of the test suite # -# Usage: mt +# Usage: mt -#top=".." top="$ERL_TOP/lib/mnesia" h=`hostname` p="-pa $top/examples -pa $top/ebin -pa $top/test -mnesia_test_verbose true" log=test_log$$ latest=test_log_latest +if [ "$1" = "true" ]; then + dist_args="-proto_dist gen_tcp_blocking" + shift +elif [ "$1" = "false" ]; then + shift +fi args=${1+"$@"} -erlcmd="erl -sname a@localhost $p $args -mnesia_test_timeout" -erlcmd1="erl -sname a1@localhost $p $args" -erlcmd2="erl -sname a2@localhost $p $args" +tmp=$(mktemp -d) +if [ "$?" != "0" ]; then + echo "Failed to create temp directory" >&2 + exit 1 +fi +trap 'rm -rf "$tmp"' EXIT +erlcmd="erl -sname a@localhost $p $args \ + $dist_args \ + -mnesia_test_timeout -s mt start_peers a1 a2 $tmp $args" if test z"$MT_TERM" = z ; then MT_TERM=xterm @@ -77,8 +88,9 @@ echo "" $MT_TERM $geom0 $title a $exec script -c "$erlcmd" -f $log & -$MT_TERM $geom1 $title a1 $exec $erlcmd1 & -$MT_TERM $geom2 $title a2 $exec $erlcmd2 & +sleep 1 +$MT_TERM $geom1 $title a1 $exec to_erl $tmp/a1/ & +$MT_TERM $geom2 $title a2 $exec to_erl $tmp/a2/ & echo "Give the following command in order to see the outcome from node a@$h"":" echo "" diff --git a/lib/mnesia/test/mt.erl b/lib/mnesia/test/mt.erl index 39d8238ccedd..70f2d32da8b3 100644 --- a/lib/mnesia/test/mt.erl +++ b/lib/mnesia/test/mt.erl @@ -33,17 +33,20 @@ -module(mt). -author('hakan@erix.ericsson.se'). -export([ - t/0, t/1, t/2, t/3, % Run test cases - loop/1, loop/2, loop/3, % loop test cases - doc/0, doc/1, % Generate test case doc - struct/0, struct/1, % View test suite struct - shutdown/0, ping/0, start_nodes/0, % Node admin - read_config/0, write_config/1 % Config admin - ]). + t/0, t/1, t/2, t/3, % Run test cases + loop/1, loop/2, loop/3, % loop test cases + doc/0, doc/1, % Generate test case doc + struct/0, struct/1, % View test suite struct + shutdown/0, ping/0, start_nodes/0, % Node admin + start_peers/1, % Peers for mt script + read_config/0, write_config/1 % Config admin + ]). -compile([{no_auto_import,[alias/1]}, {nowarn_possibly_unsafe_function, {file, consult, 1}}]). +-include("gen_tcp_blocking_dist.hrl"). + %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% %% Aliases for the (sub) test suites alias(all) -> mnesia_SUITE; @@ -278,3 +281,47 @@ 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. + +%% Peer nodes for mt script +start_peers(Args) -> + [Name2, Name3, TmpDirA | ExtraArgs0] = Args, + CodePaths = lists:append([["-pa", P] || P <- code:get_path(), P /= "."]), + DistOpts = ?NETWORK_BLOCKER_DIST_OPTS, + ExtraArgs = + case mnesia_test_lib:has_network_blocker() of + true -> + CodePaths ++ DistOpts ++ ExtraArgs0; + false -> + CodePaths ++ ExtraArgs0 + end, + TmpDir = atom_to_list(TmpDirA), + {Name2, P2, N2} = start_peer(Name2, TmpDir, ExtraArgs), + {Name3, P3, N3} = start_peer(Name3, TmpDir, ExtraArgs), + ok = mnesia_test_lib:set_peer_ref(N2, P2), + ok = mnesia_test_lib:set_peer_ref(N3, P3). + +start_peer(Name, TmpDir, ExtraArgs) -> + Dir = TmpDir ++ "/" ++ atom_to_list(Name) ++ "/", + ok = filelib:ensure_dir(Dir), + Erl = case init:get_argument(progname) of + {ok, [[Prog]]} -> + case os:find_executable(Prog) of + false -> "erl"; + Found -> Found + end; + _ -> "erl" + end, + RunErl = os:find_executable("run_erl"), + {ok, Peer, Node} = peer:start(#{ + name => Name, + host => "localhost", + connection => 0, + detached => false, + exec => {RunErl, ["-daemon"]}, + post_process_args => fun(Args) -> + Cmd = lists:flatten([Erl, " ", lists:join(" ", Args)]), + [Dir, Dir, "exec " ++ Cmd] + end, + args => ExtraArgs + }), + {Name, Peer, Node}.