Implement force_register/2,3.

README.md

@@ -161,15 +161,14 @@ Types:
 
 > You don't need to unregister names of processes that are about to die, since they are monitored by Syn and they will be removed automatically. If you manually unregister a process just before it dies, the callback on process exit (see here below) might not get called.
 
-To reregister a a previously registered Name with a different Pid:
+To register a previously registered Name with a _different_ Pid:
 
 ```erlang
-syn:reregister(Name, Pid) ->
-    syn:reregister(Name, Pid, undefined).
+syn:force_register(Name, Pid) ->
+    syn:force_register(Name, Pid, undefined).
 ```
-
 ```erlang
-syn:reregister(Name, Pid, Meta) -> ok.
+syn:force_register(Name, Pid, Meta) -> ok.
 
 Types:
     Name = any()
@@ -177,7 +176,7 @@ Types:
     Meta = any()
 ```
 
-> Re-registering is specifically useful if you are re-registering a process on a different node from where the process currently registered with Name is running on, as `reregister` will ensure that the registration succeeds.
+> Force registering is specifically useful if you are force registering a process on a different node from where the process currently registered with `Name` is running on, as `force_register/2,3` will ensure that the registration succeeds and propagates properly. You may otherwise experience a `{error, taken}` response to the registration call if you were to sequentially `unregister/1` and `register/2,3` a process, due to race conditions. Note that the previously registered process will not be killed and will be demonitored, so that the `on_process_exit/4` callback will _not_ be called (even if implemented) when the process dies.
 
 To retrieve the count of total registered processes running in the cluster:
 

src/syn.erl

@@ -28,7 +28,7 @@
 %% API
 -export([start/0, stop/0]).
 -export([register/2, register/3]).
--export([reregister/2, reregister/3]).
+-export([force_register/2, force_register/3]).
 -export([unregister/1]).
 -export([whereis/1, whereis/2]).
 -export([registry_count/0, registry_count/1]).
@@ -67,13 +67,13 @@ register(Name, Pid) ->
 register(Name, Pid, Meta) ->
     syn_registry:register(Name, Pid, Meta).
 
--spec reregister(Name :: any(), Pid :: pid()) -> ok | {error, Reason :: any()}.
-reregister(Name, Pid) ->
-    syn_registry:reregister(Name, Pid).
+-spec force_register(Name :: any(), Pid :: pid()) -> ok | {error, Reason :: any()}.
+force_register(Name, Pid) ->
+    syn_registry:force_register(Name, Pid).
 
--spec reregister(Name :: any(), Pid :: pid(), Meta :: any()) -> ok | {error, Reason :: any()}.
-reregister(Name, Pid, Meta) ->
-    syn_registry:reregister(Name, Pid, Meta).
+-spec force_register(Name :: any(), Pid :: pid(), Meta :: any()) -> ok | {error, Reason :: any()}.
+force_register(Name, Pid, Meta) ->
+    syn_registry:force_register(Name, Pid, Meta).
 
 -spec unregister(Name :: any()) -> ok | {error, Reason :: any()}.
 unregister(Name) ->

src/syn_registry.erl

@@ -29,13 +29,13 @@
 %% API
 -export([start_link/0]).
 -export([register/2, register/3]).
--export([reregister/2, reregister/3]).
+-export([force_register/2, force_register/3]).
 -export([unregister/1]).
 -export([whereis/1, whereis/2]).
 -export([count/0, count/1]).
 
 %% sync API
--export([sync_register/5, sync_unregister/3]).
+-export([sync_register/6, sync_unregister/3]).
 -export([sync_demonitor_and_kill_on_node/5]).
 -export([sync_get_local_registry_tuples/1]).
 -export([force_cluster_sync/0]).
@@ -70,38 +70,16 @@ register(Name, Pid) ->
 -spec register(Name :: any(), Pid :: pid(), Meta :: any()) -> ok | {error, Reason :: any()}.
 register(Name, Pid, Meta) when is_pid(Pid) ->
     Node = node(Pid),
