first cut phase 1 compaction impl

Author: kjnilsson

src/ra_log.erl

@@ -216,7 +216,6 @@ init(#{uid := UId,
     %% to get the actual valua
     {ok, LiveIndexes} = ra_snapshot:indexes(
                           ra_snapshot:current_snapshot_dir(SnapshotState)),
-    ct:pal("log init live indexes ~p", [LiveIndexes]),
 
     AccessPattern = maps:get(initial_access_pattern, Conf, sequential),
     {ok, Mt0} = ra_log_ets:mem_table_please(Names, UId),

src/ra_log_segment_writer.erl

@@ -273,7 +273,7 @@ flush_mem_table_ranges({ServerUId, TidSeqs0},
     TidSeqs = lists:foldl(
                 fun ({T, Seq0}, []) ->
                         case ra_seq:floor(SmallestIdx, Seq0) of
-                            undefined ->
+                            [] ->
                                 [];
                             Seq ->
                                 [{T, Seq}]
@@ -282,7 +282,7 @@ flush_mem_table_ranges({ServerUId, TidSeqs0},
                         Start = ra_seq:first(PrevSeq),
                         Seq1 = ra_seq:floor(SmallestIdx, Seq0),
                         case ra_seq:limit(Start, Seq1) of
-                            undefined ->
+                            [] ->
                                 Acc;
                             Seq ->
                                 [{T, Seq} | Acc]

src/ra_log_wal.erl

@@ -381,26 +381,29 @@ recover_wal(Dir, #conf{system = System,
              end || File <- Files0,
                     filename:extension(File) == ".wal"],
     WalFiles = lists:sort(Files),
-    AllWriters =
-        [begin
-             ?DEBUG("WAL in ~ts: recovering ~ts, Mode ~s",
-                    [System, F, Mode]),
-             Fd = open_at_first_record(filename:join(Dir, F)),
-             {Time, #recovery{ranges = Ranges,
-                              writers = Writers}} =
-             timer:tc(fun () -> recover_wal_chunks(Conf, Fd, Mode) end),
-
-             ok = ra_log_segment_writer:accept_mem_tables(SegWriter, Ranges, F),
-
-             close_existing(Fd),
-             ?DEBUG("WAL in ~ts: recovered ~ts time taken ~bms - recovered ~b writers",
-                    [System, F, Time div 1000, map_size(Writers)]),
-             Writers
-         end || F <- WalFiles],
-
-    FinalWriters = lists:foldl(fun (New, Acc) ->
-                                       maps:merge(Acc, New)
-                               end, #{}, AllWriters),
+    FinalWriters =
+    lists:foldl(fun (F, Writers0) ->
+                        ?DEBUG("WAL in ~ts: recovering ~ts, Mode ~s",
+                               [System, F, Mode]),
+                        Fd = open_at_first_record(filename:join(Dir, F)),
+                        {Time, #recovery{ranges = Ranges,
+                                         writers = Writers}} =
+                            timer:tc(fun () ->
+                                             recover_wal_chunks(Conf, Fd,
+                                                                Writers0, Mode)
+                                     end),
+
+                        ok = ra_log_segment_writer:accept_mem_tables(SegWriter,
+                                                                     Ranges, F),
+                        close_existing(Fd),
+                        ?DEBUG("WAL in ~ts: recovered ~ts time taken ~bms - recovered ~b writers",
+                               [System, F, Time div 1000, map_size(Writers)]),
+                        Writers
+                end, #{}, WalFiles),
+
+    % FinalWriters = lists:foldl(fun (New, Acc) ->
+    %                                    maps:merge(Acc, New)
+    %                            end, #{}, AllWriters),
 
     ?DEBUG("WAL in ~ts: final writers recovered ~b",
            [System, map_size(FinalWriters)]),
@@ -781,9 +784,10 @@ dump_records(<<_:1/unsigned, 1:1/unsigned, _:22/unsigned,
 dump_records(<<>>, Entries) ->
     Entries.
 
-recover_wal_chunks(#conf{} = Conf, Fd, Mode) ->
+recover_wal_chunks(#conf{} = Conf, Fd, Writers, Mode) ->
     Chunk = read_wal_chunk(Fd, Conf#conf.recovery_chunk_size),
-    recover_records(Conf, Fd, Chunk, #{}, #recovery{mode = Mode}).
+    recover_records(Conf, Fd, Chunk, #{}, #recovery{mode = Mode,
+                                                    writers = Writers}).
 % All zeros indicates end of a pre-allocated wal file
 recover_records(_, _Fd, <<0:1/unsigned, 0:1/unsigned, 0:22/unsigned,
                           IdDataLen:16/unsigned, _:IdDataLen/binary,
@@ -824,10 +828,11 @@ recover_records(#conf{names = Names} = Conf, Fd,
                     %               W ->
                     %                   W#{UId => {in_seq, SmallestIdx}}
                     %           end,
-                    W = State0#recovery.writers,
-                    Writers =  W#{UId => {in_seq, SmallestIdx - 1}},
+                    Writers = State0#recovery.writers,
+                    % Writers =  W#{UId => {in_seq, SmallestIdx - 1}},
                     recover_records(Conf, Fd, Rest, Cache,
-                                    State0#recovery{writers = Writers});
+                                    State0#recovery{writers =
+                                                        maps:remove(UId, Writers)});
                 error ->
                     System = Conf#conf.system,
                     ?DEBUG("WAL in ~ts: record failed CRC check. If this is the last record"
@@ -1004,7 +1009,16 @@ recover_entry(Names, UId, {Idx, _, _} = Entry, SmallestIdx,
                   {ok, M} = ra_log_ets:mem_table_please(Names, UId),
                   M
           end,
-    case ra_mt:insert(Entry, Mt0) of
+    %% always use write_sparse as there is nothing to indicate in the wal
+    %% data if an entry was written as such. this way we recover all writes
+    %% so should be ok for all types of writes
+    PrevIdx = case Writers of
+                  #{UId := {in_seq, I}} ->
+                      I;
+                  _ ->
+                      undefined
+              end,
+    case ra_mt:insert_sparse(Entry, PrevIdx, Mt0) of
         {ok, Mt1} ->
             Ranges = update_ranges(Ranges0, UId, ra_mt:tid(Mt1),
                                    SmallestIdx, [Idx]),
@@ -1014,7 +1028,8 @@ recover_entry(Names, UId, {Idx, _, _} = Entry, SmallestIdx,
         {error, overwriting} ->
             %% create successor memtable
             {ok, Mt1} = ra_log_ets:new_mem_table_please(Names, UId, Mt0),
-            {retry, State#recovery{tables = Tables#{UId => Mt1}}}
+            {retry, State#recovery{tables = Tables#{UId => Mt1},
+                                   writers = maps:remove(UId, Writers)}}
     end;
 recover_entry(Names, UId, {Idx, Term, _}, SmallestIdx,
               #recovery{mode = post_boot,
@@ -1049,6 +1064,7 @@ handle_trunc(false, _UId, _Idx, State) ->
     State;
 handle_trunc(true, UId, Idx, #recovery{mode = Mode,
                                        ranges = Ranges0,
+                                       writers = Writers,
                                        tables = Tbls} = State) ->
     case Tbls of
         #{UId := Mt0} when Mode == initial ->
@@ -1065,9 +1081,10 @@ handle_trunc(true, UId, Idx, #recovery{mode = Mode,
                      end,
 
             State#recovery{tables = Tbls#{UId => Mt},
+                           writers = maps:remove(UId, Writers),
                            ranges = Ranges};
         _ ->
-            State
+            State#recovery{writers = maps:remove(UId, Writers)}
     end.
 
 named_cast(To, Msg) when is_pid(To) ->

src/ra_mt.erl

@@ -120,11 +120,13 @@ insert({Idx, _, _} = _Entry,
     end.
 
 -spec insert_sparse(log_entry(), undefined | ra:index(), state()) ->
-    {ok, state()} | {error, gap_detected | limit_reached}.
+    {ok, state()} | {error, overwriting | gap_detected | limit_reached}.
 insert_sparse({Idx, _, _} = Entry, LastIdx,
-       #?MODULE{tid = Tid,
-                indexes = Seq} = State) ->
-    case ra_seq:last(Seq) == LastIdx of
+              #?MODULE{tid = Tid,
+                       indexes = Seq} = State) ->
+    LastSeq = ra_seq:last(Seq),
+    IsOverwriting = Idx =< LastSeq andalso is_integer(LastSeq),
+    case LastSeq == LastIdx andalso not IsOverwriting of
         true ->
             case ra_seq:length(Seq) > ?MAX_MEMTBL_ENTRIES of
                 true ->
@@ -134,7 +136,12 @@ insert_sparse({Idx, _, _} = Entry, LastIdx,
                     {ok, State#?MODULE{indexes = ra_seq:append(Idx, Seq)}}
             end;
         false ->
-            {error, gap_detected}
+            case IsOverwriting of
+                true ->
+                    {error, overwriting};
+                false ->
+                    {error, gap_detected}
+            end
     end.
 
 -spec stage(log_entry(), state()) ->

src/ra_server.erl

@@ -1532,7 +1532,6 @@ handle_receive_snapshot(#install_snapshot_rpc{term = Term,
     ?DEBUG("~ts: receiving snapshot chunk: ~b / ~w, index ~b, term ~b",
            [LogId, Num, ChunkFlag, SnapIndex, SnapTerm]),
     SnapState0 = ra_log:snapshot_state(Log0),
-    ct:pal("SnapStat0 ~p", [SnapState0]),
     {ok, SnapState, Effs0} = ra_snapshot:accept_chunk(Data, Num, ChunkFlag,
                                                       SnapState0),
     Reply = #install_snapshot_result{term = CurTerm,

test/ra_log_memory.erl

@@ -41,6 +41,8 @@
 
 -include("src/ra.hrl").
 
+-dialyzer({nowarn_function, install_snapshot/4}).
+
 -type ra_log_memory_meta() :: #{atom() => term()}.
 
 -record(state, {last_index = 0 :: ra_index(),
@@ -50,7 +52,7 @@
                 meta = #{} :: ra_log_memory_meta(),
                 snapshot :: option({ra_snapshot:meta(), term()})}).
 
--type ra_log_memory_state() :: #state{} | ra_log:state().
+-opaque ra_log_memory_state() :: #state{} | ra_log:state().
 
 -export_type([ra_log_memory_state/0]).
 
@@ -170,17 +172,18 @@ last_written(#state{last_written = LastWritten}) ->
 
 -spec handle_event(ra_log:event_body(), ra_log_memory_state()) ->
     {ra_log_memory_state(), list()}.
-handle_event({written, Term, {_From, Idx} = Range0}, State0) ->
+handle_event({written, Term, Seq0}, State0) when is_list(Seq0) ->
+    Idx = ra_seq:last(Seq0),
     case fetch_term(Idx, State0) of
         {Term, State} ->
             {State#state{last_written = {Idx, Term}}, []};
         _ ->
-            case ra_range:limit(Idx, Range0) of
-                undefined ->
+            case ra_seq:limit(Idx - 1, Seq0) of
+                [] ->
                     % if the term doesn't match we just ignore it
                     {State0, []};
-                Range ->
-                    handle_event({written, Term, Range}, State0)
+                Seq ->
+                    handle_event({written, Term, Seq}, State0)
             end
     end;
 handle_event(_Evt, State0) ->
@@ -210,6 +213,8 @@ fetch_term(Idx, #state{entries = Log} = State) ->
 
 flush(_Idx, Log) -> Log.
 
+-spec install_snapshot(ra_idxterm(), term(), module(), ra_log_memory_state()) ->
+    {ra_snapshot:meta(), term(), ra_log_memory_state(), list()}.
 install_snapshot({Index, Term}, Data, _MacMod,
                  #state{entries = Log0} = State0)
   when is_tuple(Data) ->

test/ra_log_meta_SUITE.erl

@@ -38,6 +38,7 @@ init_per_group(_, Config) ->
     Config.
 
 end_per_group(_, Config) ->
+    application:stop(ra),
     Config.
 
 init_per_testcase(TestCase, Config) ->