Merge pull request #523 from rabbitmq/ra_log_ets-timeout

Use infinity timeout for ra_log_ets:mem_table_please

Author: kjnilsson

src/ra_lib.erl

@@ -303,7 +303,9 @@ partition_parallel(F, Es) ->
 
 partition_parallel(F, Es, Timeout) ->
     Parent = self(),
-    Running = [{spawn_monitor(fun() -> Parent ! {self(), F(E)} end), E}
+    Running = [{spawn_monitor(fun() ->
+                                      Parent ! {self(), F(E)}
+                              end), E}
                || E <- Es],
     collect(Running, {[], []}, Timeout).
 
@@ -317,8 +319,8 @@ collect([{{Pid, MRef}, E} | Next], {Left, Right}, Timeout) ->
         {Pid, false} ->
             erlang:demonitor(MRef, [flush]),
             collect(Next, {Left, [E | Right]}, Timeout);
-        {'DOWN', MRef, process, Pid, _Reason} ->
-            collect(Next, {Left, [E | Right]}, Timeout)
+        {'DOWN', MRef, process, Pid, Reason} ->
+            collect(Next, {Left, [{E, Reason} | Right]}, Timeout)
     after Timeout ->
               exit(partition_parallel_timeout)
     end.

src/ra_log.erl

@@ -669,9 +669,9 @@ handle_event({written, _Term, {FromIdx, _}} = Evt,
                   [LogId, FromIdx, Expected]),
             {resend_from(Expected, State0), []};
         false ->
-            ?INFO("~ts: ra_log: written gap detected at ~b but is outside
-                  of mem table range. Updating last written index to ~b!",
-                  [LogId, FromIdx, Expected]),
+            ?DEBUG("~ts: ra_log: written gap detected at ~b but is outside
+                  of mem table range ~w. Updating last written index to ~b!",
+                   [LogId, FromIdx, MtRange, Expected]),
             %% if the entry is not in the mem table we may have missed a
             %% written event due to wal crash. Accept written event by updating
             %% last written index term and recursing
@@ -686,7 +686,8 @@ handle_event({segments, TidRanges, NewSegs},
                       readers = Readers
                      } = State0) ->
     Reader = ra_log_reader:update_segments(NewSegs, Reader0),
-    put_counter(Cfg, ?C_RA_SVR_METRIC_NUM_SEGMENTS, ra_log_reader:segment_ref_count(Reader)),
+    put_counter(Cfg, ?C_RA_SVR_METRIC_NUM_SEGMENTS,
+                ra_log_reader:segment_ref_count(Reader)),
     %% 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(

src/ra_log_ets.erl

@@ -54,7 +54,7 @@ mem_table_please(#{log_ets := Name,
                    open_mem_tbls := OpnMemTbls}, UId, Mode) ->
     case ets:lookup(OpnMemTbls, UId) of
         [] ->
-            case gen_server:call(Name, {mem_table_please, UId, #{}}) of
+            case gen_server:call(Name, {mem_table_please, UId, #{}}, infinity) of
                 {ok, [Tid | Rem]} ->
                     Mt = lists:foldl(
                            fun (T, Acc) ->
@@ -137,18 +137,20 @@ handle_call(Request, _From, State) ->
 
 handle_cast({exec_delete, UId, Spec},
             #state{names = #{open_mem_tbls := MemTables}} = State) ->
+
+    %% delete from open mem tables if {delete, tid()}
     case Spec of
         {delete, Tid} ->
             ets:delete_object(MemTables, {UId, Tid});
         _ ->
             ok
     end,
 
-    %% TODO: delete from ra_log_open_mem_tables if {delete, tid()}
     try timer:tc(fun () -> ra_mt:delete(Spec) end) of
         {Time, Num} ->
-            ?DEBUG_IF(Time > 25_000, "ra_log_ets: ra_mt:delete/1 took ~bms to delete ~w ~b entries",
-                   [Time div 1000, Spec, Num]),
+            ?DEBUG_IF(Time > 25_000,
+                      "ra_log_ets: ra_mt:delete/1 took ~bms to delete ~w ~b entries",
+                      [Time div 1000, Spec, Num]),
             ok
     catch
         _:Err ->
@@ -166,8 +168,9 @@ handle_cast({delete_mem_tables, UId},
     [begin
          try timer:tc(fun () -> ets_delete(Tid) end) of
              {Time, true} ->
-                 ?DEBUG_IF(Time > 25_000, "ra_log_ets: ets:delete/1 took ~bms to delete ~w",
-                        [Time div 1000, Tid]),
+                 ?DEBUG_IF(Time > 25_000,
+                           "ra_log_ets: ets:delete/1 took ~bms to delete ~w",
+                           [Time div 1000, Tid]),
                  ok
          catch
              _:Err ->

src/ra_log_segment_writer.erl

@@ -150,8 +150,8 @@ handle_cast({mem_tables, Ranges, WalFile}, #state{data_dir = Dir,
                                false ->
                                    %% delete all tids as the uid is not
                                    %% registered
-                                   ?DEBUG("segment_writer in '~w': deleting memtable "
-                                          "for ~w as not a registered uid",
+                                   ?DEBUG("segment_writer in '~ts': deleting memtable "
+                                          "for ~ts as not a registered uid",
                                           [System, UId]),
                                    ok = ra_log_ets:delete_mem_tables(Names, UId),
                                    Acc