Merge pull request #2001 from rabbitmq/rabbitmq-server-2000

Move check for active readers to message store GC action function

(cherry picked from commit 5c80bea)

Conflicts:
	src/rabbit_msg_store.erl
	test/queue_parallel_SUITE.erl

Author: michaelklishin

src/rabbit_msg_store.erl

@@ -23,7 +23,7 @@
          client_ref/1, close_all_indicated/1,
          write/3, write_flow/3, read/2, contains/2, remove/2]).
 
--export([set_maximum_since_use/2, has_readers/2, combine_files/3,
+-export([set_maximum_since_use/2, combine_files/3,
          delete_file/2]). %% internal
 
 -export([transform_dir/3, force_recovery/2]). %% upgrade
@@ -192,10 +192,7 @@
 -spec remove([rabbit_types:msg_id()], client_msstate()) -> 'ok'.
 
 -spec set_maximum_since_use(server(), non_neg_integer()) -> 'ok'.
--spec has_readers(non_neg_integer(), gc_state()) -> boolean().
--spec combine_files(non_neg_integer(), non_neg_integer(), gc_state()) ->
-                              deletion_thunk().
--spec delete_file(non_neg_integer(), gc_state()) -> deletion_thunk().
+
 -spec force_recovery(file:filename(), server()) -> 'ok'.
 -spec transform_dir(file:filename(), server(),
         fun ((any()) -> (rabbit_types:ok_or_error2(msg(), any())))) -> 'ok'.
@@ -1965,28 +1962,48 @@ cleanup_after_file_deletion(File,
 %% garbage collection / compaction / aggregation -- external
 %%----------------------------------------------------------------------------
 
-has_readers(File, #gc_state { file_summary_ets = FileSummaryEts }) ->
-    [#file_summary { locked = true, readers = Count }] =
-        ets:lookup(FileSummaryEts, File),
-    Count /= 0.
+-spec combine_files(non_neg_integer(), non_neg_integer(), gc_state()) ->
+                              {ok, deletion_thunk()} | {defer, non_neg_integer()}.
 
 combine_files(Source, Destination,
-              State = #gc_state { file_summary_ets = FileSummaryEts,
-                                  file_handles_ets = FileHandlesEts,
-                                  dir              = Dir,
-                                  msg_store        = Server }) ->
-    [#file_summary {
+              State = #gc_state { file_summary_ets = FileSummaryEts }) ->
+    [#file_summary{locked = true} = SourceSummary] =
+        ets:lookup(FileSummaryEts, Source),
+
+    [#file_summary{locked = true} = DestinationSummary] =
+        ets:lookup(FileSummaryEts, Destination),
+
+    case {SourceSummary, DestinationSummary} of
+        {#file_summary{readers = 0}, #file_summary{readers = 0}} ->
+            {ok, do_combine_files(SourceSummary, DestinationSummary,
+                                  Source, Destination, State)};
+        _ ->
+            rabbit_log:debug("Asked to combine files ~p and ~p but they have active readers. Deferring.",
+                             [Source, Destination]),
+            DeferredFiles = [FileSummary#file_summary.file
+                             || FileSummary <- [SourceSummary, DestinationSummary],
+                             FileSummary#file_summary.readers /= 0],
+            {defer, DeferredFiles}
+    end.
+
+do_combine_files(SourceSummary, DestinationSummary,
+                 Source, Destination,
+                 State = #gc_state { file_summary_ets = FileSummaryEts,
+                                     file_handles_ets = FileHandlesEts,
+                                     dir              = Dir,
+                                     msg_store        = Server }) ->
+    #file_summary {
         readers          = 0,
         left             = Destination,
         valid_total_size = SourceValid,
         file_size        = SourceFileSize,
-        locked           = true }] = ets:lookup(FileSummaryEts, Source),
-    [#file_summary {
+        locked           = true } = SourceSummary,
+    #file_summary {
         readers          = 0,
         right            = Source,
         valid_total_size = DestinationValid,
         file_size        = DestinationFileSize,
-        locked           = true }] = ets:lookup(FileSummaryEts, Destination),
+        locked           = true } = DestinationSummary,
 
     SourceName           = filenum_to_name(Source),
     DestinationName      = filenum_to_name(Destination),
@@ -2043,20 +2060,30 @@ combine_files(Source, Destination,
               {#file_summary.file_size,        TotalValidData}]),
 
     Reclaimed = SourceFileSize + DestinationFileSize - TotalValidData,
+    rabbit_log:debug("Combined segment files number ~p (source) and ~p (destination), reclaimed ~p bytes",
+                     [Source, Destination, Reclaimed]),
     gen_server2:cast(Server, {combine_files, Source, Destination, Reclaimed}),
     safe_file_delete_fun(Source, Dir, FileHandlesEts).
 
+-spec delete_file(non_neg_integer(), gc_state()) -> {ok, deletion_thunk()} | {defer, non_neg_integer()}.
+
 delete_file(File, State = #gc_state { file_summary_ets = FileSummaryEts,
                                       file_handles_ets = FileHandlesEts,
                                       dir              = Dir,
                                       msg_store        = Server }) ->
-    [#file_summary { valid_total_size = 0,
-                     locked           = true,
-                     file_size        = FileSize,
-                     readers          = 0 }] = ets:lookup(FileSummaryEts, File),
-    {[], 0} = load_and_vacuum_message_file(File, State),
-    gen_server2:cast(Server, {delete_file, File, FileSize}),
-    safe_file_delete_fun(File, Dir, FileHandlesEts).
+    case ets:lookup(FileSummaryEts, File) of
+        [#file_summary { valid_total_size = 0,
+                         locked           = true,
+                         file_size        = FileSize,
+                         readers          = 0 }] ->
+            {[], 0} = load_and_vacuum_message_file(File, State),
+            gen_server2:cast(Server, {delete_file, File, FileSize}),
+            {ok, safe_file_delete_fun(File, Dir, FileHandlesEts)};
+        [#file_summary{readers = Readers}] when Readers > 0 ->
+            rabbit_log:debug("Asked to delete file ~p but it has active readers. Deferring.",
+                             [File]),
+            {defer, [File]}
+    end.
 
 load_and_vacuum_message_file(File, State = #gc_state { dir = Dir }) ->
     %% Messages here will be end-of-file at start-of-list

src/rabbit_msg_store_gc.erl

@@ -116,15 +116,13 @@ attempt_action(Action, Files,
                State = #state { pending_no_readers = Pending,
                                 on_action          = Thunks,
                                 msg_store_state    = MsgStoreState }) ->
-    case [File || File <- Files,
-                  rabbit_msg_store:has_readers(File, MsgStoreState)] of
-        []         -> State #state {
-                        on_action = lists:filter(
-                                      fun (Thunk) -> not Thunk() end,
-                                      [do_action(Action, Files, MsgStoreState) |
-                                       Thunks]) };
-        [File | _] -> Pending1 = maps:put(File, {Action, Files}, Pending),
-                      State #state { pending_no_readers = Pending1 }
+    case do_action(Action, Files, MsgStoreState) of
+        {ok, OkThunk} ->
+            State#state{on_action = lists:filter(fun (Thunk) -> not Thunk() end,
+                                                 [OkThunk | Thunks])};
+        {defer, [File | _]} ->
+            Pending1 = maps:put(File, {Action, Files}, Pending),
+            State #state { pending_no_readers = Pending1 }
     end.
 
 do_action(combine, [Source, Destination], MsgStoreState) ->

test/confirms_rejects_SUITE.erl

@@ -36,7 +36,7 @@ init_per_group(Group, Config) ->
       rabbit_ct_broker_helpers:setup_steps() ++
       rabbit_ct_client_helpers:setup_steps()).
 
-end_per_group(Group, Config) ->
+end_per_group(_Group, Config) ->
     rabbit_ct_helpers:run_steps(Config,
       rabbit_ct_client_helpers:teardown_steps() ++
       rabbit_ct_broker_helpers:teardown_steps()).
@@ -255,4 +255,4 @@ clean_acks_mailbox() ->
         {'basic.nack', _, _, _} -> clean_acks_mailbox()
     after
         1000 -> done
-    end.
+    end.

test/unit_inbroker_non_parallel_SUITE.erl

@@ -315,7 +315,6 @@ log_file_fails_to_initialise_during_startup(Config) ->
 
 log_file_fails_to_initialise_during_startup1(_Config) ->
     [LogFile|_] = rabbit:log_locations(),
-    Suffix = ".0",
 
     %% start application with logging to directory with no
     %% write permissions