Port the deadlock-breaking code from bug 25852.

2 files changed, 40 insertions, 30 deletions

diff --git a/src/rabbit_mirror_queue_misc.erl b/src/rabbit_mirror_queue_misc.erl
index 8ad7c62f..47a44278 100644
--- a/src/rabbit_mirror_queue_misc.erl
+++ b/src/rabbit_mirror_queue_misc.erl
@@ -19,7 +19,8 @@
 
 -export([remove_from_queue/3, on_node_up/0, add_mirrors/2, add_mirror/2,
          report_deaths/4, store_updated_slaves/1, suggested_queue_nodes/1,
-         is_mirrored/1, update_mirrors/2, validate_policy/1]).
+         is_mirrored/1, update_mirrors/2, validate_policy/1,
+         maybe_auto_sync/1]).
 
 %% for testing only
 -export([module/1]).
@@ -94,10 +95,14 @@ remove_from_queue(QueueName, Self, LiveGMPids) ->
                               %% Either master hasn't changed, so
                               %% we're ok to update mnesia; or we have
                               %% become the master.
-                              store_updated_slaves(
-                                Q #amqqueue { pid        = QPid1,
+                              Q1 = Q#amqqueue{pid        = QPid1,
                                               slave_pids = SPids1,
-                                              gm_pids    = GMPids1 }),
+                                              gm_pids    = GMPids1},
+                              store_updated_slaves(Q1),
+                              %% If we add and remove nodes at the same time
+                              %% it's possible we need to sync after removing
+                              %% the master. Let's check.
+                              maybe_auto_sync(Q1),
                               {ok, QPid1, [QPid | SPids] -- Alive};
                           _ ->
                               %% Master has changed, and we're not it,
@@ -179,26 +184,15 @@ add_mirror(QName, MirrorNode) ->
               end
       end).
 
-start_child(Name, MirrorNode, Q) ->
+start_child(_Name, MirrorNode, Q) ->
+    %% TODO re-add some log stuff here.
     case rabbit_misc:with_exit_handler(
-           rabbit_misc:const({ok, down}),
+           rabbit_misc:const(down),
            fun () ->
                    rabbit_mirror_queue_slave_sup:start_child(MirrorNode, [Q])
            end) of
-        {ok, SPid} when is_pid(SPid)  ->
-            maybe_auto_sync(Q),
-            rabbit_log:info("Adding mirror of ~s on node ~p: ~p~n",
-                            [rabbit_misc:rs(Name), MirrorNode, SPid]),
-            {ok, started};
-        {error, {{stale_master_pid, StalePid}, _}} ->
-            rabbit_log:warning("Detected stale HA master while adding "
-                               "mirror of ~s on node ~p: ~p~n",
-                               [rabbit_misc:rs(Name), MirrorNode, StalePid]),
-            {ok, stale_master};
-        {error, {{duplicate_live_master, _}=Err, _}} ->
-            Err;
-        Other ->
-            Other
+        {ok, SPid} -> rabbit_mirror_queue_slave:go(SPid);
+        _          -> ok
     end.
 
 report_deaths(_MirrorPid, _IsMaster, _QueueName, []) ->
diff --git a/src/rabbit_mirror_queue_slave.erl b/src/rabbit_mirror_queue_slave.erl
index 6f78d1d2..aa69aeed 100644
--- a/src/rabbit_mirror_queue_slave.erl
+++ b/src/rabbit_mirror_queue_slave.erl
@@ -24,7 +24,7 @@
 %% All instructions from the GM group must be processed in the order
 %% in which they're received.
 
--export([start_link/1, set_maximum_since_use/2, info/1]).
+-export([start_link/1, set_maximum_since_use/2, info/1, go/1]).
 
 -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2,
          code_change/3, handle_pre_hibernate/1, prioritise_call/4,
@@ -78,7 +78,14 @@ set_maximum_since_use(QPid, Age) ->
 
 info(QPid) -> gen_server2:call(QPid, info, infinity).
 
-init(Q = #amqqueue { name = QName }) ->
+init(Q) ->
+    {ok, {not_started, Q}, hibernate,
+     {backoff, ?HIBERNATE_AFTER_MIN, ?HIBERNATE_AFTER_MIN,
+      ?DESIRED_HIBERNATE}}.
+
+go(SPid) -> gen_server2:cast(SPid, go).
+
+handle_go({not_started, Q = #amqqueue { name = QName }} = NotStarted) ->
     %% We join the GM group before we add ourselves to the amqqueue
     %% record. As a result:
     %% 1. We can receive msgs from GM that correspond to messages we will
@@ -124,22 +131,24 @@ init(Q = #amqqueue { name = QName }) ->
                    },
             ok = gm:broadcast(GM, request_depth),
             ok = gm:validate_members(GM, [GM | [G || {G, _} <- GMPids]]),
-            {ok, State, hibernate,
-             {backoff, ?HIBERNATE_AFTER_MIN, ?HIBERNATE_AFTER_MIN,
-              ?DESIRED_HIBERNATE}};
+            rabbit_mirror_queue_misc:maybe_auto_sync(Q1),
+            {noreply, State};
         {stale, StalePid} ->
-            {stop, {stale_master_pid, StalePid}};
+            gm:leave(GM),
+            {stop, {stale_master_pid, StalePid}, NotStarted};
         duplicate_live_master ->
-            {stop, {duplicate_live_master, Node}};
+            gm:leave(GM),
+            {stop, {duplicate_live_master, Node}, NotStarted};
         existing ->
             gm:leave(GM),
-            ignore;
+            {stop, normal, NotStarted};
         master_in_recovery ->
+            gm:leave(GM),
             %% The queue record vanished - we must have a master starting
             %% concurrently with us. In that case we can safely decide to do
             %% nothing here, and the master will start us in
             %% master:init_with_existing_bq/3
-            ignore
+            {stop, normal, NotStarted}
     end.
 
 init_it(Self, GM, Node, QName) ->
@@ -208,6 +217,9 @@ handle_call({gm_deaths, LiveGMPids}, From,
 handle_call(info, _From, State) ->
     reply(infos(?INFO_KEYS, State), State).
 
+handle_cast(go, {not_started, _Q} = NotStarted) ->
+    handle_go(NotStarted);
+
 handle_cast({run_backing_queue, Mod, Fun}, State) ->
     noreply(run_backing_queue(Mod, Fun, State));
 
@@ -293,6 +305,8 @@ handle_info({bump_credit, Msg}, State) ->
 handle_info(Msg, State) ->
     {stop, {unexpected_info, Msg}, State}.
 
+terminate(normal, {not_started, _Q}) ->
+    ok;
 terminate(_Reason, #state { backing_queue_state = undefined }) ->
     %% We've received a delete_and_terminate from gm, thus nothing to
     %% do here.
@@ -403,7 +417,9 @@ handle_msg([SPid], _From, Msg) ->
     ok = gen_server2:cast(SPid, {gm, Msg}).
 
 inform_deaths(SPid, Live) ->
-    case gen_server2:call(SPid, {gm_deaths, Live}, infinity) of
+    case rabbit_misc:with_exit_handler(
+           rabbit_misc:const(ok),
+           fun() -> gen_server2:call(SPid, {gm_deaths, Live}, infinity) end) of
         ok              -> ok;
         {promote, CPid} -> {become, rabbit_mirror_queue_coordinator, [CPid]}
     end.