Sparse writes

Most things work now. Instead of passing ranges around we now
pass ra_seq instances which is a compacted sequence of items.

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,
@@ -685,7 +696,18 @@ handle_event({segments, TidRanges, NewSegs},
     %% the tid ranges arrive in the reverse order they were written
     %% (new -> old) so we need to foldr here to process the oldest first
     Mt = lists:foldr(
-           fun ({Tid, Range}, Acc0) ->
+           fun ({Tid, Seq}, Acc0) ->
+                   %% TODO: HACK: only handles single range ra_seqs
+                   Range = case Seq of
+                               [] ->
+                                   undefined;
+                               [{_, _} = R] ->
+                                   R;
+                               [I] -> ra_range:new(I, I);
+                               [I2, I] when I+1 == I2 ->
+                                   ra_range:new(I, I2)
+                           end,
+                   ct:pal("Range ~p Seq ~p", [Range, Seq]),
                    {Spec, Acc} = ra_mt:record_flushed(Tid, Range, Acc0),
                     ok = ra_log_ets:execute_delete(Names, UId, Spec),
                     Acc
@@ -1200,7 +1222,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 +1345,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

@@ -65,7 +65,7 @@ start_link(#{name := Name} = Config) ->
     gen_server:start_link({local, Name}, ?MODULE, [Config], []).
 
 -spec accept_mem_tables(atom() | pid(),
-                        #{ra_uid() => [{ets:tid(), ra:range()}]},
+                        #{ra_uid() => [{ets:tid(), ra_seq:state()}]},
                         string()) -> ok.
 accept_mem_tables(_SegmentWriter, Tables, undefined)
   when map_size(Tables) == 0 ->
@@ -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, Seq},
                       #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 "
@@ -324,8 +325,7 @@ flush_mem_table_range(ServerUId, {Tid, {StartIdx0, EndIdx}},
             %% clean up the tables for this process
             [];
         Segment0 ->
-            case append_to_segment(ServerUId, Tid, StartIdx, EndIdx,
-                                   Segment0, State) of
+            case append_to_segment(ServerUId, Tid, Seq, Segment0, State) of
                 undefined ->
                     ?WARN("segment_writer: skipping segments for ~w as
                            directory ~ts disappeared whilst writing",
@@ -366,31 +366,37 @@ send_segments(System, ServerUId, TidRanges, SegRefs) ->
                    [ServerUId, "No Pid"]),
             %% delete from the memtable on the non-running server's behalf
             [begin
+                 %% TODO: HACK: this is a hack to get a full range out of a
+                 %% sequent, ideally the mt should take the ra_seq and
+                 %% delete from that
+                 Range = {ra_seq:first(Seq), ra_seq:last(Seq)},
                  _  = catch ra_mt:delete({range, Tid, Range})
-             end || {Tid, Range} <- TidRanges],
+             end || {Tid, Seq} <- TidRanges],
             ok;
         Pid ->
             Pid ! {ra_log_event, {segments, TidRanges, SegRefs}},
             ok
     end.
 
-append_to_segment(UId, Tid, StartIdx0, EndIdx, Seg, State) ->
-    StartIdx = start_index(UId, StartIdx0),
-    % EndIdx + 1 because FP
-    append_to_segment(UId, Tid, StartIdx, EndIdx+1, Seg, [], State).
+append_to_segment(UId, Tid, Seq0, Seg, State) ->
+    FirstIdx = ra_seq:first(Seq0),
+    StartIdx = start_index(UId, FirstIdx),
+    %% TODO combine flor and iterator into one operation
+    Seq = ra_seq:floor(StartIdx, Seq0),
+    SeqIter = ra_seq:iterator(Seq),
+    append_to_segment(UId, Tid, ra_seq:next(SeqIter), Seg, [], State).
 
-append_to_segment(_, _, StartIdx, EndIdx, Seg, Closed, _State)
-  when StartIdx >= EndIdx ->
+append_to_segment(_, _, end_of_seq, Seg, Closed, _State) ->
     {Seg, Closed};
-append_to_segment(UId, Tid, Idx, EndIdx, Seg0, Closed, State) ->
+append_to_segment(UId, Tid, {Idx, SeqIter} = Cur, Seg0, Closed, State) ->
     try ets:lookup(Tid, Idx) of
         [] ->
             StartIdx = start_index(UId, Idx),
             case Idx < StartIdx of
                 true ->
                     %% a snapshot must have been completed after we last checked
                     %% the start idx, continue flush from new start index.
-                    append_to_segment(UId, Tid, StartIdx, EndIdx, Seg0,
+                    append_to_segment(UId, Tid, next_ra_seq(StartIdx, SeqIter), Seg0,
                                       Closed, State);
                 false ->
                     %% oh dear, an expected index was not found in the mem table.
@@ -419,7 +425,7 @@ append_to_segment(UId, Tid, Idx, EndIdx, Seg0, Closed, State) ->
                     %% the segment index but is probably good enough to get comparative
                     %% data rates for different Ra components
                     ok = counters:add(State#state.counter, ?C_BYTES_WRITTEN, DataSize),
-                    append_to_segment(UId, Tid, Idx+1, EndIdx, Seg, Closed, State);
+                    append_to_segment(UId, Tid, ra_seq:next(SeqIter), Seg, Closed, State);
                 {error, full} ->
                     % close and open a new segment
                     case open_successor_segment(Seg0, State#state.segment_conf) of
@@ -432,8 +438,14 @@ append_to_segment(UId, Tid, Idx, EndIdx, Seg0, Closed, State) ->
                             %% re-evaluate snapshot state for the server in case
                             %% a snapshot has completed during segment flush
                             StartIdx = start_index(UId, Idx),
-                            append_to_segment(UId, Tid, StartIdx, EndIdx, Seg,
-                                              [Seg0 | Closed], State)
+                            Next = case StartIdx == Idx of
+                                       true ->
+                                           Cur;
+                                       false ->
+                                           next_ra_seq(StartIdx, SeqIter)
+                                   end,
+                            append_to_segment(UId, Tid, Next,
+                                              Seg, [Seg0 | Closed], State)
                     end;
                 {error, Posix} ->
                     FileName = ra_log_segment:filename(Seg0),
@@ -549,3 +561,13 @@ maybe_upgrade_segment_file_names(System, DataDir) ->
             ok
     end.
 
+next_ra_seq(Idx, Iter0) ->
+    case ra_seq:next(Iter0) of
+        end_of_seq ->
+            end_of_seq;
+        {I, _} = Next
+          when I >= Idx ->
+            Next;
+        {_, Iter} ->
+            next_ra_seq(Idx, Iter)
+    end.