Added ability to manually set queue modes. Also altered manager so that if you are in all mixed mode, and then set one queue to disk, and then call increase_memory_footprint, you will indeed set that one queue back to mixed mode. And vice versa with mixed and decrease

5 files changed, 39 insertions, 10 deletions

diff --git a/src/rabbit_amqqueue.erl b/src/rabbit_amqqueue.erl
index 9d3cead654..05a32a093c 100644
--- a/src/rabbit_amqqueue.erl
+++ b/src/rabbit_amqqueue.erl
@@ -42,7 +42,7 @@
 -export([notify_sent/2, unblock/2]).
 -export([commit_all/2, rollback_all/2, notify_down_all/2, limit_all/3]).
 -export([on_node_down/1]).
--export([constrain_memory/2]).
+-export([constrain_memory/2, set_mode/3]).
 
 -import(mnesia).
 -import(gen_server2).
@@ -225,6 +225,12 @@ list(VHostPath) ->
 
 map(VHostPath, F) -> rabbit_misc:filter_exit_map(F, list(VHostPath)).
 
+set_mode(VHostPath, Queue, ModeBin)
+  when is_binary(VHostPath) andalso is_binary(Queue) ->
+    Mode = list_to_atom(binary_to_list(ModeBin)),
+    with(rabbit_misc:r(VHostPath, queue, Queue),
+         fun(Q) -> gen_server2:cast(Q #amqqueue.pid, {set_mode, Mode}) end).
+
 info(#amqqueue{ pid = QPid }) ->
     gen_server2:pcall(QPid, 9, info, infinity).
 
@@ -318,7 +324,10 @@ unblock(QPid, ChPid) ->
     gen_server2:pcast(QPid, 10, {unblock, ChPid}).
 
 constrain_memory(QPid, Constrain) ->
-    gen_server2:pcast(QPid, 10, {constrain, Constrain}).
+    gen_server2:pcast(QPid, 10, {set_mode, case Constrain of
+                                              true -> disk;
+                                              false -> mixed
+                                           end}).
 
 internal_delete(QueueName) ->
     rabbit_misc:execute_mnesia_transaction(
diff --git a/src/rabbit_amqqueue_process.erl b/src/rabbit_amqqueue_process.erl
index 2245e12f92..5e607a4636 100644
--- a/src/rabbit_amqqueue_process.erl
+++ b/src/rabbit_amqqueue_process.erl
@@ -89,7 +89,9 @@
          acks_uncommitted,
          consumers,
          transactions,
-         memory]).
+         memory,
+         mode
+        ]).
          
 %%----------------------------------------------------------------------------
 
