major compactions

Author: kjnilsson

src/ra_log.erl

@@ -227,12 +227,11 @@ init(#{uid := UId,
     MtRange = ra_mt:range(Mt0),
     SegRefs = my_segrefs(UId, SegWriter),
     Reader = ra_log_segments:init(UId, Dir, MaxOpen, AccessPattern, SegRefs,
-                                Names, Counter),
+                                  Counter, LogId),
     SegmentRange = ra_log_segments:range(Reader),
     %% TODO: check ra_range:add/2 actually performas the correct logic we expect
     Range = ra_range:add(MtRange, SegmentRange),
 
-    %% TODO: review thi
     [begin
          ?DEBUG("~ts: deleting overwritten segment ~w",
                 [LogId, SR]),
@@ -1288,21 +1287,20 @@ delete_everything(#?MODULE{cfg = #cfg{uid = UId,
 
 -spec release_resources(non_neg_integer(),
                         sequential | random, state()) -> state().
-release_resources(MaxOpenSegments,
-                  AccessPattern,
+release_resources(MaxOpenSegments, AccessPattern,
                   #?MODULE{cfg = #cfg{uid = UId,
+                                      log_id = LogId,
                                       directory = Dir,
-                                      counter = Counter,
-                                      names = Names},
+                                      counter = Counter},
                            reader = Reader} = State) ->
     ActiveSegs = ra_log_segments:segment_refs(Reader),
     % close all open segments
     % deliberately ignoring return value
     _ = ra_log_segments:close(Reader),
     %% open a new segment with the new max open segment value
     State#?MODULE{reader = ra_log_segments:init(UId, Dir, MaxOpenSegments,
-                                              AccessPattern,
-                                              ActiveSegs, Names, Counter)}.
+                                                AccessPattern, ActiveSegs,
+                                                Counter, LogId)}.
 
 
 %%% Local functions
@@ -1482,45 +1480,16 @@ my_segrefs(UId, SegWriter) ->
                         %% if a server recovered when a segment had been opened
                         %% but never had any entries written the segref would be
                         %% undefined
-                        case ra_log_segment:segref(File) of
-                            undefined ->
-                                Acc;
-                            SegRef ->
-                                [SegRef | Acc]
+                        case ra_log_segment:info(File) of
+                            #{ref := SegRef,
+                              file_type := regular}
+                              when is_tuple(SegRef) ->
+                                [SegRef | Acc];
+                            _ ->
+                                Acc
                         end
                 end, [], SegFiles).
 
-% recover_ranges(UId, MtRange, SegWriter) ->
-%     % 1. check mem_tables (this assumes wal has finished recovering
-%     % which means it is essential that ra_servers are part of the same
-%     % supervision tree
-%     % 2. check segments
-%     SegFiles = ra_log_segment_writer:my_segments(SegWriter, UId),
-%     SegRefs = lists:foldl(
-%                 fun (File, Acc) ->
-%                         %% if a server recovered when a segment had been opened
-%                         %% but never had any entries written the segref would be
-%                         %% undefined
-%                         case ra_log_segment:segref(File) of
-%                             undefined ->
-%                                 Acc;
-%                             SegRef ->
-%                                 [SegRef | Acc]
-%                         end
-%                 end, [], SegFiles),
-%     SegRanges = [Range || {Range, _} <- SegRefs],
-%     Ranges = [MtRange | SegRanges],
-%     {pick_range(Ranges, undefined), SegRefs}.
-
-% picks the current range from a sorted (newest to oldest) list of ranges
-% pick_range([], Res) ->
-%     Res;
-% pick_range([H | Tail], undefined) ->
-%     pick_range(Tail, H);
-% pick_range([{Fst, _Lst} | Tail], {CurFst, CurLst}) ->
-%     pick_range(Tail, {min(Fst, CurFst), CurLst}).
-
-
 %% TODO: implement synchronous writes using gen_batch_server:call/3
 await_written_idx(Idx, Term, Log0) ->
     receive
@@ -1536,17 +1505,6 @@ await_written_idx(Idx, Term, Log0) ->
               throw(ra_log_append_timeout)
     end.
 
-% log_update_wait_n(0) ->
-%     ok;
-% log_update_wait_n(N) ->
-%     receive
-%         ra_log_update_processed ->
-%             log_update_wait_n(N - 1)
-%     after 1500 ->
-%               %% just go ahead anyway
-%               ok
-%     end.
-
 incr_counter(#cfg{counter = Cnt}, Ix, N) when Cnt =/= undefined ->
     counters:add(Cnt, Ix, N);
 incr_counter(#cfg{counter = undefined}, _Ix, _N) ->

src/ra_log_segment.erl

@@ -22,6 +22,7 @@
          max_count/1,
          filename/1,
          segref/1,
+         info/1,
          is_same_as/2,
          copy/3]).
 
@@ -56,7 +57,7 @@
               fd :: option(file:io_device()),
               index_size :: pos_integer(),
               access_pattern :: sequential | random,
-              file_advise = normal ::  posix_file_advise(),
+              file_advise = normal :: posix_file_advise(),
               mode = append :: read | append,
               compute_checksums = true :: boolean()}).
 
@@ -453,6 +454,39 @@ segref(Filename) ->
     close(Seg),
     SegRef.
 
+-spec info(file:filename_all()) ->
+    #{size => non_neg_integer(),
+      max_count => non_neg_integer(),
+      file_type => regular | symlink,
+      ctime => integer(),
+      links => non_neg_integer(),
+      num_entries => non_neg_integer(),
+      ref => option(ra_log:segment_ref()),
+      indexes => ra_seq:state()
+     }.
+info(Filename)
+  when not is_tuple(Filename) ->
+    %% TODO: this can be much optimised by a specialised index parsing
+    %% function
+    {ok, Seg} = open(Filename, #{mode => read}),
+    Index = Seg#state.index,
+    {ok, #file_info{type = T,
+                    links = Links,
+                    ctime = CTime}} = file:read_link_info(Filename,
+                                                          [raw, {time, posix}]),
+
+    Info = #{size => Seg#state.data_write_offset,
+             file_type => T,
+             links => Links,
+             ctime => CTime,
+             max_count => max_count(Seg),
+             num_entries => maps:size(Index),
+             ref => segref(Seg),
+             indexes => ra_seq:from_list(maps:keys(Index))
+            },
+    close(Seg),
+    Info.
+
 -spec is_same_as(state(), file:filename_all()) -> boolean().
 is_same_as(#state{cfg = #cfg{filename = Fn0}}, Fn) ->
     is_same_filename_all(Fn0, Fn).