fixed.

There was a choice here of either pushing all the txn accountancy into the mixed_queue and taking it out of queue_process or just passing in all the txn pending messages to the mode switch. I chose the latter because the queue_process is already the more readable of the two modules and I didn't want to further complicate the mixed_queue. Also, this way is a smaller API change and really not that much code.

Tests pass but I'm about to rewrite the test and bulk it up a bit. Also, running the previous tests - rabbitmq-java-client/build/dist$ sh runjava.sh com/rabbitmq/examples/MulticastMain -y 50 -r 100 -s 104857 -m 100 -z 120 - whilst running (reduce|increase)_memory_footprint is a good thing to do.

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) ->