Tidying up before starting to play with tokens.

o) Removed redundant rubbish from manager
o) Wired alarms straight to disk_queue (+ minor refactor)
o) Removed reduce/increase_memory_footprint from control and amqqueue

Testing shows that one of the reasons the memory estimates are off is because they only track messages in the queue - i.e. remove on deliver, not on ack. Watching ./scripts/rabbitmqctl list_queues name memory mode messages messages_ready messages_unacknowledged messages_uncommitted during various tests shows that acks are not getting through and so we actually need to track unacknowledged message size - i.e. remove on ack, not on deliver

4 files changed, 45 insertions, 100 deletions

diff --git a/src/rabbit_amqqueue.erl b/src/rabbit_amqqueue.erl
index 4858a984ae..c045b3cae3 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, set_mode/3]).
+-export([set_mode/3]).
 
 -import(mnesia).
 -import(gen_server2).
@@ -104,7 +104,7 @@
 -spec(basic_cancel/4 :: (amqqueue(), pid(), ctag(), any()) -> 'ok').
 -spec(notify_sent/2 :: (pid(), pid()) -> 'ok').
 -spec(unblock/2 :: (pid(), pid()) -> 'ok').
--spec(constrain_memory/2 :: (pid(), bool()) -> 'ok').
+-spec(set_mode/3 :: (vhost(), amqqueue(), ('disk' | 'mixed')) -> 'ok').
 -spec(internal_declare/2 :: (amqqueue(), bool()) -> amqqueue()).
 -spec(internal_delete/1 :: (queue_name()) -> 'ok' | not_found()).
 -spec(on_node_down/1 :: (erlang_node()) -> 'ok').
@@ -323,12 +323,6 @@ notify_sent(QPid, ChPid) ->
 unblock(QPid, ChPid) ->
     gen_server2:pcast(QPid, 10, {unblock, ChPid}).
 
-constrain_memory(QPid, Constrain) ->
-    gen_server2:pcast(QPid, 10, {set_mode, case Constrain of
-                                              true -> disk;
-                                              false -> mixed
-                                           end}).
-
 internal_delete(QueueName) ->
     rabbit_misc:execute_mnesia_transaction(
       fun () ->
diff --git a/src/rabbit_control.erl b/src/rabbit_control.erl
index 291dfae07f..ab5fe1bc6b 100644
--- a/src/rabbit_control.erl
+++ b/src/rabbit_control.erl
@@ -137,9 +137,6 @@ Available commands:
   list_bindings  [-p <VHostPath>] 
   list_connections [<ConnectionInfoItem> ...]
 
-  reduce_memory_footprint
-  increase_memory_footprint
-
   set_queue_mode <QueueName> (disk|mixed)
 
 Quiet output mode is selected with the \"-q\" flag. Informational messages
@@ -281,14 +278,6 @@ action(list_connections, Node, Args, Inform) ->
                                [ArgAtoms]),
                       ArgAtoms);
 
-action(reduce_memory_footprint, Node, _Args, Inform) ->
-    Inform("Reducing memory footprint", []),
-    call(Node, {rabbit_queue_mode_manager, reduce_memory_footprint, []});
-
-action(increase_memory_footprint, Node, _Args, Inform) ->
-    Inform("Increasing memory footprint", []),
-    call(Node, {rabbit_queue_mode_manager, increase_memory_footprint, []});
-
 action(Command, Node, Args, Inform) ->
     {VHost, RemainingArgs} = parse_vhost_flag(Args),
     action(Command, Node, VHost, RemainingArgs, Inform).
diff --git a/src/rabbit_disk_queue.erl b/src/rabbit_disk_queue.erl
index bf2de56529..de31b2388c 100644
--- a/src/rabbit_disk_queue.erl
+++ b/src/rabbit_disk_queue.erl
@@ -46,7 +46,7 @@
 
 -export([length/1, filesync/0, cache_info/0]).
 
