Revert "CQ: No longer limit reads to a single segment"

This reverts commit b586ea8.

Author: lhoguin

deps/rabbit/src/rabbit_classic_queue_index_v2.erl

@@ -8,7 +8,7 @@
 -module(rabbit_classic_queue_index_v2).
 
 -export([erase/1, init/1, reset_state/1, recover/4,
-         bounds/2, info/1,
+         bounds/2, next_segment_boundary/1, info/1,
          terminate/3, delete_and_terminate/1,
          publish/7, ack/2, read/3,
          sync/1, needs_sync/1]).
@@ -757,6 +757,11 @@ 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) ->
@@ -1059,6 +1064,16 @@ 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.
@@ -1069,10 +1084,6 @@ 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,8 +1864,11 @@ 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 = min(QTailSeqLimit, QTailSeqIdEnd),
+    QTailSeqId1 =
+        lists:min([rabbit_classic_queue_index_v2:next_segment_boundary(QTailSeqId),
+                   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,12 +796,9 @@ 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 = segment_size(),
+    SegmentSize = IndexMod:next_segment_boundary(0),
     TwoSegs = SegmentSize + SegmentSize,
     MostOfASegment = trunc(SegmentSize*0.75),
     SeqIdsA = lists:seq(0, MostOfASegment-1),
@@ -995,8 +992,8 @@ v2_delete_segment_file_completely_acked(Config) ->
       ?MODULE, v2_delete_segment_file_completely_acked1, [Config]).
 
 v2_delete_segment_file_completely_acked1(_Config) ->
-    IndexMod = index_mod(),
-    SegmentSize = segment_size(),
+    IndexMod = rabbit_classic_queue_index_v2,
+    SegmentSize = IndexMod:next_segment_boundary(0),
     SeqIds = lists:seq(0, SegmentSize - 1),
 
     with_empty_test_queue(
@@ -1022,8 +1019,8 @@ v2_delete_segment_file_partially_acked(Config) ->
       ?MODULE, v2_delete_segment_file_partially_acked1, [Config]).
 
 v2_delete_segment_file_partially_acked1(_Config) ->
-    IndexMod = index_mod(),
-    SegmentSize = segment_size(),
+    IndexMod = rabbit_classic_queue_index_v2,
+    SegmentSize = IndexMod:next_segment_boundary(0),
     SeqIds = lists:seq(0, SegmentSize div 2),
     SeqIdsLen = length(SeqIds),
 
@@ -1050,8 +1047,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 = index_mod(),
-    SegmentSize = segment_size(),
+    IndexMod = rabbit_classic_queue_index_v2,
+    SegmentSize = IndexMod:next_segment_boundary(0),
     SeqIdsA = lists:seq(0, SegmentSize div 2),
     SeqIdsB = lists:seq(11 + SegmentSize div 2, SegmentSize - 1),
     SeqIdsLen = length(SeqIdsA) + length(SeqIdsB),
@@ -1110,7 +1107,8 @@ bq_queue_recover(Config) ->
       ?MODULE, bq_queue_recover1, [Config]).
 
 bq_queue_recover1(Config) ->
-    Count = 2 * segment_size(),
+    IndexMod = index_mod(),
+    Count = 2 * IndexMod:next_segment_boundary(0),
     QName0 = queue_name(Config, <<"bq_queue_recover-q">>),
     {new, Q} = rabbit_amqqueue:declare(QName0, true, false, [], none, <<"acting-user">>),
     QName = amqqueue:get_name(Q),
@@ -1166,13 +1164,14 @@ 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, []).
+      ?MODULE, variable_queue_partial_segments_q_tail_thing1, [Config]).
 
-variable_queue_partial_segments_q_tail_thing1() ->
+variable_queue_partial_segments_q_tail_thing1(Config) ->
     with_fresh_variable_queue(fun variable_queue_partial_segments_q_tail_thing2/2).
 
 variable_queue_partial_segments_q_tail_thing2(VQ0, _QName) ->
-    SegmentSize = segment_size(),
+    IndexMod = index_mod(),
+    SegmentSize = IndexMod:next_segment_boundary(0),
     HalfSegment = SegmentSize div 2,
     OneAndAHalfSegment = SegmentSize + HalfSegment,
     VQ1 = variable_queue_publish(true, OneAndAHalfSegment, VQ0),
@@ -1195,8 +1194,12 @@ variable_queue_partial_segments_q_tail_thing2(VQ0, _QName) ->
                                           SegmentSize + HalfSegment + 1, VQ5),
     VQ7 = check_variable_queue_status(
             VQ6,
-            %% We can't make assumptions about how many messages are in memory.
-            [{len, HalfSegment + 1}]),
+            %% 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}]),
     {VQ8, AckTags1} = variable_queue_fetch(HalfSegment + 1, true, false,
                                            HalfSegment + 1, VQ7),
     {_Guids, VQ9} = rabbit_variable_queue:ack(AckTags ++ AckTags1, VQ8),
@@ -1206,13 +1209,14 @@ 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, []).
+      ?MODULE, variable_queue_all_the_bits_not_covered_elsewhere_A1, [Config]).
 
-variable_queue_all_the_bits_not_covered_elsewhere_A1() ->
+variable_queue_all_the_bits_not_covered_elsewhere_A1(Config) ->
     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) ->
-    Count = 2 * segment_size(),
+    IndexMod = index_mod(),
+    Count = 2 * IndexMod:next_segment_boundary(0),
     VQ1 = variable_queue_publish(true, Count, VQ0),
     VQ2 = variable_queue_publish(false, Count, VQ1),
     {VQ4, _AckTags}  = variable_queue_fetch(Count, true, false,
@@ -1230,9 +1234,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, []).
+      ?MODULE, variable_queue_all_the_bits_not_covered_elsewhere_B1, [Config]).
 
