Merge pull request #1647 from rabbitmq/cancel-sync-slaves

Cancel synchronisation when memory alarm is triggered

Author: michaelklishin

src/rabbit_mirror_queue_master.erl

@@ -165,6 +165,8 @@ sync_mirrors(HandleInfo, EmitStats,
     S = fun(BQSN) -> State#state{backing_queue_state = BQSN} end,
     case rabbit_mirror_queue_sync:master_go(
            Syncer, Ref, Log, HandleInfo, EmitStats, SyncBatchSize, BQ, BQS) of
+        {cancelled, BQS1}      -> Log(" synchronisation cancelled ", []),
+                                  {ok, S(BQS1)};
         {shutdown,  R, BQS1}   -> {stop, R, S(BQS1)};
         {sync_died, R, BQS1}   -> Log("~p", [R]),
                                   {ok, S(BQS1)};

src/rabbit_mirror_queue_sync.erl

@@ -150,10 +150,20 @@ master_send_receive(SyncMsg, NewAcc, Syncer, Ref, Parent) ->
 
 master_done({Syncer, Ref, _Log, _HandleInfo, _EmitStats, Parent}, BQS) ->
     receive
-        {next, Ref}              -> stop_syncer(Syncer, {done, Ref}),
-                                    {ok, BQS};
-        {'EXIT', Parent, Reason} -> {shutdown,  Reason, BQS};
-        {'EXIT', Syncer, Reason} -> {sync_died, Reason, BQS}
+        {'$gen_call', From,
+         cancel_sync_mirrors}    ->
+            stop_syncer(Syncer, {cancel, Ref}),
+            gen_server2:reply(From, ok),
+            {cancelled, BQS};
+        {cancelled, Ref} ->
+            {cancelled, BQS};
+        {next, Ref}              ->
+            stop_syncer(Syncer, {done, Ref}),
+            {ok, BQS};
+        {'EXIT', Parent, Reason} ->
+            {shutdown,  Reason, BQS};
+        {'EXIT', Syncer, Reason} ->
+            {sync_died, Reason, BQS}
     end.
 
 stop_syncer(Syncer, Msg) ->
@@ -230,7 +240,7 @@ syncer_check_resources(Ref, MPid, SPids) ->
             syncer_loop(Ref, MPid, SPids);
         true ->
             case wait_for_resources(Ref, SPids) of
-                cancel -> ok;
+                cancel -> MPid ! {cancelled, Ref};
                 SPids1 -> MPid ! {next, Ref},
                           syncer_loop(Ref, MPid, SPids1)
             end
@@ -240,7 +250,7 @@ syncer_loop(Ref, MPid, SPids) ->
     receive
         {conserve_resources, memory, true} ->
             case wait_for_resources(Ref, SPids) of
-                cancel -> ok;
+                cancel -> MPid ! {cancelled, Ref};
                 SPids1 -> syncer_loop(Ref, MPid, SPids1)
             end;
         {conserve_resources, _, _} ->