Use rabbit_queue_behaviour and callback

Alternative for #1388 that does not use process dictionary.

Requires rabbitmq/rabbitmq-common#228

Fix waiting_bump values

Author: lukebakken

src/rabbit_amqqueue_process.erl

@@ -1454,10 +1454,10 @@ handle_info({bump_credit, Msg}, State = #q{backing_queue       = BQ,
     %% rabbit_variable_queue:msg_store_write/4.
     credit_flow:handle_bump_msg(Msg),
     noreply(State#q{backing_queue_state = BQ:resume(BQS)});
-handle_info(bump_reduce_memory_use, State = #q{backing_queue       = BQ,
-                                               backing_queue_state = BQS}) ->
-    put(waiting_bump, false),
-    noreply(State#q{backing_queue_state = BQ:resume(BQS)});
+handle_info(bump_reduce_memory_use, State = #q{ backing_queue       = BQ,
+                                                backing_queue_state = BQS0 }) ->
+    BQS1 = BQ:handled_bump_reduce_memory_use(BQS0),
+    noreply(State#q{ backing_queue_state = BQ:resume(BQS1) });
 
 handle_info(Info, State) ->
     {stop, {unhandled_info, Info}, State}.

src/rabbit_mirror_queue_master.erl

@@ -24,7 +24,7 @@
          dropwhile/2, fetchwhile/4, set_ram_duration_target/2, ram_duration/1,
          needs_timeout/1, timeout/1, handle_pre_hibernate/1, resume/1,
          msg_rates/1, info/2, invoke/3, is_duplicate/2, set_queue_mode/2,
-         zip_msgs_and_acks/4]).
+         zip_msgs_and_acks/4, handled_bump_reduce_memory_use/1]).
 
 -export([start/2, stop/1, delete_crashed/1]).
 
@@ -525,6 +525,11 @@ zip_msgs_and_acks(Msgs, AckTags, Accumulator,
                            backing_queue_state = BQS }) ->
     BQ:zip_msgs_and_acks(Msgs, AckTags, Accumulator, BQS).
 
+handled_bump_reduce_memory_use(State = #state{ backing_queue       = BQ,
+                                               backing_queue_state = BQS0 }) ->
+    BQS1 = BQ:handled_bump_reduce_memory_use(BQS0),
+    State#state{ backing_queue_state = BQS1 }.
+
 %% ---------------------------------------------------------------------------
 %% Other exported functions
 %% ---------------------------------------------------------------------------

src/rabbit_priority_queue.erl

@@ -41,7 +41,7 @@
          set_ram_duration_target/2, ram_duration/1, needs_timeout/1, timeout/1,
          handle_pre_hibernate/1, resume/1, msg_rates/1,
          info/2, invoke/3, is_duplicate/2, set_queue_mode/2,
-         zip_msgs_and_acks/4]).
+         zip_msgs_and_acks/4, handled_bump_reduce_memory_use/1]).
 
 -record(state, {bq, bqss, max_priority}).
 -record(passthrough, {bq, bqs}).
@@ -447,6 +447,9 @@ zip_msgs_and_acks(Msgs, AckTags, Accumulator,
                   #passthrough{bq = BQ, bqs = BQS}) ->
     BQ:zip_msgs_and_acks(Msgs, AckTags, Accumulator, BQS).
 
+handled_bump_reduce_memory_use(State = #passthrough{ bq = BQ, bqs = BQS }) ->
+    ?passthrough1(handled_bump_reduce_memory_use(BQS)).
+
 %%----------------------------------------------------------------------------
 
 bq() ->

src/rabbit_variable_queue.erl

@@ -26,7 +26,8 @@
          set_ram_duration_target/2, ram_duration/1, needs_timeout/1, timeout/1,
          handle_pre_hibernate/1, resume/1, msg_rates/1,
          info/2, invoke/3, is_duplicate/2, set_queue_mode/2,
-         zip_msgs_and_acks/4,  multiple_routing_keys/0]).
+         zip_msgs_and_acks/4,  multiple_routing_keys/0,
+         handled_bump_reduce_memory_use/1]).
 
 -export([start/2, stop/1]).
 
@@ -325,7 +326,8 @@
           memory_reduction_run_count,
           %% Queue data is grouped by VHost. We need to store it
           %% to work with queue index.
-          virtual_host
+          virtual_host,
+          waiting_bump = false
         }).
 
 -record(rates, { in, out, ack_in, ack_out, timestamp }).
@@ -2466,21 +2468,16 @@ reduce_memory_use(State = #vqstate {
     Blocked = credit_flow:blocked(),
     case {Blocked, NeedResumeA2B orelse NeedResumeB2D} of
         %% Credit bump will continue paging
-        {true, _}      -> ok;
+        {true, _}      -> State3;
         %% Finished with paging
-        {false, false} -> ok;
+        {false, false} -> State3;
         %% Planning next batch
         {false, true}  ->
             %% We don't want to use self-credit-flow, because it's harder to
             %% reason about. So the process sends a (prioritised) message to
             %% itself and sets a waiting_bump value to keep the message box clean
-            case get(waiting_bump) of
-                true -> ok;
-                _    -> self() ! bump_reduce_memory_use,
-                        put(waiting_bump, true)
-            end
-    end,
-    State3;
+            maybe_bump_reduce_memory_use(State3)
+    end;
 %% When using lazy queues, there are no alphas, so we don't need to
 %% call push_alphas_to_betas/2.
 reduce_memory_use(State = #vqstate {
@@ -2506,6 +2503,15 @@ reduce_memory_use(State = #vqstate {
     garbage_collect(),
     State3.
 
+maybe_bump_reduce_memory_use(State = #vqstate{ waiting_bump = true }) ->
+    State;
+maybe_bump_reduce_memory_use(State) ->
+    self() ! bump_reduce_memory_use,
+    State#vqstate{ waiting_bump = true }.
+
+handled_bump_reduce_memory_use(State = #vqstate{ waiting_bump = true }) ->
+    State#vqstate{ waiting_bump = false }.
+
 limit_ram_acks(0, State) ->
     {0, ui(State)};
 limit_ram_acks(Quota, State = #vqstate { ram_pending_ack  = RPA,

test/channel_operation_timeout_test_queue.erl

@@ -26,8 +26,10 @@
          set_ram_duration_target/2, ram_duration/1, needs_timeout/1, timeout/1,
          handle_pre_hibernate/1, resume/1, msg_rates/1,
          info/2, invoke/3, is_duplicate/2, set_queue_mode/2,
-         zip_msgs_and_acks/4]).
--export([start/2, stop/1]).
+         zip_msgs_and_acks/4, handled_bump_reduce_memory_use/1]).
+
+%% exported for testing only
+-export([start_msg_store/2, stop_msg_store/0, init/6]).
 
 %%----------------------------------------------------------------------------
 %% This test backing queue follows the variable queue implementation, with
@@ -303,6 +305,9 @@ set_queue_mode(Mode, State) ->
 zip_msgs_and_acks(Msgs, AckTags, Accumulator, State) ->
     rabbit_variable_queue:zip_msgs_and_acks(Msgs, AckTags, Accumulator, State).
 
+handled_bump_reduce_memory_use(State) ->
+    rabbit_variable_queue:handled_bump_reduce_memory_use(State).
+
 %% Delay
 maybe_delay(QPA) ->
   case is_timeout_test(gb_trees:values(QPA)) of