Take live data sizes into account when grouping for compaction

kjnilsson · kjnilsson · commit 13e1bf83d712 · 2025-06-23T16:45:14.000+01:00
diff --git a/src/ra_log.erl b/src/ra_log.erl
@@ -188,7 +188,7 @@ init(#{uid := UId,
                                      segment_writer := SegWriter} = Names}
       } = Conf) ->
     Dir = server_data_dir(DataDir, UId),
-    MaxOpen = maps:get(max_open_segments, Conf, 5),
+    MaxOpen = maps:get(max_open_segments, Conf, 1),
     SnapModule = maps:get(snapshot_module, Conf, ?DEFAULT_SNAPSHOT_MODULE),
     %% this has to be patched by ra_server
     LogId = maps:get(log_id, Conf, UId),
@@ -228,8 +228,12 @@ init(#{uid := UId,
     % segments it is currently processed have been finished
     MtRange = ra_mt:range(Mt0),
     SegRefs = my_segrefs(UId, SegWriter),
+    SegmentMaxCount = maps:get(segment_max_entries, Conf, ?SEGMENT_MAX_ENTRIES),
+    SegmentMaxSize = maps:get(segment_max_size_bytes, Conf, ?SEGMENT_MAX_SIZE_B),
+    CompConf = #{max_size => SegmentMaxSize,
+                 max_count => SegmentMaxCount},
     Reader = ra_log_segments:init(UId, Dir, MaxOpen, AccessPattern, SegRefs,
-                                  Counter, LogId),
+                                  Counter, CompConf, LogId),
     SegmentRange = ra_log_segments:range(Reader),
     %% TODO: check ra_range:add/2 actually performas the correct logic we expect
     Range = ra_range:add(MtRange, SegmentRange),
@@ -1306,13 +1310,14 @@ release_resources(MaxOpenSegments, AccessPattern,
                                       counter = Counter},
                            reader = Reader} = State) ->
     ActiveSegs = ra_log_segments:segment_refs(Reader),
+    CompConf = ra_log_segments:compaction_conf(Reader),
     % close all open segments
     % deliberately ignoring return value
     _ = ra_log_segments:close(Reader),
     %% open a new segment with the new max open segment value
     State#?MODULE{reader = ra_log_segments:init(UId, Dir, MaxOpenSegments,
                                                 AccessPattern, ActiveSegs,
-                                                Counter, LogId)}.
+                                                Counter, CompConf, LogId)}.
 
 
 %%% Local functions
diff --git a/src/ra_log_segment.erl b/src/ra_log_segment.erl
@@ -23,6 +23,7 @@
          filename/1,
          segref/1,
          info/1,
+         info/2,
          is_same_as/2,
          copy/3]).
 
@@ -454,17 +455,23 @@ segref(Filename) ->
     close(Seg),
     SegRef.
 
--spec info(file:filename_all()) ->
-    #{size => non_neg_integer(),
-      max_count => non_neg_integer(),
-      file_type => regular | symlink,
-      ctime => integer(),
-      links => non_neg_integer(),
-      num_entries => non_neg_integer(),
-      ref => option(ra_log:segment_ref()),
-      indexes => ra_seq:state()
-     }.
-info(Filename)
+-type infos() :: #{size => non_neg_integer(),
+                   max_count => non_neg_integer(),
+                   file_type => regular | symlink,
+                   ctime => integer(),
+                   links => non_neg_integer(),
+                   num_entries => non_neg_integer(),
+                   ref => option(ra_log:segment_ref()),
+                   indexes => ra_seq:state(),
+                   live_size => non_neg_integer()
+                  }.
+
+-spec info(file:filename_all()) -> infos().
+info(Filename) ->
+    info(Filename, undefined).
+
+-spec info(file:filename_all(), option(ra_seq:state())) -> infos().
+info(Filename, Live0)
   when not is_tuple(Filename) ->
     %% TODO: this can be much optimised by a specialised index parsing
     %% function
@@ -475,14 +482,29 @@ info(Filename)
                     ctime = CTime}} = file:read_link_info(Filename,
                                                           [raw, {time, posix}]),
 
