WIP: WAL sparse writes

Author: kjnilsson

docs/internals/COMPACTION.md

@@ -243,7 +243,7 @@ the live index replication. NB the snapshot sender process may need to call
 into the leader process to get read plans as entries _could_ be in the memtable.
 
 #### How to work out which live indexes the follower needs
-
+WA
 Gnarly example:
 
 Follower term indexes:
@@ -287,4 +287,6 @@ for decision making.
 
 WAL needs to accept sparse writes without a higher snapshot idx (snap install)
 WAL needs to accept contiguous writes with a higher snap idx with and without live indexes
-
+WAL will send ra_seq of entries written in a WAL 
+SegWriter needs to flush the live indexes preceeding the snapshot index which
+_should_ be covered in the sparse sequence of written indexes. 

src/ra_log.erl

@@ -336,7 +336,7 @@ commit_tx(#?MODULE{cfg = #cfg{uid = UId,
     {WalCommands, Num} =
         lists:foldl(fun ({Idx, Term, Cmd0}, {WC, N}) ->
                             Cmd = {ttb, term_to_iovec(Cmd0)},
-                            WalC = {append, WriterId, Tid, Idx, Term, Cmd},
+                            WalC = {append, WriterId, Tid, Idx-1, Idx, Term, Cmd},
                             {[WalC | WC], N+1}
                     end, {[], 0}, Entries),
 
@@ -1176,11 +1176,11 @@ wal_write_batch(#?MODULE{cfg = #cfg{uid = UId,
     {WalCommands, Num} =
         lists:foldl(fun ({Idx, Term, Cmd0}, {WC, N}) ->
                             Cmd = {ttb, term_to_iovec(Cmd0)},
-                            WalC = {append, WriterId, Tid, Idx, Term, Cmd},
+                            WalC = {append, WriterId, Tid, Idx-1, Idx, Term, Cmd},
                             {[WalC | WC], N+1}
                     end, {[], 0}, Entries),
 
-    [{_, _, _, LastIdx, LastTerm, _} | _] = WalCommands,
+    [{_, _, _, _PrevIdx, LastIdx, LastTerm, _} | _] = WalCommands,
     {_, Mt} = ra_mt:commit(Mt0),
     put_counter(Cfg, ?C_RA_SVR_METRIC_LAST_INDEX, LastIdx),
     ok = incr_counter(Cfg, ?C_RA_LOG_WRITE_OPS, Num),

src/ra_log_segment_writer.erl

@@ -66,14 +66,13 @@ start_link(#{name := Name} = Config) ->
 
 -spec accept_mem_tables(atom() | pid(),
                         #{ra_uid() => [{ets:tid(), ra:range()}]},
-                        string()) ->
-    ok.
+                        string()) -> ok.
 accept_mem_tables(_SegmentWriter, Tables, undefined)
   when map_size(Tables) == 0 ->
     ok;
-accept_mem_tables(SegmentWriter, Tables, WalFile)
-  when is_map(Tables) ->
-    gen_server:cast(SegmentWriter, {mem_tables, Tables, WalFile}).
+accept_mem_tables(SegmentWriter, UIdTidRanges, WalFile)
+  when is_map(UIdTidRanges) ->
+    gen_server:cast(SegmentWriter, {mem_tables, UIdTidRanges, WalFile}).
 
 -spec truncate_segments(atom() | pid(), ra_uid(), ra_log:segment_ref()) -> ok.
 truncate_segments(SegWriter, Who, SegRef) ->
@@ -135,9 +134,10 @@ segments_for(UId, #state{data_dir = DataDir}) ->
     Dir = filename:join(DataDir, ra_lib:to_list(UId)),
     segment_files(Dir).
 
-handle_cast({mem_tables, Ranges, WalFile}, #state{data_dir = Dir,
-                                                  system = System} = State) ->
-    ok = counters:add(State#state.counter, ?C_MEM_TABLES, map_size(Ranges)),
+handle_cast({mem_tables, UIdTidRanges, WalFile},
+            #state{data_dir = Dir,
+                   system = System} = State) ->
+    ok = counters:add(State#state.counter, ?C_MEM_TABLES, map_size(UIdTidRanges)),
     #{names := Names} = ra_system:fetch(System),
     Degree = erlang:system_info(schedulers),
     %% TODO: refactor to make better use of time where each uid has an
@@ -156,7 +156,7 @@ handle_cast({mem_tables, Ranges, WalFile}, #state{data_dir = Dir,
                                    ok = ra_log_ets:delete_mem_tables(Names, UId),
                                    Acc
                            end
-                   end, [], Ranges),
+                   end, [], UIdTidRanges),
 
     T1 = erlang:monotonic_time(),
     _ = [begin

src/ra_log_wal.erl

@@ -17,6 +17,7 @@
 
 -export([
          write/6,
+         write/7,
          write_batch/2,
          last_writer_seq/2,
          force_roll_over/1]).
@@ -140,7 +141,8 @@
 -export_type([wal_conf/0]).
 
 -type wal_command() ::
-    {append, writer_id(), ra_index(), ra_term(), term()}.
+    {append, writer_id(), PrevIndex :: ra:index() | -1,
+     Index :: ra:index(), Term :: ra_term(), wal_cmd()}.
 
 -type wal_op() :: {cast, wal_command()} |
                   {call, from(), wal_command()}.
@@ -149,10 +151,23 @@
 -spec write(atom() | pid(), writer_id(), ets:tid(), ra_index(), ra_term(),
             wal_cmd()) ->
     {ok, pid()} | {error, wal_down}.
-write(Wal, {_, _} = From, MtTid, Idx, Term, Cmd)
+write(Wal, From, MtTid, Idx, Term, Cmd) ->
+    %% "normal opereation where we assume a contigious sequence
+    %% this may be removed at some point
+    write(Wal, From, MtTid, Idx-1, Idx, Term, Cmd).
+
+-spec write(atom() | pid(), writer_id(), ets:tid(),
+            PrevIndex :: ra:index() | -1,
+            Index :: ra_index(),
+            Term :: ra_term(),
+            wal_cmd()) ->
+    {ok, pid()} | {error, wal_down}.
+write(Wal, {_, _} = From, MtTid, PrevIdx, Idx, Term, Cmd)
   when is_integer(Idx) andalso
-       is_integer(Term) ->
-    named_cast(Wal, {append, From, MtTid, Idx, Term, Cmd}).
+       is_integer(PrevIdx) andalso
+       is_integer(Term) andalso
+       PrevIdx < Idx ->
+    named_cast(Wal, {append, From, MtTid, PrevIdx, Idx, Term, Cmd}).
 
 -spec write_batch(Wal :: atom() | pid(), [wal_command()]) ->
     {ok, pid()} | {error, wal_down}.
@@ -476,7 +491,7 @@ write_data({UId, Pid} = Id, MtTid, Idx, Term, Data0, Trunc, SmallestIndex,
     end.
 
 
-handle_msg({append, {UId, Pid} = Id, MtTid, Idx, Term, Entry},
+handle_msg({append, {UId, Pid} = Id, MtTid, PrevIdx0, Idx, Term, Entry},
            #state{conf = Conf,
                   writers = Writers} = State0) ->
     SmallestIdx = smallest_live_index(Conf, UId),
@@ -487,13 +502,17 @@ handle_msg({append, {UId, Pid} = Id, MtTid, Idx, Term, Entry},
         _ when Idx < SmallestIdx ->
             %% the smallest live index for the last snapshot is higher than
             %% this index, just drop it
-            PrevIdx = SmallestIdx - 1,
-            State0#state{writers = Writers#{UId => {in_seq, PrevIdx}}};
+            LastIdx = SmallestIdx - 1,
+            State0#state{writers = Writers#{UId => {in_seq, LastIdx}}};
         {ok, {_, PrevIdx}}
-          when Idx =< PrevIdx + 1 orelse
+          when PrevIdx0 =< PrevIdx orelse
                Trunc ->
+            %% if the passed in previous index is less than the last written
+            %% index (gap detection) _or_ it is a truncation
+            %% then we can proceed and write the entry
             write_data(Id, MtTid, Idx, Term, Entry, Trunc, SmallestIdx, State0);
         error ->
+            %% no state for the UId is known so go ahead and write
             write_data(Id, MtTid, Idx, Term, Entry, false, SmallestIdx, State0);
         {ok, {out_of_seq, _}} ->
             % writer is out of seq simply ignore drop the write
@@ -528,6 +547,8 @@ incr_batch(#batch{num_writes = Writes,
                       %% The Tid and term is the same so add to current batch_writer
                       Range = ra_range:extend(Idx, ra_range:truncate(SmallestIdx - 1,
                                                                      Range0)),
+                      %% TODO: range nees to become a ra_seq so that we can
+                      %% capture sparse writes correctly
                       Waiting0#{Pid => W#batch_writer{range = Range,
                                                       smallest_live_idx = SmallestIdx,
                                                       term = Term

test/ra_log_wal_SUITE.erl

@@ -24,6 +24,7 @@ all() ->
 all_tests() ->
     [
      basic_log_writes,
+     sparse_writes,
      wal_filename_upgrade,
      same_uid_different_process,
      consecutive_terms_in_batch_should_result_in_two_written_events,
@@ -61,7 +62,8 @@ groups() ->
     [
      {default, [], all_tests()},
      %% uses fsync instead of the default fdatasync
-     {fsync, [], all_tests()},
+     %% just testing that the configuration and dispatch works
+     {fsync, [], [basic_log_writes]},
      {no_sync, [], all_tests()}
     ].
 
@@ -148,6 +150,44 @@ basic_log_writes(Config) ->
     meck:unload(),
     ok.
 
+sparse_writes(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),
+    Tid = ets:new(?FUNCTION_NAME, []),
+    {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, 12}),
+    debugger:start(),
+    int:i(ra_log_wal),
+    int:break(ra_log_wal, 975),
+    timer:sleep(1000),
+    {ok, _} = ra_log_wal:write(Pid, WriterId, Tid, 12, 15, 1, "value2"),
+    timer:sleep(200000),
+    ok = await_written(WriterId, 1, {15, 15}),
+    ra_log_wal:force_roll_over(Pid),
+    receive
+        {'$gen_cast',
+         {mem_tables, #{UId := [{Tid, {12, 15}}]}, _}} ->
+            ok
+    after 5000 ->
+              flush(),
+              ct:fail("receiving mem table ranges timed out")
+    end,
+    proc_lib:stop(Pid),
+    meck:unload(),
+    ok.
+
+sparse_write_same_batch(_Config) ->
+    ct:fail("~s", [?FUNCTION_NAME]).
+
+sparse_write_recover(_Config) ->
+    ct:fail("~s", [?FUNCTION_NAME]).
+
+sparse_write_overwrite(_Config) ->
+    ct:fail("~s", [?FUNCTION_NAME]).
+
 wal_filename_upgrade(Config) ->
     meck:new(ra_log_segment_writer, [passthrough]),
     meck:expect(ra_log_segment_writer, await, fun(_) -> ok end),