Add strategy for automatically triggering major compactions.

Two strategies now exist:
1. manual - no automatic triggering of major compactions.
2. {num_minors, Num} - promotes a minor compaction to major
after N minor compactions have occurred. Default: 8

Author: kjnilsson

src/ra.hrl

@@ -223,6 +223,7 @@
 -define(SEGMENT_MAX_ENTRIES, 4096).
 -define(SEGMENT_MAX_PENDING, 1024).
 -define(SEGMENT_MAX_SIZE_B, 64_000_000). %% set an upper limit on segment sizing
+-define(DEF_MAJOR_COIMPACTION_STRAT, {num_minors, 8}).
 
 %% logging shim
 -define(DEBUG_IF(Bool, Fmt, Args),
@@ -285,6 +286,8 @@
           "Number of checkpoint bytes written"},
          {checkpoints_promoted, ?C_RA_LOG_CHECKPOINTS_PROMOTED, counter,
           "Number of checkpoints promoted to snapshots"},
+         {minor_compactions, ?C_RA_LOG_COMPACTIONS_MINOR_COUNT, counter,
+          "Number of requested minor compactions"},
          {major_compactions, ?C_RA_LOG_COMPACTIONS_MAJOR_COUNT, counter,
           "Number of requested major compactions"},
          {major_compaction_segments_written,
@@ -311,10 +314,11 @@
 -define(C_RA_LOG_CHECKPOINTS_WRITTEN, 13).
 -define(C_RA_LOG_CHECKPOINT_BYTES_WRITTEN, 14).
 -define(C_RA_LOG_CHECKPOINTS_PROMOTED, 15).
--define(C_RA_LOG_COMPACTIONS_MAJOR_COUNT, 16).
--define(C_RA_LOG_COMPACTIONS_SEGMENTS_WRITTEN, 17).
--define(C_RA_LOG_COMPACTIONS_SEGMENTS_COMPACTED, 18).
--define(C_RA_LOG_RESERVED, 19).
+-define(C_RA_LOG_COMPACTIONS_MINOR_COUNT, 16).
+-define(C_RA_LOG_COMPACTIONS_MAJOR_COUNT, 17).
+-define(C_RA_LOG_COMPACTIONS_SEGMENTS_WRITTEN, 18).
+-define(C_RA_LOG_COMPACTIONS_SEGMENTS_COMPACTED, 19).
+-define(C_RA_LOG_RESERVED, 20).
 
 -define(C_RA_SRV_AER_RECEIVED_FOLLOWER, ?C_RA_LOG_RESERVED + 1).
 -define(C_RA_SRV_AER_REPLIES_SUCCESS, ?C_RA_LOG_RESERVED + 2).

src/ra_log.erl

@@ -141,7 +141,9 @@
                               snapshot_module => module(),
                               counter => counters:counters_ref(),
                               initial_access_pattern => sequential | random,
-                              max_checkpoints => non_neg_integer()}.
+                              max_checkpoints => non_neg_integer(),
+                              major_compaction_strategy =>
+                                  ra_log_segments:major_compaction_strategy()}.
 
 
 -type overview() ::
@@ -233,7 +235,10 @@ init(#{uid := UId,
     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),
+    MajorCompStrat = maps:get(major_compaction_strategy, Conf,
+                              ?DEF_MAJOR_COIMPACTION_STRAT),
     CompConf = #{max_size => SegmentMaxSize,
+                 major_strategy => MajorCompStrat,
                  max_count => SegmentMaxCount},
     Reader = ra_log_segments:init(UId, Dir, MaxOpen, AccessPattern, SegRefs,
                                   Counter, CompConf, LogId),

src/ra_log_segment_writer.erl

@@ -114,7 +114,9 @@ init([#{data_dir := DataDir,
         name := SegWriterName,
         system := System} = Conf]) ->
     process_flag(trap_exit, true),
-    CRef = ra_counters:new(SegWriterName, ?COUNTER_FIELDS, #{ra_system => System, module => ?MODULE}),
+    CRef = ra_counters:new(SegWriterName, ?COUNTER_FIELDS,
+                           #{ra_system => System,
+                             module => ?MODULE}),
     SegmentConf = maps:get(segment_conf, Conf, #{}),
     maybe_upgrade_segment_file_names(System, DataDir),
     {ok, #state{system = System,

src/ra_log_segments.erl

@@ -39,8 +39,15 @@
 
 -define(SYMLINK_KEEPFOR_S, 60).
 
+%% type for configuring automatic major compaction strategies
+-type major_compaction_strategy() :: manual |
+                          {num_minors, pos_integer()}.
+
 -type compaction_conf() :: #{max_count => non_neg_integer(),
-                             max_size => non_neg_integer()}.
+                             max_size => non_neg_integer(),
+                             major_strategy => major_compaction_strategy()}.
+%% hardly every used anymore, the sequential access pattern is only activated
+%% during recovery
 -type access_pattern() :: sequential | random.
 %% holds static or rarely changing fields
 -record(cfg, {uid :: ra_uid(),
@@ -58,7 +65,7 @@
                  segment_refs :: ra_lol:state(),
                  open_segments :: ra_flru:state(),
                  compaction :: undefined | major | minor,
-                 next_compaction :: undefined | major | minor
+                 minor_compaction_count = 0 :: non_neg_integer()
                 }).
 
 -record(compaction_result,
@@ -74,7 +81,8 @@
 -export_type([
               state/0,
               read_plan/0,
-              read_plan_options/0
+              read_plan_options/0,
+              major_compaction_strategy/0
              ]).
 
 %% PUBLIC
@@ -85,7 +93,8 @@
            map(),
            unicode:chardata()) -> state().
 init(UId, Dir, MaxOpen, AccessPattern, SegRefs0, Counter, CompConf, LogId)