-    Info = #{size => Seg#state.data_write_offset,
+    AllIndexesSeq = ra_seq:from_list(maps:keys(Index)),
+    Live = case Live0 of
+               undefined ->
+                   AllIndexesSeq;
+               _ ->
+                   Live0
+           end,
+    LiveSize = ra_seq:fold(fun (I, Acc) ->
+                                   {_, _, Sz, _} = maps:get(I, Index),
+                                   Acc + Sz
+                           end, 0, Live),
+    Info = #{
+             size => Seg#state.data_write_offset,
+             index_size => Seg#state.data_start,
              file_type => T,
              links => Links,
              ctime => CTime,
              max_count => max_count(Seg),
              num_entries => maps:size(Index),
              ref => segref(Seg),
-             indexes => ra_seq:from_list(maps:keys(Index))
+             live_size => LiveSize,
+             %% TODO: this is most likely just here for debugging
+             indexes => AllIndexesSeq
             },
     close(Seg),
     Info.
diff --git a/src/ra_log_segments.erl b/src/ra_log_segments.erl
@@ -11,7 +11,7 @@
 
 -include_lib("kernel/include/file.hrl").
 -export([
-         init/7,
+         init/8,
          close/1,
          update_segments/2,
          schedule_compaction/4,
@@ -27,7 +27,8 @@
          exec_read_plan/6,
          fetch_term/2,
          info/1,
-         purge_symlinks/2
+         purge_symlinks/2,
+         compaction_conf/1
          ]).
 
 -include("ra.hrl").
@@ -36,13 +37,16 @@
 
 -define(SYMLINK_KEEPFOR_S, 60).
 
+-type compaction_conf() :: #{max_count => non_neg_integer(),
+                             max_size => non_neg_integer()}.
 -type access_pattern() :: sequential | random.
 %% holds static or rarely changing fields
 -record(cfg, {uid :: ra_uid(),
               log_id = "" :: unicode:chardata(),
               counter :: undefined | counters:counters_ref(),
               directory :: file:filename(),
-              access_pattern = random :: access_pattern()
+              access_pattern = random :: access_pattern(),
+              compaction_conf :: compaction_conf()
              }).
 
 -type segment_ref() :: ra_log:segment_ref().
@@ -74,14 +78,16 @@
 -spec init(ra_uid(), file:filename_all(), non_neg_integer(),
            access_pattern(), [segment_ref()],
            undefined | counters:counters_ref(),
+           map(),
            unicode:chardata()) -> state().
-init(UId, Dir, MaxOpen, AccessPattern, SegRefs0, Counter, LogId)
+init(UId, Dir, MaxOpen, AccessPattern, SegRefs0, Counter, CompConf, LogId)
   when is_binary(UId) ->
     Cfg = #cfg{uid = UId,
                log_id = LogId,
                counter = Counter,
                directory = Dir,
-               access_pattern = AccessPattern},
+               access_pattern = AccessPattern,
+               compaction_conf = CompConf},
     FlruHandler = fun ({_, Seg}) ->
                           _ = ra_log_segment:close(Seg),
                           decr_counter(Cfg, ?C_RA_LOG_OPEN_SEGMENTS, 1)
@@ -150,6 +156,7 @@ update_segments(NewSegmentRefs, #?STATE{open_segments = Open0,
     [ra_server:effect()].
 schedule_compaction(Type, SnapIdx, LiveIndexes,
                     #?MODULE{cfg = #cfg{log_id = LogId,
+                                        compaction_conf = CompConf,
                                         directory = Dir} = Cfg} = State) ->
     case compactable_segrefs(SnapIdx, State) of
         [] ->
