end of day wip

Author: kjnilsson

src/ra_log.erl

@@ -62,7 +62,7 @@
 
 -type ra_meta_key() :: atom().
 -type segment_ref() :: {ra_range:range(), File :: file:filename_all()}.
--type event_body() :: {written, ra_term(), ra:range()} |
+-type event_body() :: {written, ra_term(), ra_seq:state()} |
                       {segments, [{ets:tid(), ra:range()}], [segment_ref()]} |
                       {resend_write, ra_index()} |
                       {snapshot_written, ra_idxterm(), ra_snapshot:kind()} |
@@ -610,14 +610,14 @@ set_last_index(Idx, #?MODULE{cfg = Cfg,
 
 -spec handle_event(event_body(), state()) ->
     {state(), [effect()]}.
-handle_event({written, _Term, {FromIdx, _ToIdx}},
+handle_event({written, _Term, [{FromIdx, _ToIdx}]},
              #?MODULE{last_index = LastIdx} = State)
   when FromIdx > LastIdx ->
     %% we must have reverted back, either by explicit reset or by a snapshot
     %% installation taking place whilst the WAL was processing the write
     %% Just drop the event in this case as it is stale
     {State, []};
-handle_event({written, Term, {FromIdx, ToIdx}},
+handle_event({written, Term, [{FromIdx, ToIdx}]},
              #?MODULE{cfg = Cfg,
                       last_written_index_term = {LastWrittenIdx0,
                                                  _LastWrittenTerm0},
@@ -648,10 +648,10 @@ handle_event({written, Term, {FromIdx, ToIdx}},
                     {State, []};
                 NextWrittenRange ->
                     %% retry with a reduced range
-                    handle_event({written, Term, NextWrittenRange}, State0)
+                    handle_event({written, Term, [NextWrittenRange]}, State0)
             end
     end;
-handle_event({written, _Term, {FromIdx, _}} = Evt,
+handle_event({written, _Term, [{FromIdx, _}]} = Evt,
              #?MODULE{cfg = #cfg{log_id = LogId},
                       mem_table = Mt,
                       last_written_index_term = {LastWrittenIdx, _}} = State0)
@@ -675,6 +675,17 @@ handle_event({written, _Term, {FromIdx, _}} = Evt,
             handle_event(Evt,
                          State#?MODULE{last_written_index_term = {Expected, Term}})
     end;
+handle_event({written, Term, Written}, State) ->
+    %% simple handling of ra_seqs for now
+    case Written of
+        [I] when is_integer(I) ->
+            handle_event({written, Term, [{I, I}]}, State);
+        [I2, I] when is_integer(I) andalso
+                     I + 1 == I2 ->
+            handle_event({written, Term, [{I, I2}]}, State);
+        _ ->
+            exit({sparse_written_events_not_implemented, Written})
+    end;
 handle_event({segments, TidRanges, NewSegs},
              #?MODULE{cfg = #cfg{uid = UId, names = Names} = Cfg,
                       reader = Reader0,
@@ -1200,7 +1211,7 @@ wal_write_batch(#?MODULE{cfg = #cfg{uid = UId,
 maybe_append_first_entry(State0 = #?MODULE{last_index = -1}) ->
     State = append({0, 0, undefined}, State0),
     receive
-        {ra_log_event, {written, 0, {0, 0}}} ->
+        {ra_log_event, {written, 0, [0]}} ->
             ok
     after 60000 ->
               exit({?FUNCTION_NAME, timeout})
@@ -1323,14 +1334,15 @@ pick_range([{Fst, _Lst} | Tail], {CurFst, CurLst}) ->
 
 %% TODO: implement synchronous writes using gen_batch_server:call/3
 await_written_idx(Idx, Term, Log0) ->
-    IDX = Idx,
     receive
-        {ra_log_event, {written, Term, {_, IDX}} = Evt} ->
+        {ra_log_event, {written, Term, _Seq} = Evt} ->
             {Log, _} = handle_event(Evt, Log0),
-            Log;
-        {ra_log_event, {written, _, _} = Evt} ->
-            {Log, _} = handle_event(Evt, Log0),
-            await_written_idx(Idx, Term, Log)
+            case last_written(Log) of
+                {Idx, Term} ->
+                    Log;
+                _ ->
+                    await_written_idx(Idx, Term, Log)
+            end
     after ?LOG_APPEND_TIMEOUT ->
               throw(ra_log_append_timeout)
     end.

src/ra_log_segment_writer.erl

@@ -261,7 +261,7 @@ get_overview(#state{data_dir = Dir,
     #{data_dir => Dir,
       segment_conf => Conf}.
 
-flush_mem_table_ranges({ServerUId, TidRanges0},
+flush_mem_table_ranges({ServerUId, TidSeqs0},
                        #state{system = System} = State) ->
     SmallestIdx = smallest_live_idx(ServerUId),
     %% TidRanges arrive here sorted new -> old.
@@ -270,31 +270,32 @@ flush_mem_table_ranges({ServerUId, TidRanges0},
     %% list of tid ranges to flush to disk
     %% now TidRanges are sorted old -> new, i.e the correct order of
     %% processing
-    TidRanges = lists:foldl(
-                  fun ({T, Range0}, []) ->
-                          case ra_range:truncate(SmallestIdx - 1, Range0) of
-                              undefined ->
-                                  [];
-                              Range ->
-                                  [{T, Range}]
-                          end;
-                      ({T, Range0}, [{_T, {Start, _}} | _] = Acc) ->
-                          Range1 = ra_range:truncate(SmallestIdx - 1, Range0),
-                          case ra_range:limit(Start, Range1) of
-                              undefined ->
-                                  Acc;
-                              Range ->
-                                  [{T, Range} | Acc]
-                          end
-                  end, [], TidRanges0),
+    TidSeqs = lists:foldl(
+                fun ({T, Seq0}, []) ->
+                        case ra_seq:floor(SmallestIdx, Seq0) of
+                            undefined ->
+                                [];
+                            Seq ->
+                                [{T, Seq}]
+                        end;
+                    ({T, Seq0}, [{_T, PrevSeq} | _] = Acc) ->
+                        Start = ra_seq:first(PrevSeq),
+                        Seq1 = ra_seq:floor(SmallestIdx, Seq0),
+                        case ra_seq:limit(Start, Seq1) of
+                            undefined ->
+                                Acc;
+                            Seq ->
+                                [{T, Seq} | Acc]
+                        end
+                end, [], TidSeqs0),
 
     SegRefs0 = lists:append(
                  lists:reverse(
                    %% segrefs are returned in appended order so new -> old
                    %% so we need to reverse them so that the final appended list
                    %% of segrefs is in the old -> new order
-                   [flush_mem_table_range(ServerUId, TidRange, State)
-                    || TidRange <- TidRanges])),
+                   [flush_mem_table_range(ServerUId, TidSeq, State)
+                    || TidSeq <- TidSeqs])),
 
     %% compact cases where a segment was appended in a subsequent call to
     %% flush_mem_table_range
@@ -308,14 +309,14 @@ flush_mem_table_ranges({ServerUId, TidRanges0},
                           [Seg | Acc]
                   end, [], SegRefs0)),
 
-    ok = send_segments(System, ServerUId, TidRanges0, SegRefs),
+    ok = send_segments(System, ServerUId, TidSeqs0, SegRefs),
     ok.
 
-flush_mem_table_range(ServerUId, {Tid, {StartIdx0, EndIdx}},
+flush_mem_table_range(ServerUId, {Tid, {StartIdx, EndIdx}},
                       #state{data_dir = DataDir,
                              segment_conf = SegConf} = State) ->
     Dir = filename:join(DataDir, binary_to_list(ServerUId)),
-    StartIdx = start_index(ServerUId, StartIdx0),
+    % StartIdx = start_index(ServerUId, StartIdx0),
     case open_file(Dir, SegConf) of
         enoent ->
             ?DEBUG("segment_writer: skipping segment as directory ~ts does "

src/ra_log_wal.erl

@@ -99,7 +99,7 @@
 
 -record(recovery, {mode :: initial | post_boot,
                    ranges = #{} :: #{ra_uid() =>
-                                     [{ets:tid(), {ra:index(), ra:index()}}]},
+                                     [{ets:tid(), ra_seq:state()}]},
                    tables = #{} :: #{ra_uid() => ra_mt:state()},
                    writers = #{} :: #{ra_uid() => {in_seq, ra:index()}}
                   }).
@@ -584,22 +584,29 @@ roll_over(#state{wal = Wal0, file_num = Num0,
     %% if this is the first wal since restart randomise the first
     %% max wal size to reduce the likelihood that each erlang node will
     %% flush mem tables at the same time
-    NextMaxBytes = case Wal0 of
-                       undefined ->
-                           Half = MaxBytes div 2,
-                           Half + rand:uniform(Half);
-                       #wal{ranges = Ranges,
-                            filename = Filename} ->
-                           _ = file:advise(Wal0#wal.fd, 0, 0, dont_need),
-                           ok = close_file(Wal0#wal.fd),
-                           MemTables = Ranges,
-                           %% TODO: only keep base name in state
-                           Basename = filename:basename(Filename),
-                           ok = ra_log_segment_writer:accept_mem_tables(SegWriter,
-                                                                        MemTables,
-                                                                        Basename),
-                           MaxBytes
-                   end,
+    NextMaxBytes =
+        case Wal0 of
+            undefined ->
+                Half = MaxBytes div 2,
+                Half + rand:uniform(Half);
+            #wal{ranges = Ranges,
+                 filename = Filename} ->
+                _ = file:advise(Wal0#wal.fd, 0, 0, dont_need),
+                ok = close_file(Wal0#wal.fd),
+                %% floor all sequences
+                MemTables = maps:map(
+                              fun (UId, TidRanges) ->
+                                      SmallestIdx = smallest_live_index(Conf0, UId),
+                                      [{Tid, ra_seq:floor(SmallestIdx, Seq)}
+                                       || {Tid, Seq} <- TidRanges]
+                              end, Ranges),
+                %% TODO: only keep base name in state
+                Basename = filename:basename(Filename),
+                ok = ra_log_segment_writer:accept_mem_tables(SegWriter,
+                                                             MemTables,
+                                                             Basename),
+                MaxBytes
+        end,
     {Conf, Wal} = open_wal(NextFile, NextMaxBytes, Conf0),
     State0#state{conf = Conf,
                  wal = Wal,
@@ -695,11 +702,12 @@ complete_batch(#state{batch = #batch{waiting = Waiting,
 complete_batch_writer(Pid, #batch_writer{smallest_live_idx = SmallestIdx,
                                          tid = MtTid,
                                          uid = UId,
-                                         seq = Range,
+                                         seq = Seq0,
                                          term = Term,
                                          old = undefined}, Ranges) ->
-    Pid ! {ra_log_event, {written, Term, Range}},
-    update_ranges(Ranges, UId, MtTid, SmallestIdx, Range);
+    Seq = ra_seq:floor(SmallestIdx, Seq0),
+    Pid ! {ra_log_event, {written, Term, Seq}},
+    update_ranges(Ranges, UId, MtTid, SmallestIdx, Seq);
 complete_batch_writer(Pid, #batch_writer{old = #batch_writer{} = OldBw} = Bw,
                       Ranges0) ->
     Ranges = complete_batch_writer(Pid, OldBw, Ranges0),
@@ -968,22 +976,19 @@ should_roll_wal(#state{conf = #conf{max_entries = MaxEntries},
 smallest_live_index(#conf{ra_log_snapshot_state_tid = Tid}, ServerUId) ->
     ra_log_snapshot_state:smallest(Tid, ServerUId).
 
-update_ranges(Ranges, UId, MtTid, SmallestIdx, AddSeq) ->
+update_ranges(Ranges, UId, MtTid, _SmallestIdx, AddSeq) ->
     case Ranges of
-        #{UId := [{MtTid, Seq0} | Rem]} ->
+        #{UId := [{MtTid, Seq0} | Seqs]} ->
             %% SmallestIdx might have moved to we truncate the old range first
             %% before extending
-            Seq1 = ra_seq:floor(SmallestIdx, Seq0),
+            % Seq1 = ra_seq:floor(SmallestIdx, Seq0),
             %% limit the old range by the add end start as in some resend
             %% cases we may have got back before the prior range.
-            Seq = ra_seq:add(AddSeq, Seq1),
-            Ranges#{UId => [{MtTid, Seq} | Rem]};
-        #{UId := [{OldMtTid, OldMtSeq} | Rem]} ->
+            Seq = ra_seq:add(AddSeq, Seq0),
+            Ranges#{UId => [{MtTid, Seq} | Seqs]};
+        #{UId := Seqs} ->
             %% new Tid, need to add a new range record for this
-            Ranges#{UId => [{MtTid, AddSeq},
-                            {OldMtTid,
-                             ra_seq:floor(SmallestIdx, OldMtSeq)}
-                            | Rem]};
+            Ranges#{UId => [{MtTid, AddSeq} | Seqs]};
         _ ->
             Ranges#{UId => [{MtTid, AddSeq}]}
     end.
@@ -1043,13 +1048,24 @@ recover_entry(Names, UId, {Idx, Term, _}, SmallestIdx,
 handle_trunc(false, _UId, _Idx, State) ->
     State;
 handle_trunc(true, UId, Idx, #recovery{mode = Mode,
+                                       ranges = Ranges0,
                                        tables = Tbls} = State) ->
     case Tbls of
         #{UId := Mt0} when Mode == initial ->
             %% only meddle with mem table data in initial mode
             {Specs, Mt} = ra_mt:set_first(Idx-1, Mt0),
             [_ = ra_mt:delete(Spec) || Spec <- Specs],
-            State#recovery{tables = Tbls#{UId => Mt}};
+            Ranges = case Ranges0 of
+                         #{UId := Seqs0} ->
+                             Seqs = [{T, ra_seq:floor(Idx, Seq)}
+                                     || {T, Seq} <- Seqs0],
+                             Ranges0#{UId => Seqs};
+                         _ ->
+                             Ranges0
+                     end,
+
+            State#recovery{tables = Tbls#{UId => Mt},
+                           ranges = Ranges};
         _ ->
             State
     end.

src/ra_seq.erl

@@ -11,7 +11,11 @@
 %% [55, {20, 52}, 3]
 -type state() :: [ra:index() | ra:range()].
 
--export_type([state/0]).
+-record(i, {seq :: state()}).
+-opaque iter() :: #i{}.
+
+-export_type([state/0,
+             iter/0]).
 
 
 -export([
@@ -20,7 +24,12 @@
          floor/2,
          limit/2,
          add/2,
-         fold/3
+         fold/3,
+         expand/1,
+         subtract/2,
+         first/1,
+         iterator/1,
+         next/1
         ]).
 
 -spec append(ra:index(), state()) -> state().
@@ -94,6 +103,29 @@ fold(Fun, Acc0, Seq) ->
               Fun(Idx, Acc)
       end, Acc0, Seq).
 
+-spec expand(state()) -> [ra:index()].
+expand(Seq) ->
+    fold(fun (I, Acc) -> [I | Acc] end, [], Seq).
+
+subtract(SeqA, SeqB) ->
+    %% TODO: not efficient at all but good enough for now
+    %% optimise if we end up using this in critical path
+    A = expand(SeqA),
+    B = expand(SeqB),
+    from_list(A -- B).
+
+-spec first(state()) -> undefined | ra:index().
+first(Seq) ->
+    last_index(lists:reverse(Seq)).
+
+-spec iterator(state()) -> iter() | end_of_seq.
+iterator(Seq) when is_list(Seq) ->
+    #i{seq = lists:reverse(Seq)}.
+
+-spec next(iter()) -> {ra:index(), iter() | end_of_seq}.
+next(#i{}) ->
+    end_of_seq.
+
 %% internal functions
 
 floor0(FloorIdx, [Last | Rem], Acc)
@@ -114,13 +146,11 @@ floor0(FloorIdx, [{_, _} = T | Rem], Acc) ->
 floor0(_FloorIdx, _Seq, Acc) ->
     lists:reverse(Acc).
 
-% first_index(Seq) ->
-%     last_index(lists:reverse(Seq)).
 
-% last_index([{_, I} | _]) ->
-%     I;
-% last_index([I | _])
-%   when is_integer(I) ->
-%     I;
-% last_index([]) ->
-%     undefined.
+last_index([{_, I} | _]) ->
+    I;
+last_index([I | _])
+  when is_integer(I) ->
+    I;
+last_index([]) ->
+    undefined.

test/ra_log_SUITE.erl

@@ -115,10 +115,12 @@ append_then_fetch_no_wait(Config) ->
     % results in the last written being updated
     receive
         {ra_log_event, {written, _, _} = Evt} ->
+            ct:pal("written ~p", [Evt]),
             {Log, _} = ra_log:handle_event(Evt, Log3),
             {Idx, Term} = ra_log:last_written(Log)
-    after 0 ->
-              ok
+    after 1000 ->
+              flush(),
+              ct:pal("fail written event not received")
     end,
     ok.
 
@@ -277,3 +279,12 @@ append_in(Term, Data, Log0) ->
 ra_log_take(From, To, Log0) ->
     {Acc, Log} = ra_log:fold(From, To, fun (E, Acc) -> [E | Acc] end, [], Log0),
     {lists:reverse(Acc), Log}.
+
+flush() ->
+    receive
+        Any ->
+            ct:pal("flush ~p", [Any]),
+            flush()
+    after 0 ->
+              ok
+    end.