Only enable read ahead if prior chunk can be full covered

by the next read-ahead.

Author: kjnilsson

src/osiris_log.erl

@@ -3354,15 +3354,17 @@ ra_read(ReadPos, Len, #ra{buf = {Pos, Data}})
 ra_read(_Pos, _Len, _Ra) ->
     undefined.
 
-ra_update_size(undefined, _FilterSize, LastDataSize,
+ra_update_size(undefined, FilterSize, LastDataSize,
                #ra{on = true, size = Sz} = Ra)
   when Sz < ?READ_AHEAD_LIMIT andalso
-       LastDataSize < ?READ_AHEAD_LIMIT ->
+       LastDataSize =< (?READ_AHEAD_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_update_size(undefined, _FilterSize, LastDataSize,
+ra_update_size(undefined, FilterSize, LastDataSize,
                #ra{on = true, size = ?READ_AHEAD_LIMIT} = Ra)
-  when LastDataSize < ?READ_AHEAD_LIMIT ->
+  when LastDataSize =< (?READ_AHEAD_LIMIT - ?HEADER_SIZE_B -
+                        FilterSize - ?REC_HDR_SZ_SUBBATCH_B) ->
     Ra;
 ra_update_size(_Filter, FilterSize, _LastDataSize, #ra{size = Sz} = Ra) ->
     case FilterSize + ?HEADER_SIZE_B of
@@ -3444,6 +3446,16 @@ ra_update_size_test() ->
 
     ?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,
+    ?assertMatch(#ra{size = DefSize},
+                 ra_update_size(undefined, ?DEFAULT_FILTER_SIZE, MaxEnablingDataSize + 1 ,
+                                Ra0)),
+    ?assertMatch(#ra{size = ?READ_AHEAD_LIMIT},
+                 ra_update_size(undefined, ?DEFAULT_FILTER_SIZE, MaxEnablingDataSize,
+                                Ra0)),
     ok.
 
 part_test() ->

test/osiris_log_SUITE.erl

@@ -2199,7 +2199,7 @@ read_ahead_send_file_filter(Config) ->
     RAL = 4096, %% read ahead limit
     HS = ?HEADER_SIZE_B,
     %% we store the entry size on 4 bytes, so we must substract them from the data size
-    MaxEntrySize = RAL - 4 - HS,
+    MaxEntrySize = RAL - 11 - HS,
     DFS = ?DEFAULT_FILTER_SIZE,
     % FilterSizes = [DFS * 2],
     FilterSizes = [DFS, DFS * 2],
@@ -2307,50 +2307,47 @@ read_ahead_send_file_filter(Config) ->
                   {_, W1} = write_committed(Entries, W0),
                   W1;
              (read, #{r := R0, cs := CS, ss := SS, tracer := T}) ->
-                  %% we still read ahead, so we read a small chunk at once
+                  %% read ahead is disabled, but will be reenabled
                   Entries = [{<<"banana">>, <<"aaa">>}],
                   [_, _, D, _] = ra_fake_chunk(Entries),
                   {ok, R1} = osiris_log:send_file(CS, R0),
                   {ok, Read} = recv(SS, byte_size(D) + HS),
                   ?assertEqual(D, binary:part(Read, HS, byte_size(Read) - HS)),
                   ?assertEqual(1, osiris_tracer:call_count(T, file, pread)),
-                  ?assertEqual(0, osiris_tracer:call_count(T, file, sendfile)),
+                  ?assertEqual(1, osiris_tracer:call_count(T, file, sendfile)),
                   R1
           end,
           fun(write, #{w := W0}) ->
-                  EData = binary:copy(<<"b">>, MES1 * 2),
+                  EData = binary:copy(<<"b">>, MES1 div 2),
                   Entries = [{<<"banana">>, EData}],
                   {_, W1} = write_committed(Entries, W0),
                   W1;
              (read, #{r := R0, cs := CS, ss := SS, tracer := T}) ->
-                  %% large chunk, we had the header already,
-                  %% but we must use sendfile for the data
-                  EData = binary:copy(<<"b">>, MES1 * 2),
+                  %% previous chunk enabled read ahead
+                  EData = binary:copy(<<"b">>, MES1 div 2),
                   Entries = [{<<"banana">>, EData}],
                   [_, _, D, _] = ra_fake_chunk(Entries),
                   {ok, R1} = osiris_log:send_file(CS, R0),
                   {ok, Read} = recv(SS, byte_size(D) + HS),
                   ?assertEqual(D, binary:part(Read, HS, byte_size(Read) - HS)),
-                  ?assertEqual(0, osiris_tracer:call_count(T, file, pread)),
-                  ?assertEqual(1, osiris_tracer:call_count(T, file, sendfile)),
+                  ?assertEqual(1, osiris_tracer:call_count(T, file, pread)),
+                  ?assertEqual(0, osiris_tracer:call_count(T, file, sendfile)),
                   R1
           end,
           fun(write, #{w := W0}) ->
-                  EData = binary:copy(<<"b">>, MES1 div 2),
+                  EData = binary:copy(<<"b">>, MES1 * 2),
                   Entries = [{<<"banana">>, EData}],
                   {_, W1} = write_committed(Entries, W0),
                   W1;
              (read, #{r := R0, cs := CS, ss := SS, tracer := T}) ->
-                  %% we don't read ahead anymore because the previous chunk was large
-                  %% so we read the header and use sendfile,
-                  %% even for a small chunk
-                  EData = binary:copy(<<"b">>, MES1 div 2),
+                  %% large chunk, we had the header already,
+                  EData = binary:copy(<<"b">>, MES1 * 2),
                   Entries = [{<<"banana">>, EData}],
                   [_, _, D, _] = ra_fake_chunk(Entries),
                   {ok, R1} = osiris_log:send_file(CS, R0),
                   {ok, Read} = recv(SS, byte_size(D) + HS),
                   ?assertEqual(D, binary:part(Read, HS, byte_size(Read) - HS)),
-                  ?assertEqual(1, osiris_tracer:call_count(T, file, pread)),
+                  ?assertEqual(0, osiris_tracer:call_count(T, file, pread)),
                   ?assertEqual(1, osiris_tracer:call_count(T, file, sendfile)),
                   R1
           end