-variable_queue_all_the_bits_not_covered_elsewhere_B1() ->
+variable_queue_all_the_bits_not_covered_elsewhere_B1(Config) ->
     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) ->
@@ -1248,9 +1252,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, []).
+      ?MODULE, variable_queue_drop1, [Config]).
 
-variable_queue_drop1() ->
+variable_queue_drop1(Config) ->
     with_fresh_variable_queue(fun variable_queue_drop2/2).
 
 variable_queue_drop2(VQ0, _QName) ->
@@ -1271,9 +1275,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, []).
+      ?MODULE, variable_queue_fold_msg_on_disk1, [Config]).
 
-variable_queue_fold_msg_on_disk1() ->
+variable_queue_fold_msg_on_disk1(Config) ->
     with_fresh_variable_queue(fun variable_queue_fold_msg_on_disk2/2).
 
 variable_queue_fold_msg_on_disk2(VQ0, _QName) ->
@@ -1285,9 +1289,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, []).
+      ?MODULE, variable_queue_dropfetchwhile1, [Config]).
 
-variable_queue_dropfetchwhile1() ->
+variable_queue_dropfetchwhile1(Config) ->
     with_fresh_variable_queue(fun variable_queue_dropfetchwhile2/2).
 
 variable_queue_dropfetchwhile2(VQ0, _QName) ->
@@ -1331,9 +1335,9 @@ variable_queue_dropfetchwhile2(VQ0, _QName) ->
 
 variable_queue_dropwhile_restart(Config) ->
     passed = rabbit_ct_broker_helpers:rpc(Config, 0,
-      ?MODULE, variable_queue_dropwhile_restart1, []).
+      ?MODULE, variable_queue_dropwhile_restart1, [Config]).
 
-variable_queue_dropwhile_restart1() ->
+variable_queue_dropwhile_restart1(Config) ->
     with_fresh_variable_queue(fun variable_queue_dropwhile_restart2/2).
 
 variable_queue_dropwhile_restart2(VQ0, QName) ->
@@ -1368,9 +1372,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, []).
+      ?MODULE, variable_queue_dropwhile_sync_restart1, [Config]).
 
-variable_queue_dropwhile_sync_restart1() ->
+variable_queue_dropwhile_sync_restart1(Config) ->
     with_fresh_variable_queue(fun variable_queue_dropwhile_sync_restart2/2).
 
 variable_queue_dropwhile_sync_restart2(VQ0, QName) ->
@@ -1408,9 +1412,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, []).
+      ?MODULE, variable_queue_restart_large_seq_id1, [Config]).
 
-variable_queue_restart_large_seq_id1() ->
+variable_queue_restart_large_seq_id1(Config) ->
     with_fresh_variable_queue(fun variable_queue_restart_large_seq_id2/2).
 
 variable_queue_restart_large_seq_id2(VQ0, QName) ->
@@ -1445,9 +1449,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, []).
+      ?MODULE, variable_queue_ack_limiting1, [Config]).
 
-variable_queue_ack_limiting1() ->
+variable_queue_ack_limiting1(Config) ->
     with_fresh_variable_queue(fun variable_queue_ack_limiting2/2).
 
 variable_queue_ack_limiting2(VQ0, _Config) ->
@@ -1473,9 +1477,9 @@ variable_queue_ack_limiting2(VQ0, _Config) ->
 
 variable_queue_purge(Config) ->
     passed = rabbit_ct_broker_helpers:rpc(Config, 0,
-      ?MODULE, variable_queue_purge1, []).
+      ?MODULE, variable_queue_purge1, [Config]).
 
-variable_queue_purge1() ->
+variable_queue_purge1(Config) ->
     with_fresh_variable_queue(fun variable_queue_purge2/2).
 
 variable_queue_purge2(VQ0, _Config) ->
@@ -1495,9 +1499,9 @@ variable_queue_purge2(VQ0, _Config) ->
 
 variable_queue_requeue(Config) ->
     passed = rabbit_ct_broker_helpers:rpc(Config, 0,
-      ?MODULE, variable_queue_requeue1, []).
+      ?MODULE, variable_queue_requeue1, [Config]).
 
-variable_queue_requeue1() ->
+variable_queue_requeue1(Config) ->
     with_fresh_variable_queue(fun variable_queue_requeue2/2).
 
 variable_queue_requeue2(VQ0, _Config) ->
@@ -1521,13 +1525,14 @@ 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, []).
+      ?MODULE, variable_queue_requeue_ram_beta1, [Config]).
 
-variable_queue_requeue_ram_beta1() ->
+variable_queue_requeue_ram_beta1(Config) ->
     with_fresh_variable_queue(fun variable_queue_requeue_ram_beta2/2).
 
 variable_queue_requeue_ram_beta2(VQ0, _Config) ->
-    Count = 2 + 2 * segment_size(),
+    IndexMod = index_mod(),
+    Count = IndexMod:next_segment_boundary(0)*2 + 2,
     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),
@@ -1803,7 +1808,8 @@ requeue_one_by_one(Acks, VQ) ->
 %% internal queues. Kept for completeness.
 variable_queue_with_holes(VQ0) ->
     Interval = 2048, %% should match vq:IO_BATCH_SIZE
-    Count = 2 * Interval + 2 * segment_size(),
+    IndexMod = index_mod(),
+    Count = IndexMod:next_segment_boundary(0)*2 + 2 * Interval,
     Seq = lists:seq(1, Count),
     VQ1 = variable_queue_publish(
             false, 1, Count,