diff options
| -rw-r--r-- | src/rabbit_amqqueue.erl | 2 | ||||
| -rw-r--r-- | src/rabbit_disk_queue.erl | 6 | ||||
| -rw-r--r-- | src/rabbit_mixed_queue.erl | 1 | ||||
| -rw-r--r-- | src/rabbit_queue_mode_manager.erl | 8 |
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) -> |