Took advantage of the gen_server2 priorities.

Reversed order  - i.e. now when swapping out, the first thing is to alter the disk_queue, and the 2nd thing is to alter the queues.
And vice versa.

The reasoning is as follows:
Changing the disk_queue is a BIG operation because it affects every message in there, from all queues. In order to minimise the impa
ct of this operation, we must do it first, not second, because if we do it first, only persistent messages from durable queues will
be in there, whereas if we do it second, then all messages from all queues will be in there.
Similarly, when swapping in, altering the individual queues is the first thing to do because it prevents the disk queue from growing
 further (i.e. only persistent messages to durable queues then make it to the disk queue), and each queue pulls out from the disk qu
eue all the messages in there and so subsequent delivery from the mixed queue becomes very fast (actually, this is a total lie because of the call to rabbit_disk_queue:phantom_deliver in rabbit_mixed_queue:deliver - if I could get rid of this or at least make it async then that would greatly improve matters).

4 files changed, 9 insertions, 8 deletions

diff --git a/src/rabbit_amqqueue.erl b/src/rabbit_amqqueue.erl
index 97ffcda8e6..a5c58f2358 100644
--- a/src/rabbit_amqqueue.erl
+++ b/src/rabbit_amqqueue.erl
@@ -315,7 +315,7 @@ unblock(QPid, ChPid) ->
     gen_server2:cast(QPid, {unblock, ChPid}).
 
 constrain_memory(QPid, Constrain) ->
-    gen_server2:cast(QPid, {constrain, Constrain}).
+    gen_server2:pcast(QPid, 10, {constrain, Constrain}).
 
 internal_delete(QueueName) ->
     rabbit_misc:execute_mnesia_transaction(
diff --git a/src/rabbit_disk_queue.erl b/src/rabbit_disk_queue.erl
index 1b30051f30..3fc208df7e 100644
--- a/src/rabbit_disk_queue.erl
+++ b/src/rabbit_disk_queue.erl
@@ -315,7 +315,7 @@ delete_queue(Q) ->
     gen_server2:cast(?SERVER, {delete_queue, Q}).
 
 dump_queue(Q) ->
-    gen_server2:call(?SERVER, {dump_queue, Q}, infinity).
+    gen_server2:pcall(?SERVER, {dump_queue, Q}, infinity).
 
 delete_non_durable_queues(DurableQueues) ->
     gen_server2:call(?SERVER, {delete_non_durable_queues, DurableQueues}, infinity).
@@ -327,10 +327,10 @@ stop_and_obliterate() ->
     gen_server2:call(?SERVER, stop_vaporise, infinity).
 
 to_disk_only_mode() ->
-    gen_server2:call(?SERVER, to_disk_only_mode, infinity).
+    gen_server2:pcall(?SERVER, 10, to_disk_only_mode, infinity).
 
 to_ram_disk_mode() ->
-    gen_server2:call(?SERVER, to_ram_disk_mode, infinity).
+    gen_server2:pcall(?SERVER, 10, to_ram_disk_mode, infinity).
 
 length(Q) ->
     gen_server2:call(?SERVER, {length, Q}, infinity).
diff --git a/src/rabbit_mixed_queue.erl b/src/rabbit_mixed_queue.erl
index 6a463242be..8aedc9eb9f 100644
--- a/src/rabbit_mixed_queue.erl
+++ b/src/rabbit_mixed_queue.erl
@@ -103,6 +103,7 @@ to_mixed_mode(State = #mqstate { mode = disk, queue = Q }) ->
     %% load up a new queue with everything that's on disk.
     %% don't remove non-persistent messages that happen to be on disk
     QList = rabbit_disk_queue:dump_queue(Q),
+    rabbit_log:info("Queue length: ~p ~w~n", [Q, erlang:length(QList)]),
     {MsgBuf1, NextSeq1} =
         lists:foldl(
           fun ({MsgId, MsgBin, _Size, IsDelivered, _AckTag, SeqId}, {Buf, NSeq})
diff --git a/src/rabbit_queue_mode_manager.erl b/src/rabbit_queue_mode_manager.erl
index aee57ac3b7..080607bbad 100644
--- a/src/rabbit_queue_mode_manager.erl
+++ b/src/rabbit_queue_mode_manager.erl
@@ -72,19 +72,19 @@ handle_call({register, Pid}, _From, State = #state { queues = Qs, mode = Mode })
 handle_cast({change_memory_usage, true}, State = #state { mode = disk_only }) ->
     {noreply, State};
 handle_cast({change_memory_usage, true}, State = #state { mode = ram_disk }) ->
-    ok = rabbit_disk_queue:to_disk_only_mode(),
+    constrain_queues(true, State #state.queues),
     {noreply, State #state { mode = disk_only }};
 handle_cast({change_memory_usage, true}, State = #state { mode = unlimited }) ->
-    constrain_queues(true, State #state.queues),
+    ok = rabbit_disk_queue:to_disk_only_mode(),
     {noreply, State #state { mode = ram_disk }};
 
 handle_cast({change_memory_usage, false}, State = #state { mode = unlimited }) ->
     {noreply, State};
 handle_cast({change_memory_usage, false}, State = #state { mode = ram_disk }) ->
-    constrain_queues(false, State #state.queues),
+    ok = rabbit_disk_queue:to_ram_disk_mode(),
     {noreply, State #state { mode = unlimited }};
 handle_cast({change_memory_usage, false}, State = #state { mode = disk_only }) ->
-    ok = rabbit_disk_queue:to_ram_disk_mode(),
+    constrain_queues(false, State #state.queues),
     {noreply, State #state { mode = ram_disk }}.
 
 handle_info({'EXIT', _Pid, Reason}, State) ->