Add automerge after net splits.

src/syn_backbone.erl

@@ -70,14 +70,16 @@ find_by_pid(Pid) ->
 register(Key, Pid) ->
     case find_by_key(Key) of
         undefined ->
+            %% get processes's node
+            Node = node(Pid),
             %% add to table
             mnesia:dirty_write(#syn_processes_table{
                 key = Key,
                 pid = Pid,
-                node = node()
+                node = Node
             }),
             %% link
-            gen_server:call(?MODULE, {link_process, Pid});
+            gen_server:call({?MODULE, Node}, {link_process, Pid});
         _ ->
             {error, taken}
     end.
@@ -136,7 +138,7 @@ handle_call({unlink_process, Pid}, _From, State) ->
     {reply, ok, State};
 
 handle_call(Request, From, State) ->
-    error_logger:warning_msg("Received from ~p an unknown call message: ~p", [Request, From]),
+    error_logger:warning_msg("Received from ~p an unknown call message: ~p~n", [Request, From]),
     {reply, undefined, State}.
 
 %% ----------------------------------------------------------------------------------------------------------
@@ -148,7 +150,7 @@ handle_call(Request, From, State) ->
     {stop, Reason :: any(), #state{}}.
 
 handle_cast(Msg, State) ->
-    error_logger:warning_msg("Received an unknown cast message: ~p", [Msg]),
+    error_logger:warning_msg("Received an unknown cast message: ~p~n", [Msg]),
     {noreply, State}.
 
 %% ----------------------------------------------------------------------------------------------------------
@@ -167,7 +169,8 @@ handle_info({'EXIT', Pid, Reason}, State) ->
             undefined ->
                 case Reason of
                     normal -> ok;
-                    _ -> error_logger:warning_msg("Received a crash message from an unlinked process ~p with reason: ~p", [Pid, Reason])
+                    _ ->
+                        error_logger:warning_msg("Received a crash message from an unlinked process ~p with reason: ~p~n", [Pid, Reason])
                 end;
             Key ->
                 %% delete from table
@@ -176,7 +179,7 @@ handle_info({'EXIT', Pid, Reason}, State) ->
                 case Reason of
                     normal -> ok;
                     killed -> ok;
-                    _ -> error_logger:error_msg("Process with key ~p crashed with reason: ~p", [Key, Reason])
+                    _ -> error_logger:error_msg("Process with key ~p crashed with reason: ~p~n", [Key, Reason])
                 end
         end
     end),
@@ -184,7 +187,7 @@ handle_info({'EXIT', Pid, Reason}, State) ->
     {noreply, State};
 
 handle_info(Info, State) ->
-    error_logger:warning_msg("Received an unknown info message: ~p", [Info]),
+    error_logger:warning_msg("Received an unknown info message: ~p~n", [Info]),
     {noreply, State}.
 
 %% ----------------------------------------------------------------------------------------------------------
@@ -192,7 +195,7 @@ handle_info(Info, State) ->
 %% ----------------------------------------------------------------------------------------------------------
 -spec terminate(Reason :: any(), #state{}) -> terminated.
 terminate(Reason, _State) ->
-    error_logger:info_msg("Terminating syn with reason: ~p", [Reason]),
+    error_logger:info_msg("Terminating syn with reason: ~p~n", [Reason]),
     terminated.
 
 %% ----------------------------------------------------------------------------------------------------------
@@ -219,7 +222,7 @@ initdb() ->
         {storage_properties, [{ets, [{read_concurrency, true}]}]}
     ]) of
         {atomic, ok} ->
-            error_logger:info_msg("syn_processes_table was successfully created."),
+            error_logger:info_msg("syn_processes_table was successfully created.~n"),
             ok;
         {aborted, {already_exists, syn_processes_table}} ->
             %% table already exists, try to add current node as copy
@@ -228,7 +231,7 @@ initdb() ->
             %% table already exists, try to add current node as copy
             add_table_copy_to_local_node();
         Other ->
-            error_logger:error_msg("Error while creating syn_processes_table: ~p", [Other]),
+            error_logger:error_msg("Error while creating syn_processes_table: ~p~n", [Other]),
             {error, Other}
     end.
 
@@ -237,16 +240,16 @@ add_table_copy_to_local_node() ->
     CurrentNode = node(),
     case mnesia:add_table_copy(syn_processes_table, node(), ram_copies) of
         {atomic, ok} ->
-            error_logger:info_msg("Copy of syn_processes_table was successfully added to current node."),
+            error_logger:info_msg("Copy of syn_processes_table was successfully added to current node.~n"),
             ok;
         {aborted, {already_exists, syn_processes_table}} ->
