Add stream.read_ahead configuration entry

This is a switch for the read ahead optimization in Osiris.

Author: acogoluegnes

deps/rabbit/Makefile

@@ -119,6 +119,7 @@ define PROJECT_ENV
       	{dead_letter_worker_consumer_prefetch, 32},
       	{dead_letter_worker_publisher_confirm_timeout, 180000},
 		{vhost_process_reconciliation_run_interval, 30},
+		{stream_read_ahead, true},
 		%% for testing
 		{vhost_process_reconciliation_enabled, true},
 		{license_line, "Licensed under the MPL 2.0. Website: https://rabbitmq.com"}

deps/rabbit/priv/schema/rabbit.schema

@@ -2785,6 +2785,9 @@ fun(Conf) ->
     end
 end}.
 
+{mapping, "stream.read_ahead", "rabbit.stream_read_ahead",
+    [{datatype, {enum, [true, false]}}]}.
+
 {mapping, "cluster_tags.$tag", "rabbit.cluster_tags", [
     {datatype, [binary]}
 ]}.

deps/rabbit/src/rabbit_stream_queue.erl

@@ -51,6 +51,7 @@
 -export([format_osiris_event/2]).
 -export([update_stream_conf/2]).
 -export([readers/1]).
+-export([read_ahead_on/0]).
 
 -export([parse_offset_arg/1,
          filter_spec/1]).
@@ -463,10 +464,11 @@ query_local_pid(#stream_client{stream_id = StreamId} = State) ->
 begin_stream(#stream_client{name = QName,
                             readers = Readers0,
                             local_pid = LocalPid} = State,
-             Tag, Offset, Mode, AckRequired, Filter, Options)
+             Tag, Offset, Mode, AckRequired, Filter, Options0)
   when is_pid(LocalPid) ->
     CounterSpec = {{?MODULE, QName, Tag, self()}, []},
-    {ok, Seg0} = osiris:init_reader(LocalPid, Offset, CounterSpec, Options),
+    Options1 = Options0#{read_ahead => read_ahead_on()},
+    {ok, Seg0} = osiris:init_reader(LocalPid, Offset, CounterSpec, Options1),
     NextOffset = osiris_log:next_offset(Seg0) - 1,
     osiris:register_offset_listener(LocalPid, NextOffset),
     StartOffset = case Offset of
@@ -491,7 +493,7 @@ begin_stream(#stream_client{name = QName,
                    last_consumed_offset = StartOffset,
                    log = Seg0,
                    filter = Filter,
-                   reader_options = Options},
+                   reader_options = Options1},
     {ok, State#stream_client{readers = Readers0#{Tag => Str0}}}.
 
 cancel(_Q, #{consumer_tag := ConsumerTag,
@@ -659,8 +661,8 @@ handle_event(_QName, {stream_local_member_change, Pid},
                                  osiris_log:close(Log0),
                                  CounterSpec = {{?MODULE, QName, self()}, []},
                                  ?LOG_DEBUG("Re-creating Osiris reader for consumer ~tp at offset ~tp "
-                                                  " with options ~tp",
-                                                  [T, Offset, Options]),
+                                            " with options ~tp",
+                                            [T, Offset, Options]),
                                  {ok, Log1} = osiris:init_reader(Pid, Offset, CounterSpec, Options),
                                  NextOffset = osiris_log:next_offset(Log1) - 1,
                                  ?LOG_DEBUG("Registering offset listener at offset ~tp", [NextOffset]),
@@ -1527,3 +1529,6 @@ queue_vm_stats_sups() ->
 queue_vm_ets() ->
     {[],
      []}.
+
+read_ahead_on() ->
+    application:get_env(rabbit, stream_read_ahead, true).

deps/rabbit/test/config_schema_SUITE_data/rabbit.snippets

@@ -1223,6 +1223,28 @@ credential_validator.regexp = ^abc\\d+",
    [{osiris, [
       {port_range, {4100, 4600}}
      ]}],
+   []},
+
+  %%
+  %% Stream read ahead on/off
+  %%
+
+  {stream_read_ahead,
+   "
+    stream.read_ahead = true
+   ",
+   [{rabbit, [
+      {stream_read_ahead, true}
+     ]}],
+   []},
+
+  {stream_read_ahead,
+   "
+    stream.read_ahead = false
+   ",
+   [{rabbit, [
+      {stream_read_ahead, false}
+     ]}],
    []}
 
 ].

deps/rabbitmq_stream/src/rabbit_stream_reader.erl

@@ -2812,11 +2812,14 @@ init_reader(ConnectionTransport,
             Properties,
             OffsetSpec) ->
     CounterSpec = {{?MODULE, QueueResource, SubscriptionId, self()}, []},
-    Options = maps:merge(#{transport => ConnectionTransport,
-                           chunk_selector => get_chunk_selector(Properties)},
-                         rabbit_stream_utils:filter_spec(Properties)),
-    {ok, Segment} =
-        osiris:init_reader(LocalMemberPid, OffsetSpec, CounterSpec, Options),
+    Options0 = #{transport => ConnectionTransport,
+                 chunk_selector => get_chunk_selector(Properties),
+                 read_ahead => rabbit_stream_queue:read_ahead_on()},
+
+    Options1 = maps:merge(Options0,
+                          rabbit_stream_utils:filter_spec(Properties)),
+    {ok, Segment} = osiris:init_reader(LocalMemberPid, OffsetSpec,
+                                       CounterSpec, Options1),
     ?LOG_DEBUG("Next offset for subscription ~tp is ~tp",
                      [SubscriptionId, osiris_log:next_offset(Segment)]),
     Segment.