CQ: No longer limit reads to a single segment

This should reduce the number of calls we make to the
queue's index/store and therefore improve performance.

Author: lhoguin

deps/rabbit/src/rabbit_classic_queue_index_v2.erl

@@ -20,7 +20,7 @@
 %% queue implementation itself.
 %% @todo TODO
 -export([sync/1, needs_sync/1,
-         bounds/2, next_segment_boundary/1]).
+         bounds/2]).
 
 %% Called by rabbit_vhost.
 -export([all_queue_directory_names/1]).
@@ -762,11 +762,6 @@ ack_delete_fold_fun(SeqId, Write, {Buffer, Updates, Deletes, SegmentEntryCount})
              Deletes, SegmentEntryCount}
     end.
 
-%% A better interface for read/3 would be to request a maximum
-%% of N messages, rather than first call next_segment_boundary/3
-%% and then read from S1 to S2. This function could then return
-%% either N messages or less depending on the current state.
-
 -spec read(rabbit_variable_queue:seq_id(),
            rabbit_variable_queue:seq_id(),
            State) ->
@@ -1069,16 +1064,6 @@ bounds(State = #qi{ segments = Segments }, NextSeqIdHint) ->
              State}
     end.
 
-%% The next_segment_boundary/1 function is used internally when
-%% reading. It should not be called from rabbit_variable_queue.
-
--spec next_segment_boundary(SeqId) -> SeqId when SeqId::rabbit_variable_queue:seq_id().
-
-next_segment_boundary(SeqId) ->
-    ?DEBUG("~0p", [SeqId]),
-    SegmentEntryCount = segment_entry_count(),
-    (1 + (SeqId div SegmentEntryCount)) * SegmentEntryCount.
-
 %% ----
 %%
 %% Internal.
@@ -1089,6 +1074,10 @@ segment_entry_count() ->
     %% producer produces.
     persistent_term:get(classic_queue_index_v2_segment_entry_count, 4096).
 
+next_segment_boundary(SeqId) ->
+    SegmentEntryCount = segment_entry_count(),
+    (1 + (SeqId div SegmentEntryCount)) * SegmentEntryCount.
+
 %% Note that store files will also be removed if there are any in this directory.
 %% Currently the v2 per-queue store expects this function to remove its own files.
 

deps/rabbit/src/rabbit_variable_queue.erl

@@ -1864,11 +1864,8 @@ read_from_q_tail(DelsAndAcksFun,
     %% For v2 we want to limit the number of messages read at once to lower
     %% the memory footprint. We use the consume rate to determine how many
     %% messages we read.
-    %% @todo Simply ask for N messages instead of low/high bounds.
     QTailSeqLimit = QTailSeqId + MemoryLimit,
-    QTailSeqId1 =
-        lists:min([rabbit_classic_queue_index_v2:next_segment_boundary(QTailSeqId),
-                   QTailSeqLimit, QTailSeqIdEnd]),
+    QTailSeqId1 = min(QTailSeqLimit, QTailSeqIdEnd),
     {List0, IndexState1} = rabbit_classic_queue_index_v2:read(QTailSeqId, QTailSeqId1, IndexState),
     {List, StoreState3, MCStateP3, MCStateT3} = case WhatToRead of
         messages ->

deps/rabbit/test/backing_queue_SUITE.erl

@@ -796,9 +796,12 @@ bq_queue_index(Config) ->
 index_mod() ->
     rabbit_classic_queue_index_v2.
 
+segment_size() ->
+    persistent_term:get(classic_queue_index_v2_segment_entry_count, 4096).
+
 bq_queue_index1(_Config) ->
     IndexMod = index_mod(),
-    SegmentSize = IndexMod:next_segment_boundary(0),
+    SegmentSize = segment_size(),
     TwoSegs = SegmentSize + SegmentSize,
     MostOfASegment = trunc(SegmentSize*0.75),
     SeqIdsA = lists:seq(0, MostOfASegment-1),
@@ -992,8 +995,8 @@ v2_delete_segment_file_completely_acked(Config) ->
       ?MODULE, v2_delete_segment_file_completely_acked1, [Config]).
 
 v2_delete_segment_file_completely_acked1(_Config) ->
-    IndexMod = rabbit_classic_queue_index_v2,
-    SegmentSize = IndexMod:next_segment_boundary(0),
+    IndexMod = index_mod(),
+    SegmentSize = segment_size(),
     SeqIds = lists:seq(0, SegmentSize - 1),
 
     with_empty_test_queue(
@@ -1019,8 +1022,8 @@ v2_delete_segment_file_partially_acked(Config) ->
       ?MODULE, v2_delete_segment_file_partially_acked1, [Config]).
 
 v2_delete_segment_file_partially_acked1(_Config) ->
