Flow control for the sync process.

2 files changed, 23 insertions, 5 deletions

diff --git a/src/rabbit_mirror_queue_master.erl b/src/rabbit_mirror_queue_master.erl
index 05075d0f2b..dadaef1de7 100644
--- a/src/rabbit_mirror_queue_master.erl
+++ b/src/rabbit_mirror_queue_master.erl
@@ -155,7 +155,11 @@ sync_mirrors(SPids, Name, #state { backing_queue       = BQ,
     [erlang:demonitor(MRef) || {_, MRef} <- SPidsMRefs],
     {Total, _BQS} =
         BQ:fold(fun (M = #basic_message{}, I) ->
-                        [SPid ! {sync_message, Ref, M} || SPid <- SPids1],
+                        wait_for_credit(),
+                        [begin
+                             credit_flow:send(SPid, ?CREDIT_DISC_BOUND),
+                             SPid ! {sync_message, Ref, M}
+                         end || SPid <- SPids1],
                         case I rem 1000 of
                             0 -> rabbit_log:info(
                                    "Synchronising ~s: ~p messages~n",
@@ -169,6 +173,15 @@ sync_mirrors(SPids, Name, #state { backing_queue       = BQ,
                     [rabbit_misc:rs(Name), Total]),
     ok.
 
+wait_for_credit() ->
+    case credit_flow:blocked() of
+        true  -> receive
+                     {bump_credit, Msg} -> credit_flow:handle_bump_msg(Msg),
+                                           wait_for_credit()
+                 end;
+        false -> ok
+    end.
+
 terminate({shutdown, dropped} = Reason,
           State = #state { backing_queue       = BQ,
                            backing_queue_state = BQS }) ->
diff --git a/src/rabbit_mirror_queue_slave.erl b/src/rabbit_mirror_queue_slave.erl
index ae069898d7..c1d2e8e44f 100644
--- a/src/rabbit_mirror_queue_slave.erl
+++ b/src/rabbit_mirror_queue_slave.erl
@@ -270,7 +270,8 @@ handle_info({sync_start, Ref, MPid},
     MRef = erlang:monitor(process, MPid),
     MPid ! {sync_ready, Ref, self()},
     {_MsgCount, BQS1} = BQ:purge(BQS),
-    noreply(sync_loop(Ref, MRef, State#state{backing_queue_state = BQS1}));
+    noreply(
+      sync_loop(Ref, MRef, MPid, State#state{backing_queue_state = BQS1}));
 
 handle_info(Msg, State) ->
     {stop, {unexpected_info, Msg}, State}.
@@ -839,10 +840,10 @@ record_synchronised(#amqqueue { name = QName }) ->
               end
       end).
 
-sync_loop(Ref, MRef, State = #state{backing_queue       = BQ,
+sync_loop(Ref, MRef, MPid, State = #state{backing_queue       = BQ,
                                     backing_queue_state = BQS}) ->
     receive
-        {'DOWN', MRef, process, _MPid, _Reason} ->
+        {'DOWN', MRef, process, MPid, _Reason} ->
             %% If the master dies half way we are not in the usual
             %% half-synced state (with messages nearer the tail of the
             %% queue; instead we have ones nearer the head. If we then
@@ -850,14 +851,18 @@ sync_loop(Ref, MRef, State = #state{backing_queue       = BQ,
             %% messages from it, we have a hole in the middle. So the
             %% only thing to do here is purge.)
             State#state{backing_queue_state = BQ:purge(BQS)};
+        {bump_credit, Msg} ->
+            credit_flow:handle_bump_msg(Msg),
+            sync_loop(Ref, MRef, MPid, State);
         {sync_complete, Ref} ->
             erlang:demonitor(MRef),
             set_delta(0, State);
         {sync_message, Ref, M} ->
+            credit_flow:ack(MPid, ?CREDIT_DISC_BOUND),
             %% TODO expiry needs fixing
             Props = #message_properties{expiry           = undefined,
                                         needs_confirming = false,
                                         delivered        = true},
             BQS1 = BQ:publish(M, Props, none, BQS),
-            sync_loop(Ref, MRef, State#state{backing_queue_state = BQS1})
+            sync_loop(Ref, MRef, MPid, State#state{backing_queue_state = BQS1})
     end.