-  when is_binary(UId) ->
+  when is_binary(UId) andalso
+       is_map(CompConf) ->
     Cfg = #cfg{uid = UId,
                log_id = LogId,
                counter = Counter,
@@ -173,27 +182,42 @@ update_segments(NewSegmentRefs, #?STATE{open_segments = Open0,
 -spec schedule_compaction(minor | major, ra:index(),
                           ra_seq:state(), state()) ->
     {state(), [ra_server:effect()]}.
+schedule_compaction(minor, SnapIdx, LiveIndexes,
+                    #?MODULE{cfg =
+                             #cfg{compaction_conf =
+                                  #{major_strategy :=
+                                    {num_minors, NumMinors}}},
+                             minor_compaction_count = MinorCount} = State)
+  when MinorCount >= NumMinors ->
+    %% promote to major compaction
+    schedule_compaction(major, SnapIdx, LiveIndexes, State);
 schedule_compaction(Type, SnapIdx, LiveIndexes,
                     #?MODULE{cfg = #cfg{log_id = LogId,
                                         compaction_conf = CompConf,
                                         directory = Dir} = Cfg,
+                             minor_compaction_count = MinorCompCnt,
                              compaction = undefined} = State) ->
     case compactable_segrefs(SnapIdx, State) of
         [] ->
             {State, []};
         SegRefs when LiveIndexes == [] ->
+
             %% if LiveIndexes is [] we can just delete all compactable
             %% segment refs
             Unreferenced = [F || {F, _} <- SegRefs],
+            ok = incr_counter(Cfg, ?C_RA_LOG_COMPACTIONS_MINOR_COUNT, 1),
             Result = #compaction_result{unreferenced = Unreferenced},
-            {State#?MODULE{compaction = minor},
+            {State#?MODULE{compaction = minor,
+                           minor_compaction_count = MinorCompCnt + 1},
              [{next_event,
                {ra_log_event, {compaction_result, Result}}}]};
         SegRefs when Type == minor ->
-            %% TODO evaluate if minor compactions are fast enough to run
+            %% TODO: evaluate if minor compactions are fast enough to run
             %% in server process
+            ok = incr_counter(Cfg, ?C_RA_LOG_COMPACTIONS_MINOR_COUNT, 1),
             Result = minor_compaction(SegRefs, LiveIndexes),
-            {State#?MODULE{compaction = minor},
+            {State#?MODULE{compaction = minor,
+                           minor_compaction_count = MinorCompCnt + 1},
              [{next_event,
                {ra_log_event, {compaction_result, Result}}}]};
         SegRefs ->
