QQ: wip - improve message expiry

Author: kjnilsson

deps/rabbit/src/rabbit_fifo.erl

@@ -250,7 +250,7 @@
          % rabbit_fifo_index can be slow when calculating the smallest
          % index when there are large gaps but should be faster than gb_trees
          % for normal appending operations as it's backed by a map
-         unused_0 = ?NIL,
+         last_command_time = 0,
          unused_1 = ?NIL,
          % consumers need to reflect consumer state at time of snapshot
          consumers = #{} :: #{consumer_key() => consumer()},

deps/rabbit/src/rabbit_fifo.hrl

@@ -20,7 +20,10 @@
          from_lqueue/1,
          indexes/1,
          get_lowest_index/1,
-         overview/1
+         overview/1,
+         take_while/2,
+         any_priority_next/2,
+         fold_priorities_next/3
         ]).
 
 -define(STATE, pq).
@@ -156,8 +159,75 @@ overview(#?STATE{len = Len,
       num_active_priorities => map_size(Buckets),
       lowest_index => get_lowest_index(State)}.
 
+-spec take_while(fun ((msg()) -> boolean()), state()) ->
+    {[msg()], state()}.
+take_while(Fun, #?STATE{len = Len,
+                        buckets = Buckets0} = State)
+  when is_function(Fun) ->
+    {Buckets, Acc} = maps:fold(
+                       fun (P, Q0, {B0, Items0}) ->
+                               case take_while(Q0, Fun, Items0) of
+                                   {?EMPTY, Items} ->
+                                       {maps:remove(P, B0), Items};
+                                   {Q, Items} ->
+                                       {B0#{P => Q}, Items}
+                               end
+                       end,
+                       {Buckets0, []},
+                       maps:iterator(Buckets0, ordered)),
+
+    %% TODO: optimise updates
+    %% update bitmap
+    Bitmap = maps:fold(fun (P, _Q, B) -> B bor (1 bsl P) end, 0, Buckets),
+
+    {lists:reverse(Acc),
+     State#?STATE{len = Len - length(Acc),
+                  buckets = Buckets,
+                  bitmap = Bitmap}}.
+
+-spec any_priority_next(fun ((msg()) -> boolean()), state()) ->
+    boolean().
+any_priority_next(Fun, #?STATE{buckets = Buckets0})
+  when is_function(Fun) ->
+    maps_any(Fun, maps:next(maps:iterator(Buckets0))).
+
+-spec fold_priorities_next(fun ((msg(), Acc) -> Acc), Acc, state()) ->
+    Acc when Acc :: term().
+fold_priorities_next(Fun, Acc, #?STATE{buckets = Buckets0})
+  when is_function(Fun) ->
+    maps:fold(fun (_P, Q, A) ->
+                      Fun(peek(Q), A)
+              end, Acc, Buckets0).
+
 %% INTERNAL
 
+maps_any(_Fun, none) ->
+    false;
+maps_any(Fun, {_, Q, I}) ->
+    case Fun(peek(Q)) of
+        true ->
+            true;
+        false ->
+            maps_any(Fun, maps:next(I))
+    end.
+
+take_while(?EMPTY, _Fun, Acc) ->
+    {?EMPTY, Acc};
+take_while(Q, Fun, Acc) ->
+    case peek(Q) of
+        empty ->
+            {Q, Acc};
+        Msg ->
+            case Fun(Msg) of
+                true ->
+                    take_while(drop(Q), Fun, [Msg | Acc]);
+                false ->
+                    {Q, Acc}
+            end
+    end.
+
+
+
 %% invariant, if the queue is non empty so is the Out (right) list.
 in(X, ?EMPTY) ->
     {1, [], [X]};

deps/rabbit/src/rabbit_fifo_pq.erl

@@ -1987,7 +1987,7 @@ make_ra_conf(Q, ServerId, Membership, MacVersion)
                  Membership, MacVersion).
 
 make_ra_conf(Q, ServerId, TickTimeout,
-             _SnapshotInterval, CheckpointInterval,
+             _SnapshotInterval, _CheckpointInterval,
              Membership, MacVersion) ->
     QName = amqqueue:get_name(Q),
     #resource{name = QNameBin} = QName,
@@ -1998,8 +1998,9 @@ make_ra_conf(Q, ServerId, TickTimeout,
     Formatter = {?MODULE, format_ra_event, [QName]},
     LogCfg = #{uid => UId,
                min_snapshot_interval => 0,
-               min_checkpoint_interval => CheckpointInterval,
-               max_checkpoints => 3},
+               % min_checkpoint_interval => CheckpointInterval,
+               % max_checkpoints => 3,
+               major_compaction_strategy => {num_minors, 32}},
     rabbit_misc:maps_put_truthy(membership, Membership,
                                 #{cluster_name => ClusterName,
                                   id => ServerId,

deps/rabbit/src/rabbit_quorum_queue.erl

@@ -2147,6 +2147,8 @@ dead_letter_policy(Config) ->
 %% Test that messages are at most once dead letter in the correct order
 %% for reason 'maxlen'.
 at_most_once_dead_letter_order_maxlen(Config) ->
+    check_quorum_queues_v8_compat(Config),
+
     [Server | _] = Servers = rabbit_ct_broker_helpers:get_node_configs(Config, nodename),
 
     Ch = rabbit_ct_client_helpers:open_channel(Config, Server),
@@ -2261,16 +2263,18 @@ at_most_once_dead_letter_order_delivery_limit(Config) ->
                            #'basic.publish'{routing_key = QQ},
                            #amqp_msg{payload = <<"m2">>}),
 
-    ok = subscribe(Ch, QQ, false),
+    Ch2 = rabbit_ct_client_helpers:open_channel(Config, Server),
+    ok = subscribe(Ch2, QQ, false),
     receive {_, #amqp_msg{payload = P1}} ->
                 ?assertEqual(<<"m1">>, P1)
     end,
     receive {_, #amqp_msg{payload = P2}} ->
                 ?assertEqual(<<"m2">>, P2)
     end,
-    ok = amqp_channel:call(Ch, #'basic.nack'{delivery_tag = 0,
-                                             multiple = true,
-                                             requeue = true}),
+    amqp_channel:close(Ch2),
+    % ok = amqp_channel:call(Ch, #'basic.nack'{delivery_tag = 0,
+    %                                          multiple = true,
+    %                                          requeue = true}),
 
     wait_for_consensus(DLQ, Config),
     wait_for_messages_ready(Servers,  ra_name(DLQ), 2),
@@ -5337,7 +5341,7 @@ check_quorum_queues_v8_compat(Config) ->
         true ->
             ok;
         false ->
-            throw({skip, "test will only work on QQ machine version > 8"})
+            throw({skip, "test will only work on QQ machine version >= 8"})
     end.
 
 lists_interleave([], _List) ->

deps/rabbit/test/quorum_queue_SUITE.erl

@@ -1620,8 +1620,11 @@ single_active_consumer_state_enter_leader_include_waiting_consumers_test(Config)
               ],
     {State1, _} = run_log(Config, State0, Entries),
     Effects = rabbit_fifo:state_enter(leader, State1),