-    gen_server:call({?MODULE, Node}, {register_on_node, Name, Pid, Meta}).
-
--spec reregister(Name :: any(), Pid :: pid()) -> ok | {error, Reason :: any()}.
-reregister(Name, Pid) ->
-    reregister(Name, Pid, undefined).
-
--spec reregister(Name :: any(), Pid :: pid(), Meta :: any()) -> ok | {error, Reason :: any()}.
-reregister(Name, Pid, Meta) when is_pid(Pid) ->
-    reregister(Name, Pid, Meta, 0).
-
--spec reregister(Name :: any(), Pid :: pid(), Meta :: any(), RetryCount :: non_neg_integer()) ->
-    ok | {error, Reason :: any()}.
-reregister(Name, Pid, Meta, RetryCount) when RetryCount > 40 ->
-    exit(self(), {timeout, {gen_server, call, [?MODULE, reregister, {Name, Pid, Meta}]}});
-reregister(Name, Pid, Meta, RetryCount) when is_pid(Pid) ->
-    ?MODULE:unregister(Name),
-    case find_registry_tuple_by_name(Name) of
-        undefined ->
-            case ?MODULE:register(Name, Pid, Meta) of
-                {error, taken} ->
-                    %% race conditions, retry
-                    timer:sleep(100),
-                    reregister(Name, Pid, Meta, RetryCount + 1);
-
-                Result ->
-                    Result
-            end;
+    gen_server:call({?MODULE, Node}, {register_on_node, Name, Pid, Meta, false}).
 
-        {Name, _, _, _} ->
-            timer:sleep(100),
-            reregister(Name, Pid, Meta, RetryCount + 1)
-    end.
+-spec force_register(Name :: any(), Pid :: pid()) -> ok | {error, Reason :: any()}.
+force_register(Name, Pid) ->
+    force_register(Name, Pid, undefined).
+
+-spec force_register(Name :: any(), Pid :: pid(), Meta :: any()) -> ok | {error, Reason :: any()}.
+force_register(Name, Pid, Meta) when is_pid(Pid) ->
+    Node = node(Pid),
+    gen_server:call({?MODULE, Node}, {register_on_node, Name, Pid, Meta, true}).
 
 -spec unregister(Name :: any()) -> ok | {error, Reason :: any()}.
 unregister(Name) ->
@@ -140,10 +118,17 @@ count(Node) ->
         [true]
     }]).
 
--spec sync_register(RemoteNode :: node(), Name :: any(), RemotePid :: pid(), RemoteMeta :: any(), RemoteTime :: integer()) ->
+-spec sync_register(
+    RemoteNode :: node(),
+    Name :: any(),
+    RemotePid :: pid(),
+    RemoteMeta :: any(),
+    RemoteTime :: integer(),
+    Force :: boolean()
+) ->
     ok.
-sync_register(RemoteNode, Name, RemotePid, RemoteMeta, RemoteTime) ->
-    gen_server:cast({?MODULE, RemoteNode}, {sync_register, Name, RemotePid, RemoteMeta, RemoteTime}).
+sync_register(RemoteNode, Name, RemotePid, RemoteMeta, RemoteTime, Force) ->
+    gen_server:cast({?MODULE, RemoteNode}, {sync_register, Name, RemotePid, RemoteMeta, RemoteTime, Force}).
 
 -spec sync_unregister(RemoteNode :: node(), Name :: any(), Pid :: pid()) -> ok.
 sync_unregister(RemoteNode, Name, Pid) ->
