CQ: Tune reads from index based on message rate

We want to reduce the number of times we cross over to the next
segment when reading. We do this by computing a threshold based
on read rate, then reading a little more if we are close enough
to the end of the current segment, and a little less if we are
not reading enough into the next segment.

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, next_segment_boundary/1, info/1,
+         bounds/2, tune_read/2, info/1,
          terminate/3, delete_and_terminate/1,
          publish/7, ack/2, read/3,
          sync/1, needs_sync/1]).
@@ -757,10 +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.
+%% Before calling read/3 it is recommended to call tune_read/2
+%% so that the index can inform the queue how to most efficiently
+%% read messages, as the index has knowledge of segment boundaries
+%% and can decide whether it is worth it to read from one or two
+%% segments at a time.
 
 -spec read(rabbit_variable_queue:seq_id(),
            rabbit_variable_queue:seq_id(),
@@ -1064,13 +1065,62 @@ bounds(State = #qi{ segments = Segments }, NextSeqIdHint) ->
              State}
     end.
 
+%% We tune the read request so that we only cross over segment
+%% boundaries when it makes sense. We compute a threshold
+%% based on the number of messages requested, and look whether
+%% the number of messages to be read from the next segment is
+%% higher than the threshold, otherwise we stop at the segment
+%% boundary. Similarly, if the number of messages requested
+%% almost reaches the segment boundary, we read a few more
+%% messages up to the segment boundary.
+%%
+%% This is meant to reduce the number of file reads when the
+%% outgoing message rate is moderate, while still sticking to
+%% segment file boundaries when the message rate is maxed.
+
+-spec tune_read(rabbit_variable_queue:seq_id(), rabbit_variable_queue:seq_id())
+    -> rabbit_variable_queue:seq_id().
+
+tune_read(FromSeqId, ToSeqId)
+        when FromSeqId =:= ToSeqId ->
+    %% Nothing will be read as From is inclusive but To is exclusive.
+    ToSeqId;
+tune_read(FromSeqId, ToSeqId) ->
+    %% How much we are reading.
+    ReqCount = ToSeqId - FromSeqId,
+    %% How much remains in the current segment.
+    NextSeqId = next_segment_boundary(FromSeqId),
+    RemCount = NextSeqId - FromSeqId,
+    %% How much we are willing to accept as extra messages to read.
+    Threshold = max(1, ReqCount div 7),
+    if
+        %% There are messages remaining in the segment, and the number
+        %% of messages remaining is less than the threshold: read up
+        %% to the end of the segment (To is exclusive).
+        (RemCount >= ReqCount) andalso (RemCount - ReqCount =< Threshold) ->
+            NextSeqId;
+        %% There are messages remaining in the segment but the number
+        %% of messages remaining is more than we are willing to read:
+        %% only read what was originally requested.
+        (RemCount >= ReqCount) ->
+            ToSeqId;
+        %% We are requested to read past the end of the current segment.
+        %% This would require us to read from two different segments,
+        %% which we only want to do if this involves a good number of
+        %% messages. If this number is below the threshold, we reduce
+        %% the number of messages to read.
+        (ReqCount - RemCount =< Threshold) ->
+            NextSeqId;
+        %% Otherwise we cross over into the next segment, meaning we
+        %% only read what was originally requested.
+        true ->
+            ToSeqId
+    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.
 

deps/rabbit/src/rabbit_variable_queue.erl

@@ -1866,9 +1866,7 @@ read_from_q_tail(DelsAndAcksFun,
     %% 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 = rabbit_classic_queue_index_v2:tune_read(QTailSeqId, 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_entry_count() ->
+    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_entry_count(),
     TwoSegs = SegmentSize + SegmentSize,
     MostOfASegment = trunc(SegmentSize*0.75),
     SeqIdsA = lists:seq(0, MostOfASegment-1),
@@ -993,7 +996,7 @@ v2_delete_segment_file_completely_acked(Config) ->
 
 v2_delete_segment_file_completely_acked1(_Config) ->
     IndexMod = rabbit_classic_queue_index_v2,
-    SegmentSize = IndexMod:next_segment_boundary(0),
+    SegmentSize = segment_entry_count(),
     SeqIds = lists:seq(0, SegmentSize - 1),
 
     with_empty_test_queue(
@@ -1020,7 +1023,7 @@ v2_delete_segment_file_partially_acked(Config) ->
 
 v2_delete_segment_file_partially_acked1(_Config) ->
     IndexMod = rabbit_classic_queue_index_v2,
-    SegmentSize = IndexMod:next_segment_boundary(0),
+    SegmentSize = segment_entry_count(),
     SeqIds = lists:seq(0, SegmentSize div 2),
     SeqIdsLen = length(SeqIds),
 
@@ -1048,7 +1051,7 @@ v2_delete_segment_file_partially_acked_with_holes(Config) ->
 
 v2_delete_segment_file_partially_acked_with_holes1(_Config) ->
     IndexMod = rabbit_classic_queue_index_v2,
-    SegmentSize = IndexMod:next_segment_boundary(0),
+    SegmentSize = segment_entry_count(),
     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_entry_count(),
     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_entry_count(),
     HalfSegment = SegmentSize div 2,
     OneAndAHalfSegment = SegmentSize + HalfSegment,
     VQ1 = variable_queue_publish(true, OneAndAHalfSegment, VQ0),
@@ -1194,12 +1195,9 @@ 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}]),
+            %% The length is the only predictible stat we have since
+            %% the contents of q_head and q_tail depend on the rate.
+            [{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 +1207,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_entry_count(),
     VQ1 = variable_queue_publish(true, Count, VQ0),
     VQ2 = variable_queue_publish(false, Count, VQ1),
     {VQ4, _AckTags}  = variable_queue_fetch(Count, true, false,
@@ -1234,9 +1231,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 +1249,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 +1272,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 +1286,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 +1332,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 +1369,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 +1409,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 +1446,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 +1474,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 +1496,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 +1522,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_entry_count(),
     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 +1804,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_entry_count(),
     Seq = lists:seq(1, Count),
     VQ1 = variable_queue_publish(
             false, 1, Count,