@@ -213,7 +237,8 @@ schedule_compaction(Type, SnapIdx, LiveIndexes,
                           ok
                   end,
 
-            {State#?MODULE{compaction = major},
+            {State#?MODULE{compaction = major,
+                           minor_compaction_count = 0},
              [{bg_work, Fun,
                fun (Err) ->
                        %% send an empty compaction result to ensure the
@@ -385,9 +410,11 @@ fetch_term(Idx, #?STATE{cfg = #cfg{} = Cfg} = State0) ->
 
 -spec info(state()) -> map().
 info(#?STATE{cfg = #cfg{} = _Cfg,
+             minor_compaction_count = MinorCount,
              open_segments = Open} = State) ->
     #{max_size => ra_flru:max_size(Open),
-      num_segments => segment_ref_count(State)}.
+      num_segments => segment_ref_count(State),
+      minor_compactions_count => MinorCount}.
 
 -spec purge_symlinks(file:filename_all(),
                      OlderThanSec :: non_neg_integer()) -> ok.

test/ra_kv_SUITE.erl

@@ -69,7 +69,7 @@ snapshot_replication_interrupted(_Config) ->
                                      #{members => Members}),
     ra:transfer_leadership(KvId, KvId),
     Data = crypto:strong_rand_bytes(100_000),
-    %% write 10k entries of the same key
+    %% write 10k entries of different keys
     [{ok, #{}} = ra_kv:put(KvId, term_to_binary(I), Data, 5000)
      || I <- lists:seq(1, 10_000)],
     ?assertMatch({ok, #{machine := #{num_keys := 10_000}}, KvId},
@@ -119,7 +119,7 @@ snapshot_replication(_Config) ->
                                      #{members => Members}),
     ra:transfer_leadership(KvId, KvId),
     {ok, #{}} = ra_kv:put(KvId, <<"k1">>, <<"k1-value01">>, 5000),
-    %% write 10k entries of the same key
+    %% write 10k entries with different keys
     [{ok, #{}} = ra_kv:put(KvId, integer_to_binary(I), I, 5000)
      || I <- lists:seq(1, 5000)],
 

test/ra_log_2_SUITE.erl

@@ -933,10 +933,6 @@ recovery(Config) ->
     {20, 3} = ra_log:last_index_term(Log5),
     Log6 = validate_fold(1, 4, 1, Log5),
     Log7 = validate_fold(5, 14, 2, Log6),
-    % debugger:start(),
-    % int:i(ra_log),
-    % int:break(ra_log, 413),
-
     Log8 = validate_fold(15, 20, 3, Log7),
     ra_log:close(Log8),
 

test/ra_log_segments_SUITE.erl

@@ -9,6 +9,7 @@
 -compile(nowarn_export_all).
 -compile(export_all).
 
+-include("src/ra.hrl").
 -include_lib("common_test/include/ct.hrl").
 -include_lib("eunit/include/eunit.hrl").
 -include_lib("kernel/include/file.hrl").
@@ -35,7 +36,8 @@ all_tests() ->
      minor,
      overwrite,
      result_after_segments,
-     result_after_segments_overwrite
+     result_after_segments_overwrite,
+     major_strategy_num_minors
     ].
 
 groups() ->
@@ -49,6 +51,7 @@ init_per_testcase(TestCase, Config) ->
     ok = ra_lib:make_dir(Dir),
     CompConf = #{max_count => 128,
                  max_size => 128_000},
+    ra_counters:init(),
     [{uid, atom_to_binary(TestCase, utf8)},
      {comp_conf, CompConf},
      {test_case, TestCase},
@@ -609,13 +612,60 @@ major(Config) ->
               length(Files) == 5
       end}
     ],
-
     SegConf = #{max_count => 128},
     Segs0 = ra_log_segments_init(Config, Dir, seg_refs(Dir)),
     run_scenario([{seg_conf, SegConf} | Config], Segs0, Scen),
 
     ok.
 
+major_strategy_num_minors(Config) ->
+    SegConf = #{max_count => 16},
+    Dir = ?config(dir, Config),
+    Entries1 = lists:seq(1, 32),
+    Entries2 = lists:seq(33, 64),
+    Entries3 = lists:seq(65, 96),
+    Live1 = ra_seq:from_list(lists:seq(1, 32, 3)),
+    Live2 = ra_seq:from_list(lists:seq(1, 64, 3)),
+    Live3 = ra_seq:from_list(lists:seq(1, 96, 3)),
+    ct:pal("Live1 ~p", [Live1]),
+    Scen =
+    [
+     {entries, 1, Entries1},
+     {assert, 1, lists:seq(1, 32)},
+     {assert, fun (S) ->
+                      SegRefs = ra_log_segments:segment_refs(S),
+                      length(SegRefs) == 2
+              end},
+     {minor, 30, Live1},
+     handle_compaction_result,
+     {assert, 1, lists:seq(1, 32, 3)},
+
+     {entries, 1, Entries2},
+     {minor, 60, Live2},
+     handle_compaction_result,
+     {entries, 1, Entries3},
+     {minor, 90, Live3},
+     handle_compaction_result
+
+    ],
+
+    Counters = ra_counters:new(?FUNCTION_NAME, ?RA_COUNTER_FIELDS),
+    CompConf  = #{max_count => 16,
+                  max_size => 128_000,
+                  major_strategy => {num_minors, 2}},
+
+    Segs0 = ra_log_segments_init(Config, Dir, seg_refs(Dir), CompConf, Counters),
+    _Segs = run_scenario([{seg_conf, SegConf} | Config], Segs0, Scen),
+    ?assertMatch(#{major_compactions := 1,
+                   minor_compactions := 2},
+                 ra_counters:counters(?FUNCTION_NAME,
+                                            [minor_compactions,
+                                             major_compactions])),
+
+    ok.
+
+
+
 %% Helpers
 
 open_last_segment(Config, SegConf) ->
@@ -766,10 +816,13 @@ with_ext(Fn, Ext) when is_binary(Fn) andalso is_list(Ext) ->
     <<(filename:rootname(Fn))/binary, (ra_lib:to_binary(Ext))/binary>>.
 
 ra_log_segments_init(Config, Dir, SegRefs) ->
-    UId = ?config(uid, Config),
     CompConf = ?config(comp_conf, Config),
+    ra_log_segments_init(Config, Dir, SegRefs, CompConf, undefined).
+
+ra_log_segments_init(Config, Dir, SegRefs, CompConf, Counters) ->
+    UId = ?config(uid, Config),
     ra_log_segments:init(UId, Dir, 1, random,
-                         SegRefs, undefined,
+                         SegRefs, Counters,
                          CompConf, "").
 
 do_compaction(Dir, CompactingFn, Live, All) ->