-    IndexMod = rabbit_classic_queue_index_v2,
-    SegmentSize = IndexMod:next_segment_boundary(0),
+    IndexMod = index_mod(),
+    SegmentSize = segment_size(),
     SeqIds = lists:seq(0, SegmentSize div 2),
     SeqIdsLen = length(SeqIds),
 
@@ -1047,8 +1050,8 @@ v2_delete_segment_file_partially_acked_with_holes(Config) ->
       ?MODULE, v2_delete_segment_file_partially_acked_with_holes1, [Config]).
 
 v2_delete_segment_file_partially_acked_with_holes1(_Config) ->
-    IndexMod = rabbit_classic_queue_index_v2,
-    SegmentSize = IndexMod:next_segment_boundary(0),
+    IndexMod = index_mod(),
+    SegmentSize = segment_size(),
     SeqIdsA = lists:seq(0, SegmentSize div 2),
     SeqIdsB = lists:seq(11 + SegmentSize div 2, SegmentSize - 1),
     SeqIdsLen = length(SeqIdsA) + length(SeqIdsB),
@@ -1107,8 +1110,7 @@ bq_queue_recover(Config) ->
       ?MODULE, bq_queue_recover1, [Config]).
 
 bq_queue_recover1(Config) ->
-    IndexMod = index_mod(),
-    Count = 2 * IndexMod:next_segment_boundary(0),
+    Count = 2 * segment_size(),
     QName0 = queue_name(Config, <<"bq_queue_recover-q">>),
     {new, Q} = rabbit_amqqueue:declare(QName0, true, false, [], none, <<"acting-user">>),
     QName = amqqueue:get_name(Q),
@@ -1164,14 +1166,13 @@ get_queue_sup_pid([], _QueuePid) ->
 
 variable_queue_partial_segments_q_tail_thing(Config) ->
     passed = rabbit_ct_broker_helpers:rpc(Config, 0,
-      ?MODULE, variable_queue_partial_segments_q_tail_thing1, [Config]).
+      ?MODULE, variable_queue_partial_segments_q_tail_thing1, []).
 
-variable_queue_partial_segments_q_tail_thing1(Config) ->
+variable_queue_partial_segments_q_tail_thing1() ->
     with_fresh_variable_queue(fun variable_queue_partial_segments_q_tail_thing2/2).
 
 variable_queue_partial_segments_q_tail_thing2(VQ0, _QName) ->
-    IndexMod = index_mod(),
-    SegmentSize = IndexMod:next_segment_boundary(0),
+    SegmentSize = segment_size(),
     HalfSegment = SegmentSize div 2,
     OneAndAHalfSegment = SegmentSize + HalfSegment,
     VQ1 = variable_queue_publish(true, OneAndAHalfSegment, VQ0),
@@ -1194,12 +1195,8 @@ variable_queue_partial_segments_q_tail_thing2(VQ0, _QName) ->
                                           SegmentSize + HalfSegment + 1, VQ5),
     VQ7 = check_variable_queue_status(
             VQ6,
-            %% We only read from q_tail up to the end of the segment, so
-            %% after fetching exactly one segment, we should have no
-            %% messages in memory.
-            [{q_head, 0},
-             {q_tail, {q_tail, SegmentSize, HalfSegment + 1, OneAndAHalfSegment + 1}},
-             {len, HalfSegment + 1}]),
+            %% We can't make assumptions about how many messages are in memory.
+            [{len, HalfSegment + 1}]),
     {VQ8, AckTags1} = variable_queue_fetch(HalfSegment + 1, true, false,
                                            HalfSegment + 1, VQ7),
     {_Guids, VQ9} = rabbit_variable_queue:ack(AckTags ++ AckTags1, VQ8),
@@ -1209,14 +1206,13 @@ variable_queue_partial_segments_q_tail_thing2(VQ0, _QName) ->
 
 variable_queue_all_the_bits_not_covered_elsewhere_A(Config) ->
     passed = rabbit_ct_broker_helpers:rpc(Config, 0,
-      ?MODULE, variable_queue_all_the_bits_not_covered_elsewhere_A1, [Config]).
+      ?MODULE, variable_queue_all_the_bits_not_covered_elsewhere_A1, []).
 
