first cut phase 1 compaction impl

Author: kjnilsson

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,17 @@ 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,
+    % ct:pal("ra_mt:insert_sparse ~b ~w", [Idx, PrevIdx]),
+    case ra_mt:insert_sparse(Entry, PrevIdx, Mt0) of
         {ok, Mt1} ->
             Ranges = update_ranges(Ranges0, UId, ra_mt:tid(Mt1),
                                    SmallestIdx, [Idx]),
@@ -1014,7 +1029,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 +1065,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 +1082,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

@@ -122,9 +122,11 @@ insert({Idx, _, _} = _Entry,
 -spec insert_sparse(log_entry(), undefined | ra:index(), state()) ->
     {ok, state()} | {error, 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, Idx, LastSeq}
+            end
     end.
 
 -spec stage(log_entry(), state()) ->

test/ra_log_wal_SUITE.erl

@@ -28,6 +28,7 @@ all_tests() ->
      sparse_write_same_batch,
      sparse_write_overwrite,
      sparse_write_recover,
+     sparse_write_recover_with_mt,
      wal_filename_upgrade,
      same_uid_different_process,
      consecutive_terms_in_batch_should_result_in_two_written_events,
@@ -207,19 +208,57 @@ sparse_write_same_batch(Config) ->
     ok.
 
 sparse_write_recover(Config) ->
+    %% no mt case
     meck:new(ra_log_segment_writer, [passthrough]),
     meck:expect(ra_log_segment_writer, await, fun(_) -> ok end),
     Conf = ?config(wal_conf, Config),
     {UId, _} = WriterId = ?config(writer_id, Config),
-    Tid = ets:new(?FUNCTION_NAME, []),
+    Names = ?config(names, Config),
+    %% create a tid that isn't registered as mt
+    Tid = ets:new(?MODULE, [set]),
     {ok, Pid} = ra_log_wal:start_link(Conf),
     {ok, _} = ra_log_wal:write(Pid, WriterId, Tid, 11, 12, 1, "value"),
     ok = await_written(WriterId, 1, [12]),
     {ok, _} = ra_log_wal:write(Pid, WriterId, Tid, 12, 15, 1, "value2"),
     ok = await_written(WriterId, 1, [15]),
+    ?assert(is_process_alive(Pid)),
+    ok = proc_lib:stop(Pid),
+    {ok, Pid2} = ra_log_wal:start_link(Conf),
+    ?assert(is_process_alive(Pid2)),
+    receive
+        {'$gen_cast',
+         {mem_tables, #{UId := [{MtTid, [15, 12]}]}, _}} ->
+            {ok, Mt0} = ra_log_ets:mem_table_please(Names, UId),
+            ?assertEqual(MtTid, ra_mt:tid(Mt0)),
+            ok
+    after 5000 ->
+              flush(),
+              ct:fail("receiving mem table ranges timed out")
+    end,
+    flush(),
+    proc_lib:stop(Pid2),
+    meck:unload(),
+    ok.
 
-    ok = proc_lib:stop(ra_log_wal),
-    {ok, _Pid2} = ra_log_wal:start_link(Conf),
+sparse_write_recover_with_mt(Config) ->
+    meck:new(ra_log_segment_writer, [passthrough]),
+    meck:expect(ra_log_segment_writer, await, fun(_) -> ok end),
+    Conf = ?config(wal_conf, Config),
+    {UId, _} = WriterId = ?config(writer_id, Config),
+    Names = ?config(names, Config),
+    {ok, Mt0} = ra_log_ets:mem_table_please(Names, UId),
+    Tid = ra_mt:tid(Mt0),
+    {ok, Pid} = ra_log_wal:start_link(Conf),
+    {ok, Mt1} = ra_mt:insert_sparse({12, 1, "value"}, undefined, Mt0),
+    {ok, _} = ra_log_wal:write(Pid, WriterId, Tid, 11, 12, 1, "value"),
+    ok = await_written(WriterId, 1, [12]),
+    {ok, _Mt} = ra_mt:insert_sparse({15, 1, "value"}, 12, Mt1),
+    {ok, _} = ra_log_wal:write(Pid, WriterId, Tid, 12, 15, 1, "value2"),
+    ok = await_written(WriterId, 1, [15]),
+    ?assert(is_process_alive(Pid)),
+    ok = proc_lib:stop(Pid),
+    {ok, Pid2} = ra_log_wal:start_link(Conf),
+    ?assert(is_process_alive(Pid2)),
     receive
         {'$gen_cast',
          {mem_tables, #{UId := [{Tid, [15, 12]}]}, _}} ->
@@ -228,7 +267,8 @@ sparse_write_recover(Config) ->
               flush(),
               ct:fail("receiving mem table ranges timed out")
     end,
-    proc_lib:stop(Pid),
+    flush(),
+    proc_lib:stop(Pid2),
     meck:unload(),
     ok.
 
@@ -976,7 +1016,7 @@ recover_overwrite(Config) ->
     _ = await_written(WriterId, 2, [{5, 20}]),
 
     flush(),
-    ok = proc_lib:stop(ra_log_wal),
+    ok = proc_lib:stop(ra_log_wal, normal, 5000),
     {ok, Pid2} = ra_log_wal:start_link(Conf),
     {ok, Mt} = ra_log_ets:mem_table_please(?config(names, Config), UId),