@@ -216,28 +201,42 @@ init([]) ->
     {stop, Reason :: any(), Reply :: any(), #state{}} |
     {stop, Reason :: any(), #state{}}.
 
-handle_call({register_on_node, Name, Pid, Meta}, _From, State) ->
+handle_call({register_on_node, Name, Pid, Meta, Force}, _From, State) ->
     %% check if pid is alive
     case is_process_alive(Pid) of
         true ->
             %% check if name available
             case find_registry_tuple_by_name(Name) of
                 undefined ->
+                    %% available
                     {ok, Time} = register_on_node(Name, Pid, Meta),
                     %% multicast
-                    multicast_register(Name, Pid, Meta, Time),
+                    multicast_register(Name, Pid, Meta, Time, false),
                     %% return
                     {reply, ok, State};
 
                 {Name, Pid, _, _} ->
+                    % same pid, overwrite
                     {ok, Time} = register_on_node(Name, Pid, Meta),
                     %% multicast
-                    multicast_register(Name, Pid, Meta, Time),
+                    multicast_register(Name, Pid, Meta, Time, false),
                     %% return
                     {reply, ok, State};
 
-                _ ->
-                    {reply, {error, taken}, State}
+                {Name, _, _, _} ->
+                    %% same name, different pid
+                    case Force of
+                        true ->
+                            %% force register
+                            {ok, Time} = register_on_node(Name, Pid, Meta),
+                            %% multicast
+                            multicast_register(Name, Pid, Meta, Time, true),
+                            %% return
+                            {reply, ok, State};
+
+                        _ ->
+                            {reply, {error, taken}, State}
+                    end
             end;
         _ ->
             {reply, {error, not_alive}, State}
@@ -266,7 +265,7 @@ handle_call(Request, From, State) ->
     {noreply, #state{}, Timeout :: non_neg_integer()} |
     {stop, Reason :: any(), #state{}}.
 
-handle_cast({sync_register, Name, RemotePid, RemoteMeta, RemoteTime}, State) ->
+handle_cast({sync_register, Name, RemotePid, RemoteMeta, RemoteTime, Force}, State) ->
     %% check for conflicts
     case find_registry_tuple_by_name(Name) of
         undefined ->
@@ -277,7 +276,20 @@ handle_cast({sync_register, Name, RemotePid, RemoteMeta, RemoteTime}, State) ->
             %% same process, no conflict, overwrite
             add_to_local_table(Name, RemotePid, RemoteMeta, RemoteTime, undefined);
 
-        {Name, TablePid, TableMeta, TableTime} ->
+        {Name, TablePid, _, _} when Force =:= true ->
+            case node(TablePid) =:= node() of
+                true ->
+                    %% demonitor
+                    MonitorRef = syn_registry:find_monitor_for_pid(TablePid),
+                    catch erlang:demonitor(MonitorRef, [flush]);
+
+                _ ->
+                    ok
+            end,
+            %% overwrite
+            add_to_local_table(Name, RemotePid, RemoteMeta, RemoteTime, undefined);
+
+        {Name, TablePid, TableMeta, TableTime} when Force =:= false ->
             %% different pid, we have a conflict
             global:trans({{?MODULE, {inconsistent_name, Name}}, self()},
                 fun() ->
@@ -444,11 +456,11 @@ code_change(_OldVsn, State, _Extra) ->
 %% ===================================================================
 %% Internal
 %% ===================================================================
--spec multicast_register(Name :: any(), Pid :: pid(), Meta :: any(), Time :: integer()) -> pid().
-multicast_register(Name, Pid, Meta, Time) ->
+-spec multicast_register(Name :: any(), Pid :: pid(), Meta :: any(), Time :: integer(), Force :: boolean()) -> pid().
+multicast_register(Name, Pid, Meta, Time, Force) ->
     spawn_link(fun() ->
         lists:foreach(fun(RemoteNode) ->
-            sync_register(RemoteNode, Name, Pid, Meta, Time)
+            sync_register(RemoteNode, Name, Pid, Meta, Time, Force)
         end, nodes())
     end).
 
@@ -522,8 +534,8 @@ add_to_local_table(Name, Pid, Meta, Time, MonitorRef) ->
         undefined ->
             undefined;
 
-        {Name, OldPid, _, _} ->
-            ets:delete(syn_registry_by_pid, {OldPid, Name})
+        {Name, PreviousPid, _, _} ->
+            ets:delete(syn_registry_by_pid, {PreviousPid, Name})
     end,
     %% overwrite & add
     ets:insert(syn_registry_by_name, {Name, Pid, Meta, Time, MonitorRef, node(Pid)}),

test/syn_registry_SUITE.erl

@@ -52,7 +52,7 @@
     two_nodes_registration_race_condition_conflict_resolution_when_process_died/1,
     two_nodes_registry_full_cluster_sync_on_boot_node_added_later/1,
     two_nodes_registry_full_cluster_sync_on_boot_syn_started_later/1,
-    two_nodes_reregister/1
+    two_nodes_force_register/1
 ]).
 -export([
     three_nodes_partial_netsplit_consistency/1,
@@ -129,7 +129,7 @@ groups() ->
             two_nodes_registration_race_condition_conflict_resolution_when_process_died,
             two_nodes_registry_full_cluster_sync_on_boot_node_added_later,
             two_nodes_registry_full_cluster_sync_on_boot_syn_started_later,
-            two_nodes_reregister
+            two_nodes_force_register
         ]},
         {three_nodes_process_registration, [shuffle], [
             three_nodes_partial_netsplit_consistency,
@@ -566,7 +566,7 @@ two_nodes_registration_race_condition_conflict_resolution_keep_remote_with_custo
     %% register
     ok = syn:register(ConflictingName, Pid0, node()),
     %% trigger conflict resolution on master node with something less recent (which would be discarded without a custom handler)
-    ok = syn_registry:sync_register(node(), ConflictingName, Pid1, keep_this_one, erlang:system_time() - 1000000000),
+    ok = syn_registry:sync_register(node(), ConflictingName, Pid1, keep_this_one, erlang:system_time() - 1000000000, false),
     timer:sleep(1000),
     %% check metadata, resolution happens on master node
     {Pid1, keep_this_one} = syn:whereis(ConflictingName, with_meta),
@@ -574,8 +574,7 @@ two_nodes_registration_race_condition_conflict_resolution_keep_remote_with_custo
     true = is_process_alive(Pid0),
     true = rpc:call(SlaveNode, erlang, is_process_alive, [Pid1]),
     %% check that discarded process is not monitored
-    {monitored_by, Monitors} = erlang:process_info(Pid0, monitored_by),
-    0 = length(Monitors).
+    {monitored_by, []} = erlang:process_info(Pid0, monitored_by).
 
 two_nodes_registration_race_condition_conflict_resolution_keep_local_with_custom_handler(Config) ->
     ConflictingName = "COMMON",
@@ -603,8 +602,7 @@ two_nodes_registration_race_condition_conflict_resolution_keep_local_with_custom
     true = is_process_alive(Pid0),
     true = rpc:call(SlaveNode, erlang, is_process_alive, [Pid1]),
     %% check that discarded process is not monitored
-    {monitored_by, Monitors} = rpc:call(SlaveNode, erlang, process_info, [Pid1, monitored_by]),
-    0 = length(Monitors).
+    {monitored_by, []} = rpc:call(SlaveNode, erlang, process_info, [Pid1, monitored_by]).
 
 two_nodes_registration_race_condition_conflict_resolution_when_process_died(Config) ->
     ConflictingName = "COMMON",
@@ -668,10 +666,13 @@ two_nodes_registry_full_cluster_sync_on_boot_syn_started_later(Config) ->
     Pid = syn:whereis(<<"proc">>),
     Pid = rpc:call(SlaveNode, syn, whereis, [<<"proc">>]).
 
-two_nodes_reregister(Config) ->
+two_nodes_force_register(Config) ->
     Name = "common name",
     %% get slave
     SlaveNode = proplists:get_value(slave_node, Config),
+    %% use custom handler
+    syn_test_suite_helper:use_custom_handler(),
+    rpc:call(SlaveNode, syn_test_suite_helper, use_custom_handler, []),
     %% start
     ok = syn:start(),
     ok = rpc:call(SlaveNode, syn, start, []),
@@ -679,14 +680,23 @@ two_nodes_reregister(Config) ->
     %% start processes
     PidLocal = syn_test_suite_helper:start_process(),
     PidRemote = syn_test_suite_helper:start_process(SlaveNode),
-    ok = rpc:call(SlaveNode, syn, register, [Name, PidRemote]),
-    %% fast unreg-reg
-    ok = syn:reregister(Name, PidLocal),
+    %% register
+    TestPid = self(),
+    ok = rpc:call(SlaveNode, syn, register, [Name, PidRemote, {pid_remote, TestPid}]),
+    timer:sleep(250),
+    {PidRemote, {pid_remote, TestPid}} = syn:whereis(Name, with_meta),
+    {PidRemote, {pid_remote, TestPid}} = rpc:call(SlaveNode, syn, whereis, [Name, with_meta]),
+    %% force
+    ok = syn:force_register(Name, PidLocal, {pid_local, TestPid}),
     timer:sleep(1000),
-    PidLocal = syn:whereis(Name),
-    ok = rpc:call(SlaveNode, syn, reregister, [Name, PidRemote, some_meta]),
+    {PidLocal, {pid_local, TestPid}} = syn:whereis(Name, with_meta),
+    {PidLocal, {pid_local, TestPid}} = rpc:call(SlaveNode, syn, whereis, [Name, with_meta]),
+    ok = rpc:call(SlaveNode, syn, force_register, [Name, PidRemote, {pid_remote, TestPid}]),
     timer:sleep(1000),
-    {PidRemote, some_meta} = syn:whereis(Name, with_meta).
+    {PidRemote, {pid_remote, TestPid}} = syn:whereis(Name, with_meta),
+    {PidRemote, {pid_remote, TestPid}} = rpc:call(SlaveNode, syn, whereis, [Name, with_meta]),
+    %% check that overwritten process is not monitored
+    {monitored_by, []} = erlang:process_info(PidLocal, monitored_by).
 
 three_nodes_partial_netsplit_consistency(Config) ->
     %% get slaves
@@ -972,10 +982,8 @@ three_nodes_start_syn_before_connecting_cluster_with_custom_conflict_resolution_
     true = rpc:call(SlaveNode1, erlang, is_process_alive, [Pid1]),
     true = rpc:call(SlaveNode2, erlang, is_process_alive, [Pid2]),
     %% check that discarded processes are not monitored
-    {monitored_by, Monitors0} = erlang:process_info(Pid0, monitored_by),
-    0 = length(Monitors0),
-    {monitored_by, Monitors2} = rpc:call(SlaveNode2, erlang, process_info, [Pid2, monitored_by]),
-    0 = length(Monitors2).
+    {monitored_by, []} = erlang:process_info(Pid0, monitored_by),
+    {monitored_by, []} = rpc:call(SlaveNode2, erlang, process_info, [Pid2, monitored_by]).
 
 three_nodes_registration_race_condition_custom_conflict_resolution(Config) ->
     ConflictingName = "COMMON",
@@ -1162,7 +1170,7 @@ three_nodes_resolve_conflict_on_all_nodes(Config) ->
     {Pid1, SlaveNode1} = rpc:call(SlaveNode1, syn, whereis, [CommonName, with_meta]),
     {Pid1, SlaveNode1} = rpc:call(SlaveNode2, syn, whereis, [CommonName, with_meta]),
     %% force a sync registration conflict on master node from slave 2
-    syn_registry:sync_register(node(), CommonName, Pid2, SlaveNode2, erlang:system_time() + 1000000000),
+    syn_registry:sync_register(node(), CommonName, Pid2, SlaveNode2, erlang:system_time() + 1000000000, false),
     timer:sleep(1000),
     %% check
     {Pid2, SlaveNode2} = syn:whereis(CommonName, with_meta),