Use read ahead data in chunk iterator if available

Author: acogoluegnes

src/osiris_log.erl

@@ -2,7 +2,7 @@
 %% License, v. 2.0. If a copy of the MPL was not distributed with this
 %% file, You can obtain one at https://mozilla.org/MPL/2.0/.
 %%
-%% Copyright (c) 2007-2023 Broadcom. All Rights Reserved. The term Broadcom refers to Broadcom Inc. and/or its subsidiaries.
+%% Copyright (c) 2007-2025 Broadcom. All Rights Reserved. The term Broadcom refers to Broadcom Inc. and/or its subsidiaries.
 %%
 
 -module(osiris_log).
@@ -1443,8 +1443,9 @@ chunk_iterator(State) ->
     {error, {invalid_chunk_header, term()}}.
 chunk_iterator(#?MODULE{cfg = #cfg{},
                         mode = #read{type = RType,
-                                     chunk_selector = Selector}
-                       } = State0, CreditHint)
+                                     chunk_selector = Selector,
+                                     filter_size = RaFs}} = State0,
+               CreditHint)
   when (is_integer(CreditHint) andalso CreditHint > 0) orelse
        is_atom(CreditHint) ->
     %% reads the next chunk of unparsed chunk data
@@ -1458,15 +1459,20 @@ chunk_iterator(#?MODULE{cfg = #cfg{},
            data_size := DataSize,
            filter_size := FilterSize,
            position := Pos,
-           next_position := NextPos} = Header, _,
-         #?MODULE{fd = Fd, mode = #read{next_offset = ChId} = Read} = State1} ->
-            State = State1#?MODULE{mode = Read#read{next_offset = ChId + NumRecords,
-                                                    position = NextPos}},
+           next_position := NextPos} = Header, MaybeData,
+         #?MODULE{fd = Fd, mode = #read{next_offset = ChId} = Read1} = State1} ->
             case needs_handling(RType, Selector, ChType) of
                 true ->
+                    Read = Read1#read{next_offset = ChId + NumRecords,
+                                      position = NextPos,
+                                      last_data_size = DataSize,
+                                      filter_size = read_ahead_fsize(RaFs,
+                                                                     FilterSize)},
+
+                    State = State1#?MODULE{mode = Read},
                     DataPos = Pos + ?HEADER_SIZE_B + FilterSize,
                     Data = iter_read_ahead(Fd, DataPos, ChId, Crc, CreditHint,
-                                           DataSize, NumEntries),
+                                           DataSize, NumEntries, MaybeData),
                     Iterator = #iterator{fd = Fd,
                                          data = Data,
                                          next_offset = ChId,
@@ -1475,6 +1481,9 @@ chunk_iterator(#?MODULE{cfg = #cfg{},
                     {ok, Header, Iterator, State};
                 false ->
                     %% skip
+                    Read = Read1#read{next_offset = ChId + NumRecords,
+                                      position = NextPos},
+                    State = State1#?MODULE{mode = Read},
                     chunk_iterator(State, CreditHint)
             end;
         Other ->
@@ -1681,13 +1690,13 @@ send_file(Sock,
                header_data := HeaderData} = Header,
          MaybeData,
          #?MODULE{fd = Fd,
-                  mode = #read{next_offset = ChId} = Read0} = State1} ->
+                  mode = #read{next_offset = ChId} = Read1} = State1} ->
 
             %% only sendfile if either the reader is a data reader
             %% or the chunk is a user type (for offset readers)
             case needs_handling(RType, Selector, ChType) of
                 true ->
