diff options
| -rw-r--r-- | src/rabbit_amqqueue_process.erl | 9 | ||||
| -rw-r--r-- | src/rabbit_mixed_queue.erl | 44 | ||||
| -rw-r--r-- | src/rabbit_tests.erl | 18 |
3 files changed, 53 insertions, 18 deletions
diff --git a/src/rabbit_amqqueue_process.erl b/src/rabbit_amqqueue_process.erl index 6869846ddc..9fe6f50d38 100644 --- a/src/rabbit_amqqueue_process.erl +++ b/src/rabbit_amqqueue_process.erl @@ -790,10 +790,13 @@ handle_cast({limit, ChPid, LimiterPid}, State) -> end)); handle_cast({constrain, Constrain}, 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/1; - false -> fun rabbit_mixed_queue:to_mixed_mode/1 - end)(MS), + true -> fun rabbit_mixed_queue:to_disk_only_mode/2; + false -> fun rabbit_mixed_queue:to_mixed_mode/2 + end)(PendingMessages, MS), noreply(State #q { mixed_state = MS1 }). handle_info({'DOWN', MonitorRef, process, DownPid, _Reason}, diff --git a/src/rabbit_mixed_queue.erl b/src/rabbit_mixed_queue.erl index 9b99ab7f90..bb9b90a3d7 100644 --- a/src/rabbit_mixed_queue.erl +++ b/src/rabbit_mixed_queue.erl @@ -39,7 +39,7 @@ 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/1, to_mixed_mode/1]). +-export([to_disk_only_mode/2, to_mixed_mode/2]). -record(mqstate, { mode, msg_buf, @@ -80,6 +80,9 @@ -spec(length/1 :: (mqstate()) -> non_neg_integer()). -spec(is_empty/1 :: (mqstate()) -> bool()). +-spec(to_disk_only_mode/2 :: ([message()], mqstate()) -> okmqs()). +-spec(to_mixed_mode/2 :: ([message()], mqstate()) -> okmqs()). + -endif. init(Queue, IsDurable, disk) -> @@ -88,12 +91,13 @@ init(Queue, IsDurable, disk) -> is_durable = IsDurable, length = 0 }); init(Queue, IsDurable, mixed) -> {ok, State} = init(Queue, IsDurable, disk), - to_mixed_mode(State). + to_mixed_mode([], State). -to_disk_only_mode(State = #mqstate { mode = disk }) -> +to_disk_only_mode(_TxnMessages, State = #mqstate { mode = disk }) -> {ok, State}; -to_disk_only_mode(State = - #mqstate { mode = mixed, queue = Q, msg_buf = MsgBuf }) -> +to_disk_only_mode(TxnMessages, State = + #mqstate { mode = mixed, queue = Q, msg_buf = MsgBuf, + is_durable = IsDurable }) -> rabbit_log:info("Converting queue to disk only mode: ~p~n", [Q]), %% We enqueue _everything_ here. This means that should a message %% already be in the disk queue we must remove it and add it back @@ -125,11 +129,24 @@ to_disk_only_mode(State = true -> rabbit_disk_queue:requeue_with_seqs(Q, lists:reverse(Requeue)) end, + %% tx_publish txn messages. Some of these will have been already + %% published if they really are durable and persistent which is + %% why we can't just use our own tx_publish/2 function (would end + %% up publishing twice, so refcount would go wrong in disk_queue). + lists:foreach( + fun (Msg = #basic_message { is_persistent = IsPersistent }) -> + ok = case IsDurable andalso IsPersistent of + true -> ok; + _ -> rabbit_disk_queue:tx_publish(Msg) + end + end, TxnMessages), {ok, State #mqstate { mode = disk, msg_buf = queue:new() }}. -to_mixed_mode(State = #mqstate { mode = mixed }) -> +to_mixed_mode(_TxnMessages, State = #mqstate { mode = mixed }) -> {ok, State}; -to_mixed_mode(State = #mqstate { mode = disk, queue = Q, length = Length }) -> +to_mixed_mode(TxnMessages, State = + #mqstate { mode = disk, queue = Q, length = Length, + is_durable = IsDurable }) -> rabbit_log:info("Converting queue to mixed mode: ~p~n", [Q]), %% load up a new queue with everything that's on disk. %% don't remove non-persistent messages that happen to be on disk @@ -140,6 +157,19 @@ to_mixed_mode(State = #mqstate { mode = disk, queue = Q, length = Length }) -> {Buf, L}) -> {queue:in({Msg, IsDelivered, true}, Buf), L+1} end, {queue:new(), 0}, QList), + %% remove txn messages from disk which are neither persistent and + %% durable. This is necessary to avoid leaks. This is also pretty + %% much the inverse behaviour of our own tx_cancel/2 which is why + %% we're not using it. + Cancel = + lists:foldl( + fun (Msg = #basic_message { is_persistent = IsPersistent }, Acc) -> + case IsDurable andalso IsPersistent of + true -> Acc; + _ -> [Msg #basic_message.guid | Acc] + end + end, [], TxnMessages), + ok = rabbit_disk_queue:tx_cancel(lists:reverse(Cancel)), {ok, State #mqstate { mode = mixed, msg_buf = MsgBuf1 }}. purge_non_persistent_messages(State = #mqstate { mode = disk, queue = Q, diff --git a/src/rabbit_tests.erl b/src/rabbit_tests.erl index f53ce6e6bc..37c0121cda 100644 --- a/src/rabbit_tests.erl +++ b/src/rabbit_tests.erl @@ -1003,10 +1003,10 @@ rdq_test_mixed_queue_modes() -> end, MS4, lists:seq(1,10)), 30 = rabbit_mixed_queue:length(MS6), io:format("Published a mixture of messages~n"), - {ok, MS7} = rabbit_mixed_queue:to_disk_only_mode(MS6), + {ok, MS7} = rabbit_mixed_queue:to_disk_only_mode([], MS6), 30 = rabbit_mixed_queue:length(MS7), io:format("Converted to disk only mode~n"), - {ok, MS8} = rabbit_mixed_queue:to_mixed_mode(MS7), + {ok, MS8} = rabbit_mixed_queue:to_mixed_mode([], MS7), 30 = rabbit_mixed_queue:length(MS8), io:format("Converted to mixed mode~n"), MS10 = @@ -1020,7 +1020,7 @@ rdq_test_mixed_queue_modes() -> end, MS8, lists:seq(1,10)), 20 = rabbit_mixed_queue:length(MS10), io:format("Delivered initial non persistent messages~n"), - {ok, MS11} = rabbit_mixed_queue:to_disk_only_mode(MS10), + {ok, MS11} = rabbit_mixed_queue:to_disk_only_mode([], MS10), 20 = rabbit_mixed_queue:length(MS11), io:format("Converted to disk only mode~n"), rdq_stop(), @@ -1040,7 +1040,7 @@ rdq_test_mixed_queue_modes() -> 0 = rabbit_mixed_queue:length(MS14), {ok, MS15} = rabbit_mixed_queue:ack(AckTags, MS14), io:format("Delivered and acked all messages~n"), - {ok, MS16} = rabbit_mixed_queue:to_disk_only_mode(MS15), + {ok, MS16} = rabbit_mixed_queue:to_disk_only_mode([], MS15), 0 = rabbit_mixed_queue:length(MS16), io:format("Converted to disk only mode~n"), rdq_stop(), @@ -1064,8 +1064,9 @@ rdq_test_mode_conversion_mid_txn() -> {ok, MS1a} = rabbit_mixed_queue:tx_publish(Msg, MS1), {MS1a, [Msg | Acc]} end, {MS, []}, MsgIds), - {ok, MS3} = rabbit_mixed_queue:to_disk_only_mode(MS2), - {ok, MS4} = rabbit_mixed_queue:tx_commit(lists:reverse(Msgs), [], MS3), + MsgsOrdered = lists:reverse(Msgs), + {ok, MS3} = rabbit_mixed_queue:to_disk_only_mode(MsgsOrdered, MS2), + {ok, MS4} = rabbit_mixed_queue:tx_commit(MsgsOrdered, [], MS3), MS6 = lists:foldl( fun (N, MS5) -> @@ -1084,8 +1085,9 @@ rdq_test_mode_conversion_mid_txn() -> {ok, MS8a} = rabbit_mixed_queue:tx_publish(Msg, MS8), {MS8a, [Msg | Acc]} end, {MS7, []}, MsgIds), - {ok, MS10} = rabbit_mixed_queue:to_mixed_mode(MS9), - {ok, MS11} = rabbit_mixed_queue:tx_commit(lists:reverse(Msgs1), [], MS10), + Msgs1Ordered = lists:reverse(Msgs1), + {ok, MS10} = rabbit_mixed_queue:to_mixed_mode(Msgs1Ordered, MS9), + {ok, MS11} = rabbit_mixed_queue:tx_commit(Msgs1Ordered, [], MS10), MS13 = lists:foldl( fun (N, MS12) -> |