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

@@ -1456,7 +1456,6 @@ handle_info({bump_credit, Msg}, State = #q{backing_queue       = BQ,
     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(Info, State) ->

src/rabbit_variable_queue.erl

@@ -325,7 +325,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 }).
@@ -911,7 +912,10 @@ timeout(State = #vqstate { index_state = IndexState }) ->
 handle_pre_hibernate(State = #vqstate { index_state = IndexState }) ->
     State #vqstate { index_state = rabbit_queue_index:flush(IndexState) }.
 
-resume(State) -> a(reduce_memory_use(State)).
+resume(State = #vqstate{waiting_bump = true}) ->
+    resume(State#vqstate{waiting_bump = false});
+resume(State) ->
+    a(reduce_memory_use(State)).
 
 msg_rates(#vqstate { rates = #rates { in  = AvgIngressRate,
                                       out = AvgEgressRate } }) ->
@@ -2466,21 +2470,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 +2505,12 @@ 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}.
+
 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,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]).
--export([start/2, stop/1]).
+         start/2, stop/1, zip_msgs_and_acks/4]).
 
 %%----------------------------------------------------------------------------
 %% This test backing queue follows the variable queue implementation, with
@@ -91,7 +90,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 }).
@@ -261,7 +261,7 @@ ackfold(MsgFun, Acc, State, AckTags) ->
 fold(Fun, Acc, State) ->
     rabbit_variable_queue:fold(Fun, Acc, State).
 
-len(#vqstate { qi_pending_ack = QPA } = State) ->
+len(#vqstate{ qi_pending_ack = QPA } = State) ->
     maybe_delay(QPA),
     rabbit_variable_queue:len(State).