-variable_queue_all_the_bits_not_covered_elsewhere_A1(Config) ->
+variable_queue_all_the_bits_not_covered_elsewhere_A1() ->
     with_fresh_variable_queue(fun variable_queue_all_the_bits_not_covered_elsewhere_A2/2).
 
 variable_queue_all_the_bits_not_covered_elsewhere_A2(VQ0, QName) ->
-    IndexMod = index_mod(),
-    Count = 2 * IndexMod:next_segment_boundary(0),
+    Count = 2 * segment_size(),
     VQ1 = variable_queue_publish(true, Count, VQ0),
     VQ2 = variable_queue_publish(false, Count, VQ1),
     {VQ4, _AckTags}  = variable_queue_fetch(Count, true, false,
@@ -1234,9 +1230,9 @@ variable_queue_all_the_bits_not_covered_elsewhere_A2(VQ0, QName) ->
 
 variable_queue_all_the_bits_not_covered_elsewhere_B(Config) ->
     passed = rabbit_ct_broker_helpers:rpc(Config, 0,
-      ?MODULE, variable_queue_all_the_bits_not_covered_elsewhere_B1, [Config]).
+      ?MODULE, variable_queue_all_the_bits_not_covered_elsewhere_B1, []).
 
-variable_queue_all_the_bits_not_covered_elsewhere_B1(Config) ->
+variable_queue_all_the_bits_not_covered_elsewhere_B1() ->
     with_fresh_variable_queue(fun variable_queue_all_the_bits_not_covered_elsewhere_B2/2).
 
 variable_queue_all_the_bits_not_covered_elsewhere_B2(VQ1, QName) ->
@@ -1252,9 +1248,9 @@ variable_queue_all_the_bits_not_covered_elsewhere_B2(VQ1, QName) ->
 
 variable_queue_drop(Config) ->
     passed = rabbit_ct_broker_helpers:rpc(Config, 0,
-      ?MODULE, variable_queue_drop1, [Config]).
+      ?MODULE, variable_queue_drop1, []).
 
-variable_queue_drop1(Config) ->
+variable_queue_drop1() ->
     with_fresh_variable_queue(fun variable_queue_drop2/2).
 
 variable_queue_drop2(VQ0, _QName) ->
@@ -1275,9 +1271,9 @@ variable_queue_drop2(VQ0, _QName) ->
 
 variable_queue_fold_msg_on_disk(Config) ->
     passed = rabbit_ct_broker_helpers:rpc(Config, 0,
-      ?MODULE, variable_queue_fold_msg_on_disk1, [Config]).
+      ?MODULE, variable_queue_fold_msg_on_disk1, []).
 
-variable_queue_fold_msg_on_disk1(Config) ->
+variable_queue_fold_msg_on_disk1() ->
     with_fresh_variable_queue(fun variable_queue_fold_msg_on_disk2/2).
 
 variable_queue_fold_msg_on_disk2(VQ0, _QName) ->
@@ -1289,9 +1285,9 @@ variable_queue_fold_msg_on_disk2(VQ0, _QName) ->
 
 variable_queue_dropfetchwhile(Config) ->
     passed = rabbit_ct_broker_helpers:rpc(Config, 0,
-      ?MODULE, variable_queue_dropfetchwhile1, [Config]).
+      ?MODULE, variable_queue_dropfetchwhile1, []).
 
-variable_queue_dropfetchwhile1(Config) ->
+variable_queue_dropfetchwhile1() ->
     with_fresh_variable_queue(fun variable_queue_dropfetchwhile2/2).
 
 variable_queue_dropfetchwhile2(VQ0, _QName) ->
@@ -1335,9 +1331,9 @@ variable_queue_dropfetchwhile2(VQ0, _QName) ->
 
 variable_queue_dropwhile_restart(Config) ->
     passed = rabbit_ct_broker_helpers:rpc(Config, 0,
-      ?MODULE, variable_queue_dropwhile_restart1, [Config]).
+      ?MODULE, variable_queue_dropwhile_restart1, []).
 
-variable_queue_dropwhile_restart1(Config) ->
+variable_queue_dropwhile_restart1() ->
     with_fresh_variable_queue(fun variable_queue_dropwhile_restart2/2).
 
 variable_queue_dropwhile_restart2(VQ0, QName) ->
@@ -1372,9 +1368,9 @@ variable_queue_dropwhile_restart2(VQ0, QName) ->
 
 variable_queue_dropwhile_sync_restart(Config) ->
     passed = rabbit_ct_broker_helpers:rpc(Config, 0,
-      ?MODULE, variable_queue_dropwhile_sync_restart1, [Config]).
+      ?MODULE, variable_queue_dropwhile_sync_restart1, []).
 