-            %% a copy of syn_processes_table is already added to current node
+            error_logger:info_msg("Copy of syn_processes_table is already added to current node.~n"),
             ok;
         {aborted, {already_exists, syn_processes_table, CurrentNode}} ->
-            %% a copy of syn_processes_table is already added to current node
+            error_logger:info_msg("Copy of syn_processes_table is already added to current node.~n"),
             ok;
         {aborted, Reason} ->
-            error_logger:error_msg("Error while creating copy of syn_processes_table: ~p", [Reason]),
+            error_logger:error_msg("Error while creating copy of syn_processes_table: ~p~n", [Reason]),
             {error, Reason}
     end.
 

src/syn_netsplits.erl

@@ -35,6 +35,10 @@
 %% gen_server callbacks
 -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]).
 
+%% internal
+-export([get_processes_info_of_node/1]).
+-export([write_processes_info_to_node/2]).
+
 %% records
 -record(state, {}).
 
@@ -63,6 +67,8 @@ start_link() ->
     ignore |
     {stop, Reason :: any()}.
 init([]) ->
+    %% trap linked processes signal
+    process_flag(trap_exit, true),
     %% monitor mnesia events
     mnesia:subscribe(system),
     {ok, #state{}}.
@@ -79,7 +85,7 @@ init([]) ->
     {stop, Reason :: any(), #state{}}.
 
 handle_call(Request, From, State) ->
-    error_logger:warning_msg("Received from ~p an unknown call message: ~p", [Request, From]),
+    error_logger:warning_msg("Received from ~p an unknown call message: ~p~n", [Request, From]),
     {reply, undefined, State}.
 
 %% ----------------------------------------------------------------------------------------------------------
@@ -91,7 +97,7 @@ handle_call(Request, From, State) ->
     {stop, Reason :: any(), #state{}}.
 
 handle_cast(Msg, State) ->
-    error_logger:warning_msg("Received an unknown cast message: ~p", [Msg]),
+    error_logger:warning_msg("Received an unknown cast message: ~p~n", [Msg]),
     {noreply, State}.
 
 %% ----------------------------------------------------------------------------------------------------------
@@ -103,13 +109,13 @@ handle_cast(Msg, State) ->
     {stop, Reason :: any(), #state{}}.
 
 handle_info({mnesia_system_event, {inconsistent_database, Context, Node}}, State) ->
-    error_logger:warning_msg("MNESIA signalled an inconsistent database on node: ~p with context: ~p, initiating automerge", [Node, Context]),
-    %% automerge(Node),
+    error_logger:warning_msg("MNESIA signalled an inconsistent database on node: ~p with context: ~p, initiating automerge~n", [Node, Context]),
+    automerge(Node),
     {noreply, State};
 
 handle_info({mnesia_system_event, {mnesia_down, Node}}, State) when Node =/= node() ->
-    error_logger:warning_msg("Received a MNESIA down event, removing all pids of node ~p", [Node]),
-    %% delete_pids_of_disconnected_node(Node),
+    error_logger:warning_msg("Received a MNESIA down event, removing all pids of node ~p~n", [Node]),
+    delete_pids_of_disconnected_node(Node),
     {noreply, State};
 
 handle_info({mnesia_system_event, _MnesiaEvent}, State) ->
@@ -117,7 +123,7 @@ handle_info({mnesia_system_event, _MnesiaEvent}, State) ->
     {noreply, State};
 
 handle_info(Info, State) ->
-    error_logger:warning_msg("Received an unknown info message: ~p", [Info]),
+    error_logger:warning_msg("Received an unknown info message: ~p~n", [Info]),
     {noreply, State}.
 
 %% ----------------------------------------------------------------------------------------------------------
@@ -125,7 +131,7 @@ handle_info(Info, State) ->
 %% ----------------------------------------------------------------------------------------------------------
 -spec terminate(Reason :: any(), #state{}) -> terminated.
 terminate(Reason, _State) ->
-    error_logger:info_msg("Terminating syn with reason: ~p", [Reason]),
+    error_logger:info_msg("Terminating syn with reason: ~p~n", [Reason]),
     terminated.
 
 %% ----------------------------------------------------------------------------------------------------------
@@ -134,3 +140,126 @@ terminate(Reason, _State) ->
 -spec code_change(OldVsn :: any(), #state{}, Extra :: any()) -> {ok, #state{}}.
 code_change(_OldVsn, State, _Extra) ->
     {ok, State}.
+
+
+%% ===================================================================
+%% Internal
+%% ===================================================================
+-spec delete_pids_of_disconnected_node(Node :: atom()) -> pid().
+delete_pids_of_disconnected_node(Node) ->
+    %% don't lock gen server
+    spawn(fun() ->
+        %% build match specs
+        MatchHead = #syn_processes_table{key = '$1', node = '$2', _ = '_'},
+        Guard = {'=:=', '$2', Node},
+        IdFormat = '$1',
+        %% delete
+        DelF = fun(Id) -> mnesia:dirty_delete({syn_processes_table, Id}) end,
+        NodePids = mnesia:dirty_select(syn_processes_table, [{MatchHead, [Guard], [IdFormat]}]),
+        lists:foreach(DelF, NodePids)
+    end).
+
+-spec automerge(RemoteNode :: atom()) -> ok.
+automerge(RemoteNode) ->
+    global:trans({{?MODULE, automerge}, self()},
+        fun() ->
+            error_logger:warning_msg("AUTOMERGE starting for remote node ~s (global lock is set)~n", [RemoteNode]),
+            check_stitch(RemoteNode),
+            error_logger:warning_msg("AUTOMERGE done (global lock will be unset)~n")
+        end).
+
+-spec check_stitch(RemoteNode :: atom()) -> ok.
+check_stitch(RemoteNode) ->
+    case lists:member(RemoteNode, mnesia:system_info(running_db_nodes)) of
+        true ->
+            ok;
+        false ->
+            stitch(RemoteNode),
+            ok
+    end.
+
+-spec stitch(RemoteNode :: atom()) -> {'ok', any()} | {'error', any()}.
+stitch(RemoteNode) ->
+    mnesia_controller:connect_nodes(
+        [RemoteNode],
+        fun(MergeF) ->
+            case MergeF([syn_processes_table]) of
+                {merged, _, _} = Res ->
+                    stitch_tab(RemoteNode),
+                    Res;
+                Other ->
+                    Other
+            end
+        end).
+
+-spec stitch_tab(RemoteNode :: atom()) -> ok.
+stitch_tab(RemoteNode) ->
+    %% get remote processes info
+    RemoteProcessesInfo = rpc:call(RemoteNode, ?MODULE, get_processes_info_of_node, [RemoteNode]),
+    %% get local processes info
+    LocalProcessesInfo = get_processes_info_of_node(node()),
+    %% purge doubles (if any)
+    {LocalProcessesInfo1, RemoteProcessesInfo1} = purge_double_processes_from_local_node(LocalProcessesInfo, RemoteProcessesInfo),
+    %% write
+    write_remote_processes_to_local(RemoteNode, RemoteProcessesInfo1),
+    write_local_processes_to_remote(RemoteNode, LocalProcessesInfo1).
+
+-spec purge_double_processes_from_local_node(LocalProcessesInfo :: list(), RemoteProcessesInfo :: list()) ->
+    {LocalProcessesInfo :: list(), RemoteProcessesInfo :: list()}.
+purge_double_processes_from_local_node(LocalProcessesInfo, RemoteProcessesInfo) ->
+    %% create ETS table
+    Tab = ets:new(syn_automerge_doubles_table, [set]),
+
+    %% insert local processes info
+    ets:insert(Tab, LocalProcessesInfo),
+
+    %% find doubles
+    F = fun({Key, _RemoteProcessPid}) ->
+        case ets:lookup(Tab, Key) of
+            [] -> ok;
+            [{Key, LocalProcessPid}] ->
+                error_logger:warning_msg("Found a double process for ~s, killing it on local node~n", [Key]),
+                %% remove it from local mnesia table
+                mnesia:dirty_delete(syn_processes_table, Key),
+                %% remove it from ETS
+                ets:delete(Tab, Key),
+                %% kill the process
+                exit(LocalProcessPid, kill)
+        end
+    end,
+    lists:foreach(F, RemoteProcessesInfo),
+
+    %% compute local processes without doubles
+    LocalProcessesInfo1 = ets:tab2list(Tab),
+    %% delete ETS table
+    ets:delete(Tab),
+    %% return
+    {LocalProcessesInfo1, RemoteProcessesInfo}.
+
+-spec write_remote_processes_to_local(RemoteNode :: atom(), RemoteProcessesInfo :: list()) -> ok.
+write_remote_processes_to_local(RemoteNode, RemoteProcessesInfo) ->
+    write_processes_info_to_node(RemoteNode, RemoteProcessesInfo).
+
+-spec write_local_processes_to_remote(RemoteNode :: atom(), LocalProcessesInfo :: list()) -> ok.
+write_local_processes_to_remote(RemoteNode, LocalProcessesInfo) ->
+    ok = rpc:call(RemoteNode, ?MODULE, write_processes_info_to_node, [node(), LocalProcessesInfo]).
+
+-spec get_processes_info_of_node(Node :: atom()) -> list().
+get_processes_info_of_node(Node) ->
+    %% build match specs
+    MatchHead = #syn_processes_table{key = '$1', pid = '$2', node = '$3'},
+    Guard = {'=:=', '$3', Node},
+    ProcessInfoFormat = {{'$1', '$2'}},
+    %% select
+    mnesia:dirty_select(syn_processes_table, [{MatchHead, [Guard], [ProcessInfoFormat]}]).
+
+-spec write_processes_info_to_node(Node :: atom(), ProcessesInfo :: list()) -> ok.
+write_processes_info_to_node(Node, ProcessesInfo) ->
+    FWrite = fun({Key, ProcessPid}) ->
+        mnesia:dirty_write(#syn_processes_table{
+            key = Key,
+            pid = ProcessPid,
+            node = Node
+        })
+    end,
+    lists:foreach(FWrite, ProcessesInfo).

test/syn_create_mnesia_SUITE.erl

@@ -125,7 +125,7 @@ end_per_suite(_Config) -> ok.
 %% Reason = term()
 %% -------------------------------------------------------------------
 init_per_group(two_nodes_mnesia_creation, Config) ->
-    %% get slave node short name
+    %% start slave
     SlaveNodeShortName = proplists:get_value(slave_node_short_name, Config),
     {ok, SlaveNodeName} = syn_test_suite_helper:start_slave(SlaveNodeShortName),
     %% config
@@ -168,10 +168,8 @@ init_per_testcase(_TestCase, Config) ->
 % ----------------------------------------------------------------------------------------------------------
 end_per_testcase(_TestCase, Config) ->
     %% get slave
-    case proplists:get_value(slave_node_name, Config) of
-        undefined -> syn_test_suite_helper:clean_after_test();
-        SlaveNodeName -> syn_test_suite_helper:clean_after_test(SlaveNodeName)
-    end.
+    SlaveNodeName = proplists:get_value(slave_node_name, Config),
+    syn_test_suite_helper:clean_after_test(SlaveNodeName).
 
 %% ===================================================================
 %% Tests

test/syn_netsplits_SUITE.erl

@@ -0,0 +1,228 @@
+%% ==========================================================================================================
+%% Syn - A global process registry.
+%%
+%% Copyright (C) 2015, Roberto Ostinelli <roberto@ostinelli.net>.
+%% All rights reserved.
+%%
+%% The MIT License (MIT)
+%%
+%% Copyright (c) 2015 Roberto Ostinelli
+%%
+%% Permission is hereby granted, free of charge, to any person obtaining a copy
+%% of this software and associated documentation files (the "Software"), to deal
+%% in the Software without restriction, including without limitation the rights
+%% to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+%% copies of the Software, and to permit persons to whom the Software is
+%% furnished to do so, subject to the following conditions:
+%%
+%% The above copyright notice and this permission notice shall be included in
+%% all copies or substantial portions of the Software.
+%%
+%% THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+%% IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+%% FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+%% AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+%% LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+%% OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+%% THE SOFTWARE.
+-module(syn_netsplits_SUITE).
+
+%% callbacks
+-export([all/0]).
+-export([init_per_suite/1, end_per_suite/1]).
+-export([groups/0, init_per_group/2, end_per_group/2]).
+-export([init_per_testcase/2, end_per_testcase/2]).
+
+%% tests
+-export([
+    two_nodes_netsplit_when_there_are_no_conflicts/1
+]).
+
+%% include
+-include_lib("common_test/include/ct.hrl").
+
+
+%% ===================================================================
+%% Callbacks
+%% ===================================================================
+
+%% -------------------------------------------------------------------
+%% Function: all() -> GroupsAndTestCases | {skip,Reason}
+%% GroupsAndTestCases = [{group,GroupName} | TestCase]
+%% GroupName = atom()
+%% TestCase = atom()
+%% Reason = term()
+%% -------------------------------------------------------------------
+all() ->
+    [
+        {group, two_nodes_netsplits}
+    ].
+
+%% -------------------------------------------------------------------
+%% Function: groups() -> [Group]
+%% Group = {GroupName,Properties,GroupsAndTestCases}
+%% GroupName = atom()
+%% Properties = [parallel | sequence | Shuffle | {RepeatType,N}]
+%% GroupsAndTestCases = [Group | {group,GroupName} | TestCase]
+%% TestCase = atom()
+%% Shuffle = shuffle | {shuffle,{integer(),integer(),integer()}}
+%% RepeatType = repeat | repeat_until_all_ok | repeat_until_all_fail |
+%%			   repeat_until_any_ok | repeat_until_any_fail
+%% N = integer() | forever
+%% -------------------------------------------------------------------
+groups() ->
+    [
+        {two_nodes_netsplits, [shuffle], [
+            two_nodes_netsplit_when_there_are_no_conflicts
+        ]}
+    ].
+%% -------------------------------------------------------------------
+%% Function: init_per_suite(Config0) ->
+%%				Config1 | {skip,Reason} |
+%%              {skip_and_save,Reason,Config1}
+%% Config0 = Config1 = [tuple()]
+%% Reason = term()
+%% -------------------------------------------------------------------
+init_per_suite(Config) ->
+    %% init
+    SlaveNodeShortName = syn_slave,
+    %% start slave
+    {ok, SlaveNodeName} = syn_test_suite_helper:start_slave(SlaveNodeShortName),
+    %% config
+    [
+        {slave_node_short_name, SlaveNodeShortName},
+        {slave_node_name, SlaveNodeName}
+        | Config
+    ].
+
+%% -------------------------------------------------------------------
+%% Function: end_per_suite(Config0) -> void() | {save_config,Config1}
+%% Config0 = Config1 = [tuple()]
+%% -------------------------------------------------------------------
+end_per_suite(Config) ->
+    %% get slave node name
+    SlaveNodeShortName = proplists:get_value(slave_node_short_name, Config),
+    %% stop slave
+    syn_test_suite_helper:stop_slave(SlaveNodeShortName).
+
+%% -------------------------------------------------------------------
+%% Function: init_per_group(GroupName, Config0) ->
+%%				Config1 | {skip,Reason} |
+%%              {skip_and_save,Reason,Config1}
+%% GroupName = atom()
+%% Config0 = Config1 = [tuple()]
+%% Reason = term()
+%% -------------------------------------------------------------------
+init_per_group(_GroupName, Config) -> Config.
+
+%% -------------------------------------------------------------------
+%% Function: end_per_group(GroupName, Config0) ->
+%%				void() | {save_config,Config1}
+%% GroupName = atom()
+%% Config0 = Config1 = [tuple()]
+%% -------------------------------------------------------------------
+end_per_group(_GroupName, _Config) -> ok.
+
+% ----------------------------------------------------------------------------------------------------------
+% Function: init_per_testcase(TestCase, Config0) ->
+%				Config1 | {skip,Reason} | {skip_and_save,Reason,Config1}
+% TestCase = atom()
+% Config0 = Config1 = [tuple()]
+% Reason = term()
+% ----------------------------------------------------------------------------------------------------------
+init_per_testcase(_TestCase, Config) ->
+    %% get slave
+    SlaveNodeName = proplists:get_value(slave_node_name, Config),
+    %% set schema location
+    application:set_env(mnesia, schema_location, ram),
+    rpc:call(SlaveNodeName, mnesia, schema_location, [ram]),
+    %% start syn
+    ok = syn:start(),
+    ok = rpc:call(SlaveNodeName, syn, start, []),
+    timer:sleep(100),
+    Config.
+
+% ----------------------------------------------------------------------------------------------------------
+% Function: end_per_testcase(TestCase, Config0) ->
+%				void() | {save_config,Config1} | {fail,Reason}
+% TestCase = atom()
+% Config0 = Config1 = [tuple()]
+% Reason = term()
+% ----------------------------------------------------------------------------------------------------------
+end_per_testcase(_TestCase, Config) ->
+    %% get slave
+    SlaveNodeName = proplists:get_value(slave_node_name, Config),
+    syn_test_suite_helper:clean_after_test(SlaveNodeName).
+
+%% ===================================================================
+%% Tests
+%% ===================================================================
+two_nodes_netsplit_when_there_are_no_conflicts(Config) ->
+    %% get slave
+    SlaveNodeName = proplists:get_value(slave_node_name, Config),
+    CurrentNode = node(),
+
+    %% start processes
+    LocalPid = syn_test_suite_helper:start_process(),
+    SlavePidLocal = syn_test_suite_helper:start_process(SlaveNodeName),
+    SlavePidSlave = syn_test_suite_helper:start_process(SlaveNodeName),
+
+    %% register
+    ok = syn:register(local_pid, LocalPid),
+    ok = syn:register(slave_pid_local, SlavePidLocal),    %% slave registered on local node
+    ok = rpc:call(SlaveNodeName, syn, register, [slave_pid_slave, SlavePidSlave]),    %% slave registered on slave node
+    timer:sleep(100),
+
+    %% check tables
+    3 = mnesia:table_info(syn_processes_table, size),
+    3 = rpc:call(SlaveNodeName, mnesia, table_info, [syn_processes_table, size]),
+
+    LocalActiveReplicas = mnesia:table_info(syn_processes_table, active_replicas),
+    2 = length(LocalActiveReplicas),
+    true = lists:member(SlaveNodeName, LocalActiveReplicas),
+    true = lists:member(CurrentNode, LocalActiveReplicas),
+
+    SlaveActiveReplicas = rpc:call(SlaveNodeName, mnesia, table_info, [syn_processes_table, active_replicas]),
+    2 = length(SlaveActiveReplicas),
+    true = lists:member(SlaveNodeName, SlaveActiveReplicas),
+    true = lists:member(CurrentNode, SlaveActiveReplicas),
+
+    %% simulate net split
+    syn_test_suite_helper:disconnect_node(SlaveNodeName),
+    timer:sleep(1000),
+
+    %% check tables
+    1 = mnesia:table_info(syn_processes_table, size),
+    [CurrentNode] = mnesia:table_info(syn_processes_table, active_replicas),
+
+    %% reconnect
+    syn_test_suite_helper:connect_node(SlaveNodeName),
+    timer:sleep(2000),
+
+    %% check tables
+    3 = mnesia:table_info(syn_processes_table, size),
+    3 = rpc:call(SlaveNodeName, mnesia, table_info, [syn_processes_table, size]),
+
+    LocalActiveReplicas2 = mnesia:table_info(syn_processes_table, active_replicas),
+    2 = length(LocalActiveReplicas2),
+    true = lists:member(SlaveNodeName, LocalActiveReplicas2),
+    true = lists:member(CurrentNode, LocalActiveReplicas2),
+
+    SlaveActiveReplicas2 = rpc:call(SlaveNodeName, mnesia, table_info, [syn_processes_table, active_replicas]),
+    2 = length(SlaveActiveReplicas2),
+    true = lists:member(SlaveNodeName, SlaveActiveReplicas2),
+    true = lists:member(CurrentNode, SlaveActiveReplicas2),
+
+    %% check processes
+    LocalPid = syn:find_by_key(local_pid),
+    SlavePidLocal = syn:find_by_key(slave_pid_local),
+    SlavePidSlave = syn:find_by_key(slave_pid_slave),
+
+    LocalPid = rpc:call(SlaveNodeName, syn, find_by_key, [local_pid]),
+    SlavePidLocal = rpc:call(SlaveNodeName, syn, find_by_key, [slave_pid_local]),
+    SlavePidSlave = rpc:call(SlaveNodeName, syn, find_by_key, [slave_pid_slave]),
+
+    %% kill processes
+    syn_test_suite_helper:kill_process(LocalPid),
+    syn_test_suite_helper:kill_process(SlavePidLocal),
+    syn_test_suite_helper:kill_process(SlavePidSlave).

test/syn_register_processes_SUITE.erl

@@ -34,9 +34,6 @@
 -export([groups/0, init_per_group/2, end_per_group/2]).
 -export([init_per_testcase/2, end_per_testcase/2]).
 
-%% internal
--export([process_main/0]).
-
 %% tests
 -export([
     single_node_when_mnesia_is_ram_find_by_key/1,
@@ -126,7 +123,7 @@ end_per_suite(_Config) -> ok.
 %% Reason = term()
 %% -------------------------------------------------------------------
 init_per_group(two_nodes_process_registration, Config) ->
-    %% get slave node short name
+    %% start slave
     SlaveNodeShortName = proplists:get_value(slave_node_short_name, Config),
     {ok, SlaveNodeName} = syn_test_suite_helper:start_slave(SlaveNodeShortName),
     %% config
@@ -169,10 +166,8 @@ init_per_testcase(_TestCase, Config) ->
 % ----------------------------------------------------------------------------------------------------------
 end_per_testcase(_TestCase, Config) ->
     %% get slave
-    case proplists:get_value(slave_node_name, Config) of
-        undefined -> syn_test_suite_helper:clean_after_test();
-        SlaveNodeName -> syn_test_suite_helper:clean_after_test(SlaveNodeName)
-    end.
+    SlaveNodeName = proplists:get_value(slave_node_name, Config),
+    syn_test_suite_helper:clean_after_test(SlaveNodeName).
 
 %% ===================================================================
 %% Tests
@@ -183,7 +178,7 @@ single_node_when_mnesia_is_ram_find_by_key(_Config) ->
     %% start
     ok = syn:start(),
     %% start process
-    Pid = start_process(),
+    Pid = syn_test_suite_helper:start_process(),
     %% retrieve
     undefined = syn:find_by_key(<<"my proc">>),
     %% register
@@ -191,7 +186,7 @@ single_node_when_mnesia_is_ram_find_by_key(_Config) ->
     %% retrieve
     Pid = syn:find_by_key(<<"my proc">>),
     %% kill process
-    kill_process(Pid),
+    syn_test_suite_helper:kill_process(Pid),
     timer:sleep(100),
     %% retrieve
     undefined = syn:find_by_key(<<"my proc">>).
@@ -202,13 +197,13 @@ single_node_when_mnesia_is_ram_find_by_pid(_Config) ->
     %% start
     ok = syn:start(),
     %% start process
-    Pid = start_process(),
+    Pid = syn_test_suite_helper:start_process(),
     %% register
     ok = syn:register(<<"my proc">>, Pid),
     %% retrieve
     <<"my proc">> = syn:find_by_pid(Pid),
     %% kill process
-    kill_process(Pid),
+    syn_test_suite_helper:kill_process(Pid),
     timer:sleep(100),
     %% retrieve
     undefined = syn:find_by_pid(Pid).
@@ -219,15 +214,15 @@ single_node_when_mnesia_is_ram_re_register_error(_Config) ->
     %% start
     ok = syn:start(),
     %% start process
-    Pid = start_process(),
-    Pid2 = start_process(),
+    Pid = syn_test_suite_helper:start_process(),
+    Pid2 = syn_test_suite_helper:start_process(),
     %% register
     ok = syn:register(<<"my proc">>, Pid),
     {error, taken} = syn:register(<<"my proc">>, Pid2),
     %% retrieve
     Pid = syn:find_by_key(<<"my proc">>),
     %% kill process
-    kill_process(Pid),
+    syn_test_suite_helper:kill_process(Pid),
     timer:sleep(100),
     %% retrieve
     undefined = syn:find_by_key(<<"my proc">>),
@@ -236,7 +231,7 @@ single_node_when_mnesia_is_ram_re_register_error(_Config) ->
     %% retrieve
     Pid2 = syn:find_by_key(<<"my proc">>),
     %% kill process
-    kill_process(Pid),
+    syn_test_suite_helper:kill_process(Pid),
     timer:sleep(100),
     %% retrieve
     undefined = syn:find_by_pid(Pid).
@@ -247,7 +242,7 @@ single_node_when_mnesia_is_ram_unregister(_Config) ->
     %% start
     ok = syn:start(),
     %% start process
-    Pid = start_process(),
+    Pid = syn_test_suite_helper:start_process(),
     %% unregister
     {error, undefined} = syn:unregister(<<"my proc">>),
     %% register
@@ -260,7 +255,7 @@ single_node_when_mnesia_is_ram_unregister(_Config) ->
     undefined = syn:find_by_key(<<"my proc">>),
     undefined = syn:find_by_pid(Pid),
     %% kill process
-    kill_process(Pid).
+    syn_test_suite_helper:kill_process(Pid).
 
 single_node_when_mnesia_is_disc_find_by_key(_Config) ->
     %% set schema location
@@ -270,7 +265,7 @@ single_node_when_mnesia_is_disc_find_by_key(_Config) ->
     %% start
     ok = syn:start(),
     %% start process
-    Pid = start_process(),
+    Pid = syn_test_suite_helper:start_process(),
     %% retrieve
     undefined = syn:find_by_key(<<"my proc">>),
     %% register
@@ -278,7 +273,7 @@ single_node_when_mnesia_is_disc_find_by_key(_Config) ->
     %% retrieve
     Pid = syn:find_by_key(<<"my proc">>),
     %% kill process
-    kill_process(Pid),
+    syn_test_suite_helper:kill_process(Pid),
     timer:sleep(100),
     %% retrieve
     undefined = syn:find_by_key(<<"my proc">>).
@@ -294,7 +289,7 @@ two_nodes_when_mnesia_is_ram_find_by_key(Config) ->
     ok = rpc:call(SlaveNodeName, syn, start, []),
     timer:sleep(100),
     %% start process
-    Pid = start_process(),
+    Pid = syn_test_suite_helper:start_process(),
     %% retrieve
     undefined = syn:find_by_key(<<"my proc">>),
     undefined = rpc:call(SlaveNodeName, syn, find_by_key, [<<"my proc">>]),
@@ -304,7 +299,7 @@ two_nodes_when_mnesia_is_ram_find_by_key(Config) ->
     Pid = syn:find_by_key(<<"my proc">>),
     Pid = rpc:call(SlaveNodeName, syn, find_by_key, [<<"my proc">>]),
     %% kill process
-    kill_process(Pid),
+    syn_test_suite_helper:kill_process(Pid),
     timer:sleep(100),
     %% retrieve
     undefined = syn:find_by_key(<<"my proc">>),
@@ -323,30 +318,15 @@ two_nodes_when_mnesia_is_disc_find_by_pid(Config) ->
     ok = rpc:call(SlaveNodeName, syn, start, []),
     timer:sleep(100),
     %% start process
-    Pid = start_process(),
+    Pid = syn_test_suite_helper:start_process(),
     %% register
     ok = syn:register(<<"my proc">>, Pid),
     %% retrieve
     <<"my proc">> = syn:find_by_pid(Pid),
     <<"my proc">> = rpc:call(SlaveNodeName, syn, find_by_pid, [Pid]),
     %% kill process
-    kill_process(Pid),
+    syn_test_suite_helper:kill_process(Pid),
     timer:sleep(100),
     %% retrieve
     undefined = syn:find_by_pid(Pid),
     undefined = rpc:call(SlaveNodeName, syn, find_by_pid, [Pid]).
-
-%% ===================================================================
-%% Internal
-%% ===================================================================
-start_process() ->
-    Pid = spawn(?MODULE, process_main, []),
-    Pid.
-
-kill_process(Pid) ->
-    exit(Pid, kill).
-
-process_main() ->
-    receive
-        shutdown -> ok
-    end.

test/syn_test_suite_helper.erl

@@ -30,7 +30,9 @@
 
 %% API
 -export([start_slave/1, stop_slave/1]).
+-export([connect_node/1, disconnect_node/1]).
 -export([clean_after_test/0, clean_after_test/1]).
+-export([start_process/0, start_process/1, kill_process/1, process_main/0]).
 
 
 %% ===================================================================
@@ -38,17 +40,23 @@
 %% ===================================================================
 start_slave(NodeShortName) ->
     EbinFilePath = filename:join([filename:dirname(code:lib_dir(syn, ebin)), "ebin"]),
+    TestFilePath = filename:join([filename:dirname(code:lib_dir(syn, ebin)), "test"]),
     %% start slave
     {ok, NodeName} = ct_slave:start(NodeShortName, [
         {boot_timeout, 10},
-        {monitor_master, true},
-        {erl_flags, string:concat("-pa ", EbinFilePath)}
+        {erl_flags, lists:concat(["-pa ", EbinFilePath, " ", TestFilePath])}
     ]),
     {ok, NodeName}.
 
 stop_slave(NodeShortName) ->
     {ok, _} = ct_slave:stop(NodeShortName).
 
+connect_node(NodeName) ->
+    net_kernel:connect_node(NodeName).
+
+disconnect_node(NodeName) ->
+    erlang:disconnect_node(NodeName).
+
 clean_after_test() ->
     %% delete table
     {atomic, ok} = mnesia:delete_table(syn_processes_table),
@@ -59,6 +67,8 @@ clean_after_test() ->
     %% stop syn
     syn:stop().
 
+clean_after_test(undefined) ->
+    clean_after_test();
 clean_after_test(NodeName) ->
     %% delete table
     {atomic, ok} = mnesia:delete_table(syn_processes_table),
@@ -70,3 +80,20 @@ clean_after_test(NodeName) ->
     %% stop syn
     syn:stop(),
     rpc:call(NodeName, syn, stop, []).
+
+start_process() ->
+    Pid = spawn(?MODULE, process_main, []),
+    Pid.
+
+start_process(NodeName) ->
+%%     Pid = rpc:call(NodeName, erlang, spawn, [?MODULE, process_main, []]),
+    Pid = spawn(NodeName, ?MODULE, process_main, []),
+    Pid.
+
+kill_process(Pid) ->
+    exit(Pid, kill).
+
+process_main() ->
+    receive
+        shutdown -> ok
+    end.