-                    Read = Read0#read{next_offset = ChId + NumRecords,
+                    Read = Read1#read{next_offset = ChId + NumRecords,
                                       position = NextPos,
                                       last_data_size = DataSize,
                                       filter_size = read_ahead_fsize(RaFs,
@@ -1733,7 +1742,7 @@ send_file(Sock,
                             end
                     end;
                 false ->
-                    Read = Read0#read{next_offset = ChId + NumRecords,
+                    Read = Read1#read{next_offset = ChId + NumRecords,
                                       position = NextPos},
                     State = State1#?MODULE{mode = Read},
                     %% skip chunk and recurse
@@ -3362,16 +3371,21 @@ dump_crc_check(Fd) ->
             dump_crc_check(Fd)
     end.
 
-iter_read_ahead(_Fd, _Pos, _ChunkId, _Crc, 1, _DataSize, _NumEntries) ->
+iter_read_ahead(_Fd, _Pos, _ChunkId, _Crc, _Credit, DataSize, _NumEntries, RAD)
+  when RAD =/= undefined ->
+    <<Data:DataSize/binary, _/binary>> = RAD,
+    Data;
+iter_read_ahead(_Fd, _Pos, _ChunkId, _Crc, 1, _DataSize, _NumEntries, undefined) ->
+    %% FIXME is it about credit = 1 or number of entries = 1?
     %% no point reading ahead if there is only one entry to be read at this
     %% time
     undefined;
-iter_read_ahead(Fd, Pos, ChunkId, Crc, Credit, DataSize, NumEntries)
+iter_read_ahead(Fd, Pos, ChunkId, Crc, Credit, DataSize, NumEntries, undefined)
   when Credit == all orelse NumEntries == 1 ->
     {ok, Data} = file:pread(Fd, Pos, DataSize),
     validate_crc(ChunkId, Crc, Data),
     Data;
-iter_read_ahead(Fd, Pos, _ChunkId, _Crc, Credit0, DataSize, NumEntries) ->
+iter_read_ahead(Fd, Pos, _ChunkId, _Crc, Credit0, DataSize, NumEntries, undefined) ->
     %% read ahead, assumes roughly equal entry sizes which may not be the case
     %% TODO round up to nearest block?
     %% We can only practically validate CRC if we read the whole data

test/osiris_log_SUITE.erl

@@ -38,6 +38,7 @@ all_tests() ->
      subbatch,
      subbatch_compressed,
      iterator_read_chunk,
+     iterator_read_chunk_with_read_ahead,
      iterator_read_chunk_mixed_sizes_with_credit,
      read_chunk_parsed,
      read_chunk_parsed_2,
@@ -344,14 +345,16 @@ iterator_read_chunk(Config) ->
     EntriesRev = [Batch,
                   <<"ho">>,
                   {<<"filter">>, <<"hi">>}],
-    {ChId, _S1} = write_committed(EntriesRev, S0),
-    {ok, _H, I0, _R2} = osiris_log:chunk_iterator(R1),
+    {ChId, S1} = write_committed(EntriesRev, S0),
+    {ok, _H, I0, R2} = osiris_log:chunk_iterator(R1),
     HoOffs = ChId + 1,
     BatchOffs = ChId + 2,
     {{ChId, <<"hi">>}, I1} = osiris_log:iterator_next(I0),
     {{HoOffs, <<"ho">>}, I2} = osiris_log:iterator_next(I1),
     {{BatchOffs, Batch}, I} = osiris_log:iterator_next(I2),
     ?assertMatch(end_of_chunk, osiris_log:iterator_next(I)),
+    osiris_log:close(R2),
+    osiris_log:close(S1),
     ok.
 
 iterator_read_chunk_mixed_sizes_with_credit(Config) ->
@@ -365,16 +368,79 @@ iterator_read_chunk_mixed_sizes_with_credit(Config) ->
     EntriesRev = [Big,
                   <<"ho">>,
                   {<<"filter">>, <<"hi">>}],
-    {ChId, _S1} = write_committed(EntriesRev, S0),
+    {ChId, S1} = write_committed(EntriesRev, S0),
     %% this is a less than ideal case where we have one large and two very
-    %% small entries inthe same batch. The read ahead only
-    {ok, _H, I0, _R2} = osiris_log:chunk_iterator(R1, 2),
+    %% small entries in the same batch. We read ahead only the 2 first entries.
+    {ok, _H, I0, R2} = osiris_log:chunk_iterator(R1, 2),
     HoOffs = ChId + 1,
     BigOffs = ChId + 2,
     {{ChId, <<"hi">>}, I1} = osiris_log:iterator_next(I0),
     {{HoOffs, <<"ho">>}, I2} = osiris_log:iterator_next(I1),
     {{BigOffs, Big}, I} = osiris_log:iterator_next(I2),
     ?assertMatch(end_of_chunk, osiris_log:iterator_next(I)),
+    osiris_log:close(R2),
+    osiris_log:close(S1),
+    ok.
+
+iterator_read_chunk_with_read_ahead(Config) ->
+    %% the test makes sure reading ahead on header reading does not break
+    %% the iterator
+    RAL = 4096, %% read ahead limit
+    Conf = ?config(osiris_conf, Config),
+    W = osiris_log:init(Conf),
+    Shared = osiris_log:get_shared(W),
+    RConf = Conf#{shared => Shared},
+    {ok, R} = osiris_log:init_offset_reader(0, RConf),
+    Tests =
+    [
+     fun(#{w := W0, r := R0}) ->
+             %% first chunk, there won't be any data size hints in the reader
+             EntriesRev = [<<"hi">>, <<"ho">>],
+             {_, W1} = write_committed(EntriesRev, W0),
+             {ok, H, I0, R1} = osiris_log:chunk_iterator(R0),
+             {{_, <<"ho">>}, I1} = osiris_log:iterator_next(I0),
+             {{_, <<"hi">>}, I2} = osiris_log:iterator_next(I1),
+             ?assertMatch(end_of_chunk, osiris_log:iterator_next(I2)),
+             {H, W1, R1}
+     end,
+     fun(#{w := W0, r := R0}) ->
+             %% this one will be read ahead
+             EntriesRev = [<<"foo">>, <<"bar">>],
+             {_, W1} = write_committed(EntriesRev, W0),
+             {ok, H, I0, R1} = osiris_log:chunk_iterator(R0),
+             {{_, <<"bar">>}, I1} = osiris_log:iterator_next(I0),
+             {{_, <<"foo">>}, I2} = osiris_log:iterator_next(I1),
+             ?assertMatch(end_of_chunk, osiris_log:iterator_next(I2)),
+             {H, W1, R1}
+     end,
+     fun(#{w := W0, r := R0}) ->
+             %% this one will be read ahead
+             E1 = rand:bytes(RAL - 100),
+             EntriesRev = [E1 , <<"aaa">>],
+             {_, W1} = write_committed(EntriesRev, W0),
+             {ok, H, I0, R1} = osiris_log:chunk_iterator(R0),
+             {{_, <<"aaa">>}, I1} = osiris_log:iterator_next(I0),
+             {{_, E1}, I2} = osiris_log:iterator_next(I1),
+             ?assertMatch(end_of_chunk, osiris_log:iterator_next(I2)),
+             {H, W1, R1}
+     end,
+     fun(#{w := W0, r := R0}) ->
+             %% this one is too big to be read ahead
+             E1 = rand:bytes(RAL * 2),
+             EntriesRev = [E1 , <<"aaa">>],
+             {_, W1} = write_committed(EntriesRev, W0),
+             {ok, H, I0, R1} = osiris_log:chunk_iterator(R0),
+             {{_, <<"aaa">>}, I1} = osiris_log:iterator_next(I0),
+             {{_, E1}, I2} = osiris_log:iterator_next(I1),
+             ?assertMatch(end_of_chunk, osiris_log:iterator_next(I2)),
+             {H, W1, R1}
+     end
+    ],
+
+    #{w := Wr1, r := Rd1} = run_read_ahead_tests(Tests, offset,
+                                                 ?DEFAULT_FILTER_SIZE, W, R),
+    osiris_log:close(Rd1),
+    osiris_log:close(Wr1),
     ok.
 
 read_chunk_parsed(Config) ->