diff options
| -rw-r--r-- | src/rabbit_amqqueue.erl | 10 | ||||
| -rw-r--r-- | src/rabbit_control.erl | 11 | ||||
| -rw-r--r-- | src/rabbit_disk_queue.erl | 68 | ||||
| -rw-r--r-- | src/rabbit_queue_mode_manager.erl | 56 |
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). |