-    %% 2 effects for each consumer process (channel process), 1 effect for the node,
-    ?assertEqual(2 * 3 + 1 + 1 + 1, length(Effects)).
+    ct:pal("Efx ~p", [Effects]),
+    %% 2 effects for each consumer process (channel process),
+    %% 1 effect for the node,
+    %% 1 for decorators
+    ?assertEqual(2 * 3 + 1 + 1, length(Effects)).
 
 single_active_consumer_state_enter_eol_include_waiting_consumers_test(Config) ->
     Resource = rabbit_misc:r("/", queue, ?FUNCTION_NAME_B),
@@ -3223,6 +3226,39 @@ modify_test(Config) ->
 
     ok.
 
+priorities_expire_test(Config) ->
+    State0 = init(#{name => ?FUNCTION_NAME,
+                    queue_resource => rabbit_misc:r("/", queue,
+                                                    ?FUNCTION_NAME_B)}),
+    Pid1 = spawn(fun() -> ok end),
+
+    Entries =
+    [
+     {?LINE, make_enqueue(Pid1, 1,
+                          mk_mc(<<"p1">>, #'P_basic'{priority = 9,
+                                                     expiration = <<"100">>}))},
+     {?LINE, make_enqueue(Pid1, 2,
+                          mk_mc(<<"p1">>, #'P_basic'{priority = 9,
+                                                     expiration = <<"100000">>}))},
+     {?LINE, make_enqueue(Pid1, 3,
+                          mk_mc(<<"p7">>, #'P_basic'{priority = 7,
+                                                     expiration = <<"100">>}))},
+     {?LINE, make_enqueue(Pid1, 4,
+                          mk_mc(<<"p7">>, #'P_basic'{priority = 7,
+                                                     expiration = <<"100000">>}))},
+     {?LINE, make_enqueue(Pid1, 5,
+                          mk_mc(<<"p7b">>, #'P_basic'{priority = 3}))},
+
+     {?LINE + 101, {timeout, {expire_msgs, shallow}}},
+
+     ?ASSERT(_, fun(State) ->
+                        ?assertMatch(#{num_messages := 3},
+                                     rabbit_fifo:overview(State))
+                end)
+    ],
+    {_State2, _} = run_log(Config, State0, Entries),
+    ok.
+
 %% Utility
 %%
 
@@ -3232,6 +3268,7 @@ apply(Meta, Entry, State) -> rabbit_fifo:apply(Meta, Entry, State).
 init_aux(Conf) -> rabbit_fifo:init_aux(Conf).
 handle_aux(S, T, C, A, A2) -> rabbit_fifo:handle_aux(S, T, C, A, A2).
 make_checkout(C, S, M) -> rabbit_fifo:make_checkout(C, S, M).
+make_enqueue(P, S, M) -> rabbit_fifo:make_enqueue(P, S, M).
 
 cid(A) when is_atom(A) ->
     atom_to_binary(A, utf8).
@@ -3242,10 +3279,13 @@ single_active_invariant( #rabbit_fifo{consumers = Cons}) ->
                               end, Cons)).
 
 mk_mc(Body) ->
+    mk_mc(Body, #'P_basic'{}).
+
+mk_mc(Body, BasicProps) ->
     mc_amqpl:from_basic_message(
       #basic_message{routing_keys = [<<"">>],
                      exchange_name = #resource{name = <<"x">>,
                                                kind = exchange,
                                                virtual_host = <<"v">>},
-                     content = #content{properties = #'P_basic'{},
+                     content = #content{properties = BasicProps,
                                         payload_fragments_rev = [Body]}}).

