add "lock_retries" option to avoid infinite loop on concurrent CETS join

src/cets_join.erl

@@ -38,7 +38,11 @@
 -type checkpoint_handler() :: fun((checkpoint()) -> ok).
 %% Checkpoint function for debugging.
 
--type join_opts() :: #{checkpoint_handler => checkpoint_handler(), join_ref => reference()}.
+-type join_opts() :: #{
+    checkpoint_handler => checkpoint_handler(),
+    join_ref => reference(),
+    lock_retries => non_neg_integer()
+}.
 %% Joining options.
 
 -export_type([join_ref/0]).
@@ -90,8 +94,6 @@ join_loop(LockKey, Info, LocalPid, RemotePid, Start, JoinOpts) ->
     %% - to avoid deadlocks, because joining does gen_server calls
     F = fun() ->
         Diff = erlang:system_time(millisecond) - Start,
-        %% Getting the lock could take really long time in case nodes are
-        %% overloaded or joining is already in progress on another node
         ?LOG_INFO(Info#{what => join_got_lock, after_time_ms => Diff}),
         %% Do joining in a separate process to reduce GC
         FF = handle_throw(fun() -> join2(Info, LocalPid, RemotePid, JoinOpts) end),
@@ -100,7 +102,9 @@ join_loop(LockKey, Info, LocalPid, RemotePid, Start, JoinOpts) ->
     LockRequest = {LockKey, self()},
     %% Just lock all nodes, no magic here :)
     Nodes = [node() | nodes()],
-    Retries = 0,
+    %% Retries > 0 enables randomized exponential backoff in global:set_lock,
+    %% helping avoid infinite retry loops when multiple nodes contend for global locks
+    Retries = maps:get(lock_retries, JoinOpts, 1),
     %% global could abort the transaction when one of the nodes goes down.
     %% It could usually abort it during startup or update.
     case global:trans(LockRequest, F, Nodes, Retries) of

test/cets_join_SUITE.erl

@@ -104,7 +104,8 @@ seq_cases() ->
     [
         joining_not_fully_connected_node_is_not_allowed,
         joining_not_fully_connected_node_is_not_allowed2,
-        join_interrupted_when_ping_crashes
+        join_interrupted_when_ping_crashes,
+        join_on_10_nodes_concurrently_with_the_same_lock
     ].
 
 cets_seq_no_log_cases() ->
@@ -114,7 +115,7 @@ cets_seq_no_log_cases() ->
 
 init_per_suite(Config) ->
     cets_test_setup:init_cleanup_table(),
-    cets_test_peer:start([ct2, ct3, ct5], Config).
+    cets_test_peer:start([ct2, ct3, ct4, ct5, ct6, ct7, ct8, ct9, ct10], Config).
 
 end_per_suite(Config) ->
     cets_test_setup:remove_cleanup_table(),
@@ -493,7 +494,6 @@ join_retried_if_lock_is_busy(Config) ->
         cets_join:join(Lock, #{}, Pid1, Pid2, #{checkpoint_handler => SleepyF})
     end),
     receive_message(join_start),
-    %% We actually would not return from cets_join:join unless we get the lock
     proc_lib:spawn_link(fun() ->
         ok = cets_join:join(Lock, #{}, Pid1, Pid2, #{checkpoint_handler => F})
     end),
@@ -606,6 +606,51 @@ join_interrupted_when_ping_crashes(Config) ->
     ?assertMatch({error, {task_failed, ping_all_failed, #{}}}, Res),
     meck:unload().
 
+join_on_10_nodes_concurrently_with_the_same_lock(Config) ->
+    ct:timetrap({seconds, 60}),
+    Node1 = node(),
+    Nodes = proplists:get_value(nodes, Config),
+    Peers = proplists:get_value(peers, Config),
+    PeerIds = [ct2, ct3, ct4, ct5, ct6, ct7, ct8, ct9, ct10],
+    Tab = make_name(Config),
+    Lock = lock_name(Config),
+    {ok, CetsPid1} = start(Node1, Tab),
+
+    CetsWithNodes = [
+        begin
+            Peer = maps:get(Id, Peers),
+            Node = maps:get(Id, Nodes),
+            {ok, Pid} = start(Peer, Tab),
+            {Node, Pid}
+        end
+     || Id <- PeerIds
+    ],
+    %% insert one row into CETS per node
+    cets:insert(Tab, {0}),
+    [
+        ok = cets_test_rpc:insert(Node, Tab, {N})
+     || {N, {Node, _Pid}} <- lists:zip(lists:seq(1, 9), CetsWithNodes)
+    ],
+    %% concurrently join for the same lock key.
+    ReqIds = [
+        erpc:send_request(Node, cets_join, join, [Lock, #{}, Pid, CetsPid1, #{lock_retries => 1}])
+     || {Node, Pid} <- CetsWithNodes
+    ],
+    [ok = erpc:receive_response(ReqId, timer:seconds(60)) || ReqId <- ReqIds],
+    %% check all nodes
+    {CetsNodes, CetsPids} = lists:unzip(CetsWithNodes),
+    AllCetsPids = [CetsPid1 | CetsPids],
+    AllCetsNodes = lists:sort([Node1 | CetsNodes]),
+    CetsInfos = [cets:info(Pid) || Pid <- AllCetsPids],
+    [?assertEqual(AllCetsNodes, lists:sort(maps:get(nodes, Info))) || Info <- CetsInfos],
+    %% The last committed join stamped the same join_ref everywhere
+    ?assertMatch([_], lists:usort([maps:get(join_ref, Info) || Info <- CetsInfos])),
+    %% No pause left behind
+    [?assertEqual([], maps:get(pause_monitors, Info)) || Info <- CetsInfos],
+    %% Data from all nodes is merged into every replica
+    ExpectedRows = [{N} || N <- lists:seq(0, 9)],
+    [?assertEqual({ok, ExpectedRows}, cets:remote_dump(Pid)) || Pid <- AllCetsPids].
+
 %% Helpers
 
 send_join_start_back_and_wait_for_continue_joining() ->