@@ -171,21 +178,25 @@ schedule_compaction(Type, SnapIdx, LiveIndexes,
             Self = self(),
             Fun = fun () ->
                           ok = incr_counter(Cfg, ?C_RA_LOG_COMPACTIONS_MAJOR_COUNT, 1),
-                          MajConf = #{dir => Dir},
+                          MajConf = CompConf#{dir => Dir},
                           Result = major_compaction(MajConf, SegRefs,
                                                     LiveIndexes),
-                          %% TODO: this could be done on a timer if more
-                          %% timely symlink cleanup is needed
-                          purge_symlinks(Dir, ?SYMLINK_KEEPFOR_S),
                           %% need to update the ra_servers list of seg refs
                           %% _before_ the segments can actually be deleted
                           Self ! {ra_log_event,
                                   {compaction_result, Result}},
+                          %% TODO: this could be done on a timer if more
+                          %% timely symlink cleanup is needed
+                          purge_symlinks(Dir, ?SYMLINK_KEEPFOR_S),
                           ok
                   end,
 
             [{bg_work, Fun,
               fun (Err) ->
+                      %% send an empty compaction result to ensure the
+                      %% a future compaction can be performed (TODO:)
+                      Self ! {ra_log_event,
+                              {compaction_result, #compaction_result{}}},
                       ?WARN("~ts: Major compaction failed with ~p",
                             [LogId, Err]), ok
               end}]
@@ -266,6 +277,10 @@ segment_ref_count(#?STATE{segment_refs = SegmentRefs}) ->
 range(#?STATE{range = Range}) ->
     Range.
 
+-spec compaction_conf(state()) -> map().
+compaction_conf(#?STATE{cfg = #cfg{compaction_conf = Conf}}) ->
+    Conf.
+
 -spec num_open_segments(state()) -> non_neg_integer().
 num_open_segments(#?STATE{open_segments = Open}) ->
      ra_flru:size(Open).
@@ -572,7 +587,7 @@ list_files(Dir, Ext, Fun) ->
             []
     end.
 
-major_compaction(#{dir := Dir}, SegRefs, LiveIndexes) ->
+major_compaction(#{dir := Dir} = CompConf, SegRefs, LiveIndexes) ->
     {Compactable, Delete} =
         lists:foldl(fun({Fn0, Range} = S,
                         {Comps, Del}) ->
@@ -584,7 +599,7 @@ major_compaction(#{dir := Dir}, SegRefs, LiveIndexes) ->
                                     %% get the info map from each
                                     %% potential segment
                                     Fn = filename:join(Dir, Fn0),
-                                    Info = ra_log_segment:info(Fn),
+                                    Info = ra_log_segment:info(Fn, Seq),
                                     {[{Info, Seq, S} | Comps], Del}
                             end
                     end, {[], []}, SegRefs),
@@ -602,7 +617,8 @@ major_compaction(#{dir := Dir}, SegRefs, LiveIndexes) ->
          ok  = prim_file:delete(filename:join(Dir, F))
      end || F <- UnusedFiles],
     %% group compactable
-    CompactionGroups = compaction_groups(lists:reverse(Compactable), []),
+    CompactionGroups = compaction_groups(lists:reverse(Compactable), [],
+                                         CompConf),
     Compacted0 =
     [begin
          %% create a new segment with .compacting extension
@@ -706,31 +722,37 @@ make_links(Dir, To, From)
 with_ext(Fn, Ext) when is_binary(Fn) andalso is_list(Ext) ->
     <<(filename:rootname(Fn))/binary, (ra_lib:to_binary(Ext))/binary>>.
 
-compaction_groups([], Groups) ->
+compaction_groups([], Groups, _Conf) ->
     lists:reverse(Groups);
-compaction_groups(Infos, Groups) ->
-    case take_group(Infos, #{max_count => 128}, []) of
+compaction_groups(Infos, Groups, Conf) ->
+    case take_group(Infos, Conf, []) of
         {Group, RemInfos} ->
-            compaction_groups(RemInfos, [Group | Groups])
+            compaction_groups(RemInfos, [Group | Groups], Conf)
     end.
 
 %% TODO: try to take potential size into account
 take_group([], _, Acc) ->
     {lists:reverse(Acc), []};
-take_group([{#{num_entries := NumEnts}, Live, {_, _}} = E | Rem] = All,
-           #{max_count := Mc}, Acc) ->
+take_group([{#{num_entries := NumEnts,
+               live_size := LiveSize}, Live, {_, _}} = E | Rem] = All,
+           #{max_count := Mc,
+             max_size := MaxSz}, Acc) ->
     Num = ra_seq:length(Live),
-    case Num < NumEnts div 2 of
+    case Num / NumEnts < 0.5 of
         true ->
-            case Mc - Num < 0 of
+            case Mc - Num < 0 orelse
+                 MaxSz - LiveSize < 0 of
                 true ->
                     {lists:reverse(Acc), All};
                 false ->
-                    take_group(Rem, #{max_count => Mc - Num}, [E | Acc])
+                    take_group(Rem, #{max_count => Mc - Num,
+                                      max_size => MaxSz - LiveSize},
+                               [E | Acc])
             end;
             %% skip this secment
         false when Acc == [] ->
-            take_group(Rem, #{max_count => Mc}, Acc);
+            take_group(Rem, #{max_count => Mc,
+                              max_size => MaxSz}, Acc);
         false ->
             {lists:reverse(Acc), Rem}
     end.
diff --git a/src/ra_server_proc.erl b/src/ra_server_proc.erl
@@ -1927,6 +1927,8 @@ send_snapshots(Id, Term, {_, ToNode} = To, ChunkSize,
             case ra_snapshot:indexes(ra_snapshot:current_snapshot_dir(SnapState)) of
                 {ok, [_|_] = Indexes} ->
                     %% there are live indexes to send before the snapshot
+                    %% %% TODO: only send live indexes higher than the follower's
+                    %% last_applied index
                     Idxs = lists:reverse(ra_seq:expand(Indexes)),
                     Flru = lists:foldl(
                              fun (Is, F0) ->
@@ -1937,7 +1939,7 @@ send_snapshots(Id, Term, {_, ToNode} = To, ChunkSize,
                                                                      data = Ents},
                                      _Res1 = gen_statem:call(To, RPC1,
                                                             {dirty_timeout, InstallTimeout}),
-                                     %% TODO: assert REs1 is successful
+                                     %% TODO: assert Res1 is successful
                                      F
                              end, undefined, ra_lib:lists_chunk(16, Idxs)),
                     _ = ra_flru:evict_all(Flru),
diff --git a/test/coordination_SUITE.erl b/test/coordination_SUITE.erl
@@ -1431,8 +1431,6 @@ snapshot_installed(#{machine_version := _,
 
 node_setup(DataDir) ->
     ok = ra_lib:make_dir(DataDir),
-    % NodeDir = filename:join(DataDir, atom_to_list(node())),
-    % ok = ra_lib:make_dir(DataDir),
     LogFile = filename:join(DataDir, "ra.log"),
     SaslFile = filename:join(DataDir, "ra_sasl.log"),
     logger:set_primary_config(level, debug),
diff --git a/test/ra_log_segment_SUITE.erl b/test/ra_log_segment_SUITE.erl
@@ -38,6 +38,8 @@ all_tests() ->
      corrupted_segment,
      large_segment,
      segref,
+     info,
+     info_2,
      versions_v1,
      copy
     ].
@@ -207,6 +209,39 @@ segref(Config) ->
     {<<"seg1.seg">>, {1, 1}} = ra_log_segment:segref(Seg1),
     ok.
 
+info(Config) ->
+    Dir = ?config(data_dir, Config),
+    Fn = filename:join(Dir, "seg1.seg"),
+    {ok, Seg0} = ra_log_segment:open(Fn, #{max_count => 128}),
+    Info1 = ra_log_segment:info(Fn),
+    ?assertMatch(#{ref := undefined}, Info1),
+    {ok, Seg1} = ra_log_segment:append(Seg0, 1, 2, <<"Adsf">>),
+    _ = ra_log_segment:flush(Seg1),
+    Info2 = ra_log_segment:info(Fn),
+    ?assertMatch(#{ref := {<<"seg1.seg">>, {1, 1}}}, Info2),
+    ok.
+
+info_2(Config) ->
+    %% passes live indexes which will result in additional info keys
+    Dir = ?config(data_dir, Config),
+    Fn = filename:join(Dir, "seg1.seg"),
+    {ok, Seg0} = ra_log_segment:open(Fn, #{max_count => 128}),
+    Info1 = ra_log_segment:info(Fn, []),
+    ?assertMatch(#{ref := undefined,
+                   live_size := 0}, Info1),
+    {ok, Seg1} = ra_log_segment:append(Seg0, 1, 2, <<"Adsf">>),
+    {ok, Seg2} = ra_log_segment:append(Seg1, 2, 2, <<"Adsf">>),
+    _ = ra_log_segment:flush(Seg2),
+    Info2 = ra_log_segment:info(Fn, [1]),
+    ?assertMatch(#{ref := {<<"seg1.seg">>, {1, 2}},
+                   num_entries := 2,
+                   live_size := 4}, Info2),
+    Info3 = ra_log_segment:info(Fn),
+    %% info/1 assumes all indexes are "live"
+    ?assertMatch(#{ref := {<<"seg1.seg">>, {1, 2}},
+                   num_entries := 2,
+                   live_size := 8}, Info3),
+    ok.
 
 full_file(Config) ->
     Dir = ?config(data_dir, Config),
diff --git a/test/ra_log_segments_SUITE.erl b/test/ra_log_segments_SUITE.erl