deps/rabbit/test/rabbit_fifo_SUITE.erl

@@ -17,6 +17,8 @@ all() ->
 all_tests() ->
     [
      basics,
+     take_while,
+     any_priority_next,
      property
     ].
 
@@ -80,6 +82,73 @@ basics(_Config) ->
     empty = rabbit_fifo_pq:out(Q6),
     ok.
 
+take_while(_Config) ->
+    Q1 = lists:foldl(
+           fun ({P, I}, Q) ->
+                   rabbit_fifo_pq:in(P, I, Q)
+           end, rabbit_fifo_pq:new(),
+           [
+            {1, ?MSG(1)}, {1, ?MSG(2)}, {1, ?MSG(3)},
+            {2, ?MSG(1)}, {2, ?MSG(2)}, {2, ?MSG(3)},
+            {3, ?MSG(1)}, {3, ?MSG(2)}, {3, ?MSG(3)},
+            {4, ?MSG(1)}, {4, ?MSG(2)}, {4, ?MSG(3)},
+            {5, ?MSG(1, 10)}, {5, ?MSG(2, 20)}, {5, ?MSG(3, 30)}
+           ]),
+
+    {Taken, Q2} = rabbit_fifo_pq:take_while(fun (?MSG(I, _)) ->
+                                                    I < 3
+                                            end, Q1),
+    ?assertMatch([
+                  ?MSG(1, 10), ?MSG(2, 20),
+                  ?MSG(1, 1), ?MSG(2, 2),
+                  ?MSG(1, 1), ?MSG(2, 2),
+                  ?MSG(1, 1), ?MSG(2, 2),
+                  ?MSG(1, 1), ?MSG(2, 2)
+                 ], Taken),
+
+
+    ?assertEqual(5, rabbit_fifo_pq:len(Q2)),
+    ?assertEqual(10, length(Taken)),
+    {?MSG(3, 30), Q3} = rabbit_fifo_pq:out(Q2),
+    {?MSG(3), Q4} = rabbit_fifo_pq:out(Q3),
+    {?MSG(3), Q5} = rabbit_fifo_pq:out(Q4),
+    {?MSG(3), Q6} = rabbit_fifo_pq:out(Q5),
+    {?MSG(3), _Q7} = rabbit_fifo_pq:out(Q6),
+
+
+    {_Taken2, Q} = rabbit_fifo_pq:take_while(fun (?MSG(_, _)) ->
+                                                    true
+                                            end, Q2),
+
+    ct:pal("Q ~p", [Q]),
+
+    ok.
+
+any_priority_next(_Config) ->
+    Q0 = rabbit_fifo_pq:new(),
+
+    ?assertNot(rabbit_fifo_pq:any_priority_next(fun (_) -> true end, Q0)),
+
+    Q1 = lists:foldl(fun ({P, I}, Q) ->
+                             rabbit_fifo_pq:in(P, I, Q)
+                     end, Q0,
+                     [
+                      {1, ?MSG(1)}, {1, ?MSG(2)}, {1, ?MSG(3)},
+                      {2, ?MSG(1)}, {2, ?MSG(2)}, {2, ?MSG(3)},
+                      {3, ?MSG(2)}, {3, ?MSG(3)},
+                      {4, ?MSG(1)}, {4, ?MSG(2)}, {4, ?MSG(3)},
+                      {5, ?MSG(1)}, {5, ?MSG(2)}, {5, ?MSG(3)}
+                     ]),
+
+    ?assert(rabbit_fifo_pq:any_priority_next(fun (?MSG(I, _)) ->
+                                                     I > 1
+                                             end, Q1)),
+    ?assertNot(rabbit_fifo_pq:any_priority_next(fun (?MSG(I, _)) ->
+                                                        I > 6
+                                                end, Q1)),
+
+    ok.
+
 hi_is_prioritised(_Config) ->
     Q0 = rabbit_fifo_q:new(),
     %% when `hi' has a lower index than the next 'no' then it is still