@@ -512,6 +514,8 @@ i(name,        #q{q = #amqqueue{name        = Name}})       -> Name;
 i(durable,     #q{q = #amqqueue{durable     = Durable}})    -> Durable;
 i(auto_delete, #q{q = #amqqueue{auto_delete = AutoDelete}}) -> AutoDelete;
 i(arguments,   #q{q = #amqqueue{arguments   = Arguments}})  -> Arguments;
+i(mode,        #q{ mixed_state = MS })                      ->
+         rabbit_mixed_queue:info(MS);
 i(pid, _) ->
     self();
 i(messages_ready, #q { mixed_state = MS }) ->
@@ -821,15 +825,16 @@ handle_cast({limit, ChPid, LimiterPid}, State) ->
                 C#cr{limiter_pid = LimiterPid, is_limit_active = NewLimited}
         end));
 
-handle_cast({constrain, Constrain}, State = #q { mixed_state = MS }) ->
+handle_cast({set_mode, Mode}, State = #q { mixed_state = MS }) ->
     PendingMessages =
         lists:flatten([Pending || #tx { pending_messages = Pending}
                                       <- all_tx_record()]),
-    {ok, MS1} = (case Constrain of
-                    true  -> fun rabbit_mixed_queue:to_disk_only_mode/2;
-                    false -> fun rabbit_mixed_queue:to_mixed_mode/2
+    {ok, MS1} = (case Mode of
+                    disk  -> fun rabbit_mixed_queue:to_disk_only_mode/2;
+                    mixed -> fun rabbit_mixed_queue:to_mixed_mode/2
                  end)(PendingMessages, MS),
     noreply(State #q { mixed_state = MS1 }).
+                                                 
 
 handle_info({'DOWN', MonitorRef, process, DownPid, _Reason},
             State = #q{owner = {DownPid, MonitorRef}}) ->
diff --git a/src/rabbit_control.erl b/src/rabbit_control.erl
index 9c1553b874..291dfae07f 100644
--- a/src/rabbit_control.erl
+++ b/src/rabbit_control.erl
@@ -140,6 +140,8 @@ Available commands:
   reduce_memory_footprint
   increase_memory_footprint
 
+  set_queue_mode <QueueName> (disk|mixed)
+
 Quiet output mode is selected with the \"-q\" flag. Informational messages
 are suppressed when quiet mode is in effect.
 
@@ -155,8 +157,8 @@ virtual host parameter for which to display results. The default value is \"/\".
 
 <QueueInfoItem> must be a member of the list [name, durable, auto_delete, 
 arguments, node, messages_ready, messages_unacknowledged, messages_uncommitted, 
-messages, acks_uncommitted, consumers, transactions, memory]. The default is 
- to display name and (number of) messages.
+messages, acks_uncommitted, consumers, transactions, memory, mode]. The default 
+is to display name and (number of) messages.
 
 <ExchangeInfoItem> must be a member of the list [name, type, durable, 
 auto_delete, arguments]. The default is to display name and type.
@@ -291,6 +293,11 @@ action(Command, Node, Args, Inform) ->
     {VHost, RemainingArgs} = parse_vhost_flag(Args),
     action(Command, Node, VHost, RemainingArgs, Inform).
 
+action(set_queue_mode, Node, VHost, [Queue, Mode], Inform) ->
+    Inform("Setting queue mode to ~p for queue ~p in vhost ~p",
+           [Mode, Queue, VHost]),
+    call(Node, {rabbit_amqqueue, set_mode, [VHost, Queue, Mode]});
+    
 action(set_permissions, Node, VHost, [Username, CPerm, WPerm, RPerm], Inform) ->
     Inform("Setting permissions for user ~p in vhost ~p", [Username, VHost]),
     call(Node, {rabbit_access_control, set_permissions,
diff --git a/src/rabbit_mixed_queue.erl b/src/rabbit_mixed_queue.erl
index 5c00b38005..e700d3d220 100644
--- a/src/rabbit_mixed_queue.erl
+++ b/src/rabbit_mixed_queue.erl
@@ -39,7 +39,8 @@
          tx_publish/2, tx_commit/3, tx_cancel/2, requeue/2, purge/1,
          length/1, is_empty/1, delete_queue/1]).
 
--export([to_disk_only_mode/2, to_mixed_mode/2, estimate_queue_memory/1]).
+-export([to_disk_only_mode/2, to_mixed_mode/2, estimate_queue_memory/1,
+         info/1]).
 
 -record(mqstate, { mode,
                    msg_buf,
@@ -86,6 +87,7 @@
 -spec(to_mixed_mode/2 :: ([message()], mqstate()) -> okmqs()).
 
 -spec(estimate_queue_memory/1 :: (mqstate()) -> non_neg_integer).
+-spec(info/1 :: (mqstate()) -> mode()).
 
 -endif.
 
@@ -455,3 +457,6 @@ is_empty(#mqstate { length = Length }) ->
 
 estimate_queue_memory(#mqstate { memory_size = Size }) ->
     2 * Size. %% Magic number. Will probably need playing with.
+
+info(#mqstate { mode = Mode }) ->
+    Mode.
diff --git a/src/rabbit_queue_mode_manager.erl b/src/rabbit_queue_mode_manager.erl
index d3a753a63d..db174d2449 100644
--- a/src/rabbit_queue_mode_manager.erl
+++ b/src/rabbit_queue_mode_manager.erl
@@ -99,6 +99,7 @@ handle_call({register, Pid}, _From,
 
 handle_cast({change_memory_footprint, true},
             State = #state { mode = disk_only }) ->
+    constrain_queues(true, State #state.queues),
     {noreply, State};
 handle_cast({change_memory_footprint, true},
             State = #state { mode = ram_disk }) ->
@@ -111,10 +112,12 @@ handle_cast({change_memory_footprint, true},
 
 handle_cast({change_memory_footprint, false},
             State = #state { mode = unlimited }) ->
+    constrain_queues(false, State #state.queues),
     {noreply, State};
 handle_cast({change_memory_footprint, false},
             State = #state { mode = ram_disk }) ->
     ok = rabbit_disk_queue:to_ram_disk_mode(),
+    constrain_queues(false, State #state.queues),
     {noreply, State #state { mode = unlimited }};
 handle_cast({change_memory_footprint, false},
             State = #state { mode = disk_only }) ->