--export([stop/0, stop_and_obliterate/0,
+-export([stop/0, stop_and_obliterate/0, change_memory_footprint/2,
          to_disk_only_mode/0, to_ram_disk_mode/0]).
 
 -include("rabbit.hrl").
@@ -269,6 +269,7 @@
 -spec(length/1 :: (queue_name()) -> non_neg_integer()).
 -spec(filesync/0 :: () -> 'ok').
 -spec(cache_info/0 :: () -> [{atom(), term()}]).
+-spec(change_memory_footprint/2 :: (pid(), bool()) -> 'ok').
 
 -endif.
 
@@ -345,6 +346,9 @@ filesync() ->
 cache_info() ->
     gen_server2:call(?SERVER, cache_info, infinity).
 
+change_memory_footprint(_Pid, Conserve) ->
+    gen_server2:pcast(?SERVER, 9, {change_memory_footprint, Conserve}).
+
 %% ---- GEN-SERVER INTERNAL API ----
 
 init([FileSizeLimit, ReadFileHandlesLimit]) ->
@@ -357,6 +361,7 @@ init([FileSizeLimit, ReadFileHandlesLimit]) ->
     %%       brutal_kill.
     %% Otherwise, the gen_server will be immediately terminated.
     process_flag(trap_exit, true),
+    ok = rabbit_alarm:register(self(), {?MODULE, change_memory_footprint, []}),
     Node = node(),
     ok = 
         case mnesia:change_table_copy_type(rabbit_disk_queue, Node,
@@ -457,32 +462,10 @@ handle_call(stop_vaporise, _From, State) ->
      State1 #dqstate { current_file_handle = undefined,
                        read_file_handles = {dict:new(), gb_trees:empty()}}};
     %% gen_server now calls terminate, which then calls shutdown
-handle_call(to_disk_only_mode, _From,
-            State = #dqstate { operation_mode = disk_only }) ->
-    reply(ok, State);
-handle_call(to_disk_only_mode, _From,
-            State = #dqstate { operation_mode = ram_disk,
-                               msg_location_dets = MsgLocationDets,
-                               msg_location_ets = MsgLocationEts }) ->
-    rabbit_log:info("Converting disk queue to disk only mode~n", []),
-    {atomic, ok} = mnesia:change_table_copy_type(rabbit_disk_queue, node(),
-                                                 disc_only_copies),
-    ok = dets:from_ets(MsgLocationDets, MsgLocationEts),
-    true = ets:delete_all_objects(MsgLocationEts),
-    reply(ok, State #dqstate { operation_mode = disk_only });
-handle_call(to_ram_disk_mode, _From,
-            State = #dqstate { operation_mode = ram_disk }) ->
-    reply(ok, State);
-handle_call(to_ram_disk_mode, _From,
-            State = #dqstate { operation_mode = disk_only,
-                               msg_location_dets = MsgLocationDets,
-                               msg_location_ets = MsgLocationEts }) ->
-    rabbit_log:info("Converting disk queue to ram disk mode~n", []),
-    {atomic, ok} = mnesia:change_table_copy_type(rabbit_disk_queue, node(),
-                                                 disc_copies),
-    true = ets:from_dets(MsgLocationEts, MsgLocationDets),
-    ok = dets:delete_all_objects(MsgLocationDets),
-    reply(ok, State #dqstate { operation_mode = ram_disk });
+handle_call(to_disk_only_mode, _From, State) ->
+    reply(ok, to_disk_only_mode(State));
+handle_call(to_ram_disk_mode, _From, State) ->
+    reply(ok, to_ram_disk_mode(State));
 handle_call({length, Q}, _From, State = #dqstate { sequences = Sequences }) ->
     {_ReadSeqId, _WriteSeqId, Length} = sequence_lookup(Sequences, Q),
     reply(Length, State);
@@ -522,7 +505,12 @@ handle_cast({delete_queue, Q}, State) ->
     {ok, State1} = internal_delete_queue(Q, State),
     noreply(State1);
 handle_cast(filesync, State) ->
-    noreply(sync_current_file_handle(State)).
+    noreply(sync_current_file_handle(State));
+handle_cast({change_memory_footprint, Conserve}, State) ->
+    noreply((case Conserve of
+                 true -> fun to_disk_only_mode/1;
+                 false -> fun to_ram_disk_mode/1
+             end)(State)).
 
 handle_info({'EXIT', _Pid, Reason}, State) ->
     {stop, Reason, State};
@@ -563,6 +551,30 @@ code_change(_OldVsn, State, _Extra) ->
 
 %% ---- UTILITY FUNCTIONS ----
 
+to_disk_only_mode(State = #dqstate { operation_mode = disk_only }) ->
+    State;
+to_disk_only_mode(State = #dqstate { operation_mode = ram_disk,
+                                     msg_location_dets = MsgLocationDets,
+                                     msg_location_ets = MsgLocationEts }) ->
+    rabbit_log:info("Converting disk queue to disk only mode~n", []),
+    {atomic, ok} = mnesia:change_table_copy_type(rabbit_disk_queue, node(),
+                                                 disc_only_copies),
+    ok = dets:from_ets(MsgLocationDets, MsgLocationEts),
+    true = ets:delete_all_objects(MsgLocationEts),
+    State #dqstate { operation_mode = disk_only }.
+
+to_ram_disk_mode(State = #dqstate { operation_mode = ram_disk }) ->
+    State;
+to_ram_disk_mode(State = #dqstate { operation_mode = disk_only,
+                                    msg_location_dets = MsgLocationDets,
+                                    msg_location_ets = MsgLocationEts }) ->
+    rabbit_log:info("Converting disk queue to ram disk mode~n", []),
+    {atomic, ok} = mnesia:change_table_copy_type(rabbit_disk_queue, node(),
+                                                 disc_copies),
+    true = ets:from_dets(MsgLocationEts, MsgLocationDets),
+    ok = dets:delete_all_objects(MsgLocationDets),
+    State #dqstate { operation_mode = ram_disk }.
+
 noreply(NewState = #dqstate { current_dirty = true }) ->
     {noreply, start_commit_timer(NewState), 0};
 noreply(NewState) ->
diff --git a/src/rabbit_queue_mode_manager.erl b/src/rabbit_queue_mode_manager.erl
index db174d2449..cc10074c38 100644
--- a/src/rabbit_queue_mode_manager.erl
+++ b/src/rabbit_queue_mode_manager.erl
@@ -38,10 +38,7 @@
 -export([init/1, handle_call/3, handle_cast/2, handle_info/2,
          terminate/2, code_change/3]).
 
--export([register/1, change_memory_footprint/2,
-         reduce_memory_footprint/0, increase_memory_footprint/0,
-         report_memory/2
-        ]).
+-export([register/1, report_memory/2]).
 
 -define(SERVER, ?MODULE).
 
@@ -52,9 +49,6 @@
 -spec(start_link/0 :: () ->
               ({'ok', pid()} | 'ignore' | {'error', any()})).
 -spec(register/1 :: (pid()) -> {'ok', queue_mode()}).
--spec(change_memory_footprint/2 :: (pid(), bool()) -> 'ok').
--spec(reduce_memory_footprint/0 :: () -> 'ok').
--spec(increase_memory_footprint/0 :: () -> 'ok').
 -spec(report_memory/2 :: (pid(), non_neg_integer()) -> 'ok').
 
 -endif.
@@ -69,21 +63,11 @@ start_link() ->
 register(Pid) ->
     gen_server2:call(?SERVER, {register, Pid}).
 
-change_memory_footprint(_Pid, Conserve) ->
-    gen_server2:cast(?SERVER, {change_memory_footprint, Conserve}).
-
-reduce_memory_footprint() ->
-    gen_server2:cast(?SERVER, {change_memory_footprint, true}).
-                           
-increase_memory_footprint() ->
-    gen_server2:cast(?SERVER, {change_memory_footprint, false}).
-
 report_memory(Pid, Memory) ->
     gen_server2:cast(?SERVER, {report_memory, Pid, Memory}).
 
 init([]) ->
     process_flag(trap_exit, true),
-    ok = rabbit_alarm:register(self(), {?MODULE, change_memory_footprint, []}),
     {ok, #state { mode = unlimited,
                   queues = dict:new()
                 }}.
@@ -97,36 +81,8 @@ handle_call({register, Pid}, _From,
              end,
     {reply, {ok, Result}, State #state { queues = dict:store(Pid, 0, Qs) }}.
 
-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 }) ->
-    constrain_queues(true, State #state.queues),
-    {noreply, State #state { mode = disk_only }};
-handle_cast({change_memory_footprint, true},
-            State = #state { mode = unlimited }) ->
-    ok = rabbit_disk_queue:to_disk_only_mode(),
-    {noreply, State #state { mode = ram_disk }};
-
-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 }) ->
-    constrain_queues(false, State #state.queues),
-    {noreply, State #state { mode = ram_disk }};
-
-handle_cast({report_memory, Pid, Memory}, State = #state { queues = Qs }) ->
-    io:format("Queue ~w requested ~w bytes~n", [Pid, Memory]),
-    {noreply, State #state { queues = dict:store(Pid, Memory, Qs) }}.
+handle_cast(_Any, State) ->
+    {noreply, State}.
 
 handle_info({'DOWN', _MRef, process, Pid, _Reason},
             State = #state { queues = Qs }) ->
@@ -141,9 +97,3 @@ terminate(_Reason, State) ->
 
 code_change(_OldVsn, State, _Extra) ->
     {ok, State}.
-
-constrain_queues(Constrain, Qs) ->
-    dict:fold(
-      fun (QPid, _Mem, ok) ->
-              rabbit_amqqueue:constrain_memory(QPid, Constrain)
-      end, ok, Qs).