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

@@ -10,7 +10,7 @@
 -export([erase/1, init/1, reset_state/1, recover/4,
          bounds/2, next_segment_boundary/1, info/1,
          terminate/3, delete_and_terminate/1,
-         publish/7, ack/2, read/3,
+         publish/7, ack/2, tune_read/2, read/3,
          sync/1, needs_sync/1]).
 
 %% Recovery. Unlike other functions in this module, these
@@ -1067,6 +1067,52 @@ bounds(State = #qi{ segments = Segments }, NextSeqIdHint) ->
 %% The next_segment_boundary/1 function is used internally when
 %% reading. It should not be called from rabbit_variable_queue.
 
+%% @todo OK so segment boundaries are still good but we need
+%%       to be smarter about them. The queue can give the
+%%       read it is about to do and the index can decide to
+%%       lower or increase the limit to match segments (so
+%%       if we are close to a segment we increase otherwise
+%%       we lower, based on how many reads are requested).
+
+-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.
+
 -spec next_segment_boundary(SeqId) -> SeqId when SeqId::rabbit_variable_queue:seq_id().
 
 next_segment_boundary(SeqId) ->

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 ->