Backport e6512fc065d7 (Merge of bug24720; per-queue CPU cost of HA queues, even when idle)

2 files changed, 11 insertions, 15 deletions

diff --git a/src/rabbit_mirror_queue_coordinator.erl b/src/rabbit_mirror_queue_coordinator.erl
index 3e058793..10debb0b 100644
--- a/src/rabbit_mirror_queue_coordinator.erl
+++ b/src/rabbit_mirror_queue_coordinator.erl
@@ -36,8 +36,6 @@
                  length_fun
                }).
 
--define(ONE_SECOND, 1000).
-
 -ifdef(use_specs).
 
 -spec(start_link/4 :: (rabbit_types:amqqueue(), pid() | 'undefined',
@@ -325,7 +323,6 @@ init([#amqqueue { name = QueueName } = Q, GM, DeathFun, LengthFun]) ->
                   true = link(GM),
                   GM
           end,
-    ensure_gm_heartbeat(),
     {ok, #state { q          = Q,
                   gm         = GM1,
                   monitors   = pmon:new(),
@@ -359,11 +356,6 @@ handle_cast({ensure_monitoring, Pids}, State = #state { monitors = Mons }) ->
 handle_cast({delete_and_terminate, Reason}, State) ->
     {stop, Reason, State}.
 
-handle_info(send_gm_heartbeat, State = #state { gm = GM }) ->
-    gm:broadcast(GM, heartbeat),
-    ensure_gm_heartbeat(),
-    noreply(State);
-
 handle_info({'DOWN', _MonitorRef, process, Pid, _Reason},
             State = #state { monitors  = Mons,
                              death_fun = DeathFun }) ->
@@ -399,7 +391,7 @@ members_changed([_CPid], _Births, []) ->
 members_changed([CPid], _Births, Deaths) ->
     ok = gen_server2:cast(CPid, {gm_deaths, Deaths}).
 
-handle_msg([_CPid], _From, heartbeat) ->
+handle_msg([_CPid], _From, master_changed) ->
     ok;
 handle_msg([CPid], _From, request_length = Msg) ->
     ok = gen_server2:cast(CPid, Msg);
@@ -420,6 +412,3 @@ noreply(State) ->
 
 reply(Reply, State) ->
     {reply, Reply, State, hibernate}.
-
-ensure_gm_heartbeat() ->
-    erlang:send_after(?ONE_SECOND, self(), send_gm_heartbeat).
diff --git a/src/rabbit_mirror_queue_slave.erl b/src/rabbit_mirror_queue_slave.erl
index d6811b2f..c4ae307c 100644
--- a/src/rabbit_mirror_queue_slave.erl
+++ b/src/rabbit_mirror_queue_slave.erl
@@ -217,7 +217,12 @@ handle_call({gm_deaths, Deaths}, From,
                     %% master has changed to not us.
                     gen_server2:reply(From, ok),
                     erlang:monitor(process, Pid),
-                    ok = gm:broadcast(GM, heartbeat),
+                    %% GM is lazy. So we know of the death of the
+                    %% slave since it is a neighbour of ours, but
+                    %% until a message is sent, not all members will
+                    %% know. That might include the new master. So
+                    %% broadcast a no-op message to wake everyone up.
+                    ok = gm:broadcast(GM, master_changed),
                     noreply(State #state { master_pid = Pid })
             end
     end;
@@ -359,7 +364,7 @@ members_changed([_SPid], _Births, []) ->
 members_changed([SPid], _Births, Deaths) ->
     inform_deaths(SPid, Deaths).
 
-handle_msg([_SPid], _From, heartbeat) ->
+handle_msg([_SPid], _From, master_changed) ->
     ok;
 handle_msg([_SPid], _From, request_length) ->
     %% This is only of value to the master
@@ -470,7 +475,9 @@ promote_me(From, #state { q                   = Q = #amqqueue { name = QName },
                    rabbit_mirror_queue_master:length_fun()),
     true = unlink(GM),
     gen_server2:reply(From, {promote, CPid}),
-    ok = gm:confirmed_broadcast(GM, heartbeat),
+    %% TODO this has been in here since the beginning, but it's not
+    %% obvious if it is needed. Investigate...
+    ok = gm:confirmed_broadcast(GM, master_changed),
 
     %% Everything that we're monitoring, we need to ensure our new
     %% coordinator is monitoring.