-variable_queue_dropwhile_sync_restart1(Config) ->
+variable_queue_dropwhile_sync_restart1() ->
     with_fresh_variable_queue(fun variable_queue_dropwhile_sync_restart2/2).
 
 variable_queue_dropwhile_sync_restart2(VQ0, QName) ->
@@ -1412,9 +1408,9 @@ variable_queue_dropwhile_sync_restart2(VQ0, QName) ->
 
 variable_queue_restart_large_seq_id(Config) ->
     passed = rabbit_ct_broker_helpers:rpc(Config, 0,
-      ?MODULE, variable_queue_restart_large_seq_id1, [Config]).
+      ?MODULE, variable_queue_restart_large_seq_id1, []).
 
-variable_queue_restart_large_seq_id1(Config) ->
+variable_queue_restart_large_seq_id1() ->
     with_fresh_variable_queue(fun variable_queue_restart_large_seq_id2/2).
 
 variable_queue_restart_large_seq_id2(VQ0, QName) ->
@@ -1449,9 +1445,9 @@ variable_queue_restart_large_seq_id2(VQ0, QName) ->
 
 variable_queue_ack_limiting(Config) ->
     passed = rabbit_ct_broker_helpers:rpc(Config, 0,
-      ?MODULE, variable_queue_ack_limiting1, [Config]).
+      ?MODULE, variable_queue_ack_limiting1, []).
 
-variable_queue_ack_limiting1(Config) ->
+variable_queue_ack_limiting1() ->
     with_fresh_variable_queue(fun variable_queue_ack_limiting2/2).
 
 variable_queue_ack_limiting2(VQ0, _Config) ->
@@ -1477,9 +1473,9 @@ variable_queue_ack_limiting2(VQ0, _Config) ->
 
 variable_queue_purge(Config) ->
     passed = rabbit_ct_broker_helpers:rpc(Config, 0,
-      ?MODULE, variable_queue_purge1, [Config]).
+      ?MODULE, variable_queue_purge1, []).
 
-variable_queue_purge1(Config) ->
+variable_queue_purge1() ->
     with_fresh_variable_queue(fun variable_queue_purge2/2).
 
 variable_queue_purge2(VQ0, _Config) ->
@@ -1499,9 +1495,9 @@ variable_queue_purge2(VQ0, _Config) ->
 
 variable_queue_requeue(Config) ->
     passed = rabbit_ct_broker_helpers:rpc(Config, 0,
-      ?MODULE, variable_queue_requeue1, [Config]).
+      ?MODULE, variable_queue_requeue1, []).
 
-variable_queue_requeue1(Config) ->
+variable_queue_requeue1() ->
     with_fresh_variable_queue(fun variable_queue_requeue2/2).
 
 variable_queue_requeue2(VQ0, _Config) ->
@@ -1525,14 +1521,13 @@ variable_queue_requeue2(VQ0, _Config) ->
 %% requeue from ram_pending_ack into q_head, move to q_tail and then empty queue
 variable_queue_requeue_ram_beta(Config) ->
     passed = rabbit_ct_broker_helpers:rpc(Config, 0,
-      ?MODULE, variable_queue_requeue_ram_beta1, [Config]).
+      ?MODULE, variable_queue_requeue_ram_beta1, []).
 
-variable_queue_requeue_ram_beta1(Config) ->
+variable_queue_requeue_ram_beta1() ->
     with_fresh_variable_queue(fun variable_queue_requeue_ram_beta2/2).
 
 variable_queue_requeue_ram_beta2(VQ0, _Config) ->
-    IndexMod = index_mod(),
-    Count = IndexMod:next_segment_boundary(0)*2 + 2,
+    Count = 2 + 2 * segment_size(),
     VQ1 = variable_queue_publish(false, Count, VQ0),
     {VQ2, AcksR} = variable_queue_fetch(Count, false, false, Count, VQ1),
     {Back, Front} = lists:split(Count div 2, AcksR),
@@ -1808,8 +1803,7 @@ requeue_one_by_one(Acks, VQ) ->
 %% internal queues. Kept for completeness.
 variable_queue_with_holes(VQ0) ->
     Interval = 2048, %% should match vq:IO_BATCH_SIZE
-    IndexMod = index_mod(),
-    Count = IndexMod:next_segment_boundary(0)*2 + 2 * Interval,
+    Count = 2 * Interval + 2 * segment_size(),
     Seq = lists:seq(1, Count),
     VQ1 = variable_queue_publish(
             false, 1, Count,