[Internal]hotfix: copy 0.8.1.11:src/ekka_locker.erl

src/ekka_locker.erl

@@ -18,6 +18,10 @@
 
 -include_lib("stdlib/include/ms_transform.hrl").
 
+-ifdef(TEST).
+-include_lib("eunit/include/eunit.hrl").
+-endif.
+
 -behaviour(gen_server).
 
 -export([ start_link/0
@@ -86,6 +90,7 @@
 
 %% 15 seconds by default
 -define(LEASE_TIME, 15000).
+-define(MC_TIMEOUT, 30000).
 
 %%--------------------------------------------------------------------
 %% API
@@ -145,11 +150,11 @@ acquire(Name, Resource, all, Piggyback) when is_atom(Name) ->
 
 acquire_locks(Nodes, Name, LockObj, Piggyback) ->
     {ResL, _BadNodes}
-        = rpc:multicall(Nodes, ?MODULE, acquire_lock, [Name, LockObj, Piggyback]),
+        = rpc:multicall(Nodes, ?MODULE, acquire_lock, [Name, LockObj, Piggyback], ?MC_TIMEOUT),
     case merge_results(ResL) of
         Res = {true, _}  -> Res;
         Res = {false, _} ->
-            rpc:multicall(Nodes, ?MODULE, release_lock, [Name, LockObj]),
+            rpc:multicall(Nodes, ?MODULE, release_lock, [Name, LockObj], ?MC_TIMEOUT),
             Res
     end.
 
@@ -212,7 +217,7 @@ release(Name, Resource, all) ->
     release_locks(ekka_membership:nodelist(up), Name, lock_obj(Resource)).
 
 release_locks(Nodes, Name, LockObj) ->
-    {ResL, _BadNodes} = rpc:multicall(Nodes, ?MODULE, release_lock, [Name, LockObj]),
+    {ResL, _BadNodes} = rpc:multicall(Nodes, ?MODULE, release_lock, [Name, LockObj], ?MC_TIMEOUT),
     merge_results(ResL).
 
 release_lock(Name, #lock{resource = Resource, owner = Owner}) ->
@@ -264,7 +269,14 @@ handle_info(check_lease, State = #state{locks = Tab, lease = Lease, monitors = M
                   fun(#lock{resource = Resource, owner = Owner}, MonAcc) ->
                       case maps:find(Owner, MonAcc) of
                           {ok, ResourceSet} ->
-                              maps:put(Owner, set_put(Resource, ResourceSet), MonAcc);
+                              case is_set_elem(Resource, ResourceSet) of
+                                  true ->
+                                      %% force kill it as it might have hung
+                                      _ = spawn(fun() -> force_kill_lock_owner(Owner, Resource) end),
+                                      MonAcc;
+                                  false ->
+                                      maps:put(Owner, set_put(Resource, ResourceSet), MonAcc)
+                              end;
                           error ->
                               _MRef = erlang:monitor(process, Owner),
                               maps:put(Owner, set_put(Resource, #{}), MonAcc)
@@ -309,16 +321,33 @@ check_lease(Tab, #lease{expiry = Expiry}, Now) ->
 
 cancel_lease(#lease{timer = TRef}) -> timer:cancel(TRef).
 
-%% TODO: Remove code about list in next version
-set_put(Resource, ResourceSet) when is_list(ResourceSet) ->
-    NewResourceSet = lists:foldl(fun(Resrouce, Acc) ->
-                                     Acc#{Resrouce => nil}
-                                 end, #{}, ResourceSet),
-    set_put(Resource, NewResourceSet);
 set_put(Resource, ResourceSet) when is_map(ResourceSet) ->
     ResourceSet#{Resource => nil}.
 
-set_to_list(ResourceSet) when is_list(ResourceSet) ->
-    ResourceSet;
 set_to_list(ResourceSet) when is_map(ResourceSet) ->
     maps:keys(ResourceSet).
+
+is_set_elem(Resource, ResourceSet) when is_map(ResourceSet) ->
+    maps:is_key(Resource, ResourceSet).
+
+force_kill_lock_owner(Pid, Resource) ->
+    logger:error("kill ~p as it has held the lock for too long, resource: ~p", [Pid, Resource]),
+    Fields = [status, message_queue_len, current_stacktrace],
+    Status = rpc:call(node(Pid), erlang, process_info, [Pid, Fields], 5000),
+    logger:error("lock_owner_status:~n~p", [Status]),
+    _ = exit(Pid, kill),
+    ok.
+
+-ifdef(TEST).
+force_kill_test() ->
+    Pid = spawn(fun() ->
+                        receive
+                            foo ->
+                                ok
+                        end
+                end),
+    ?assert(is_process_alive(Pid)),
+    ok = force_kill_lock_owner(Pid, resource),
+    ?assertNot(is_process_alive(Pid)).
+
+-endif.