osiris_log: Read read-ahead limit from application environment

Previously the `READ_AHEAD_LIMIT` was a constant fixed at 4096. For some
usage patterns setting the limit higher helps consumer throughput when
the chunk size is medium/small - specifically streams published at
medium throughput being consumed over a TLS connection.

Author: the-mikedavis

src/osiris_log.erl

@@ -107,7 +107,6 @@
         _/binary>>).
 
 -define(SKIP_SEARCH_JUMP, 2048).
--define(READ_AHEAD_LIMIT, 4096).
 
 %% Specification of the Log format.
 %%
@@ -425,7 +424,8 @@
 -record(ra,
         {on = true :: boolean(),
          size = ?HEADER_SIZE_B + ?DEFAULT_FILTER_SIZE :: non_neg_integer(),
-         buf :: undefined | {Pos :: non_neg_integer(), binary()}
+         buf :: undefined | {Pos :: non_neg_integer(), binary()},
+         limit = read_ahead_limit() :: pos_integer()
         }).
 -record(read,
         {type :: data | offset,
@@ -3335,7 +3335,7 @@ iter_read_ahead(Fd, Pos, MinReqSize, Credit0, DataSize, NumEntries, Ra0)
                     %% needed to serve that, else we read up to the readahead
                     %% limit but not beyond the end of the chunk and not less
                     %% that the minimum request size
-                    MinSize = max(MinReqSize, min(?READ_AHEAD_LIMIT, DataSize)),
+                    MinSize = max(MinReqSize, min(read_ahead_limit(), DataSize)),
                     Size = max(MinSize, iter_guess_size(Credit0, NumEntries,
                                                         DataSize)),
                     {ok, Data} = file:pread(Fd, Pos, Size),
@@ -3364,15 +3364,15 @@ ra_read(_Pos, _Len, _Ra) ->
     undefined.
 
 ra_update_size(undefined, FilterSize, LastDataSize,
-               #ra{on = true, size = Sz} = Ra)
-  when Sz < ?READ_AHEAD_LIMIT andalso
-       LastDataSize =< (?READ_AHEAD_LIMIT - ?HEADER_SIZE_B -
+               #ra{on = true, size = Sz, limit = Limit} = Ra)
+  when Sz < Limit andalso
+       LastDataSize =< (Limit - ?HEADER_SIZE_B -
                        FilterSize - ?REC_HDR_SZ_SUBBATCH_B) ->
     %% no filter and last data size was small so enable data read ahead
-    Ra#ra{size = ?READ_AHEAD_LIMIT};
+    Ra#ra{size = Limit};
 ra_update_size(undefined, FilterSize, LastDataSize,
-               #ra{on = true, size = ?READ_AHEAD_LIMIT} = Ra)
-  when LastDataSize =< (?READ_AHEAD_LIMIT - ?HEADER_SIZE_B -
+               #ra{on = true, size = Limit, limit = Limit} = Ra)
+  when LastDataSize =< (Limit - ?HEADER_SIZE_B -
                         FilterSize - ?REC_HDR_SZ_SUBBATCH_B) ->
     Ra;
 ra_update_size(_Filter, FilterSize, _LastDataSize, #ra{size = Sz} = Ra) ->
@@ -3434,35 +3434,39 @@ write_in_chunks(ToWrite, MsgsPerChunk, Msg, W0) when ToWrite > 0 ->
 write_in_chunks(_, _, _, W) ->
     W.
 
+read_ahead_limit() ->
+    application:get_env(osiris, read_ahead_limit, 4096).
+
 -ifdef(TEST).
 -include_lib("eunit/include/eunit.hrl").
 
 ra_update_size_test() ->
+    ReadAheadLimit = read_ahead_limit(),
     DefSize = ?HEADER_SIZE_B + ?DEFAULT_FILTER_SIZE,
     ?assertMatch(#ra{size = DefSize}, #ra{}),
     Ra0 = #ra{},
-    ?assertMatch(#ra{size = ?READ_AHEAD_LIMIT},
+    ?assertMatch(#ra{size = ReadAheadLimit},
                  ra_update_size(undefined, ?DEFAULT_FILTER_SIZE, 100, Ra0)),
 
-    ?assertMatch(#ra{size = ?READ_AHEAD_LIMIT},
+    ?assertMatch(#ra{size = ReadAheadLimit},
                  ra_update_size(undefined, ?DEFAULT_FILTER_SIZE, 100, Ra0)),
-    Ra1 = #ra{size = ?READ_AHEAD_LIMIT},
+    Ra1 = #ra{size = ReadAheadLimit},
     ?assertMatch(#ra{size = DefSize},
                  ra_update_size(undefined, ?DEFAULT_FILTER_SIZE, 5000, Ra1)),
 
-    ?assertMatch(#ra{size = ?READ_AHEAD_LIMIT},
+    ?assertMatch(#ra{size = ReadAheadLimit},
                  ra_update_size(undefined, ?DEFAULT_FILTER_SIZE, 100, Ra1)),
 
     ?assertMatch(#ra{size = DefSize},
                  ra_update_size("a filter", ?DEFAULT_FILTER_SIZE, 100, Ra0)),
 
     %% we need to ensure that if we enable read ahead we can at least fulfil
     %% the prior chunk including header filter and record length header
-    MaxEnablingDataSize = ?READ_AHEAD_LIMIT - ?HEADER_SIZE_B - ?DEFAULT_FILTER_SIZE - ?REC_HDR_SZ_SUBBATCH_B,
+    MaxEnablingDataSize = ReadAheadLimit - ?HEADER_SIZE_B - ?DEFAULT_FILTER_SIZE - ?REC_HDR_SZ_SUBBATCH_B,
     ?assertMatch(#ra{size = DefSize},
                  ra_update_size(undefined, ?DEFAULT_FILTER_SIZE, MaxEnablingDataSize + 1 ,
                                 Ra0)),
-    ?assertMatch(#ra{size = ?READ_AHEAD_LIMIT},
+    ?assertMatch(#ra{size = ReadAheadLimit},
                  ra_update_size(undefined, ?DEFAULT_FILTER_SIZE, MaxEnablingDataSize,
                                 Ra0)),
     ok.