Permit the GC to cope with multiple concurrent pending GCs. In the case of a GC being blocked, we don't care if it's blocked on one file or many: we simply store that it's blocked on one, as we know we're going to test both before doing any real work, so this is safe.

2 files changed, 36 insertions, 30 deletions

diff --git a/src/rabbit_msg_store.erl b/src/rabbit_msg_store.erl
index 3c82caab6e..841f37072d 100644
--- a/src/rabbit_msg_store.erl
+++ b/src/rabbit_msg_store.erl
@@ -38,7 +38,7 @@
          write/4, read/3, contains/2, remove/2, release/2, sync/3]).
 
 -export([sync/1, gc_done/4, set_maximum_since_use/2,
-         gc/3, has_no_readers/2]). %% internal
+         gc/3, has_readers/2]). %% internal
 
 -export([init/1, handle_call/3, handle_cast/2, handle_info/2,
          terminate/2, code_change/3, prioritise_call/3, prioritise_cast/2]).
@@ -159,7 +159,7 @@
 -spec(set_maximum_since_use/2 :: (server(), non_neg_integer()) -> 'ok').
 -spec(gc/3 :: (non_neg_integer(), non_neg_integer(), gc_state()) ->
                    non_neg_integer()).
--spec(has_no_readers/2 :: (non_neg_integer(), gc_state()) -> boolean()).
+-spec(has_readers/2 :: (non_neg_integer(), gc_state()) -> boolean()).
 
 -endif.
 
@@ -1547,10 +1547,10 @@ delete_file_if_empty(File, State = #msstate {
 %% garbage collection / compaction / aggregation -- external
 %%----------------------------------------------------------------------------
 
-has_no_readers(File, #gc_state { file_summary_ets = FileSummaryEts }) ->
+has_readers(File, #gc_state { file_summary_ets = FileSummaryEts }) ->
     [#file_summary { locked = true, readers = Count }] =
         ets:lookup(FileSummaryEts, File),
-    Count == 0.
+    Count /= 0.
 
 gc(SrcFile, DstFile, State = #gc_state { file_summary_ets = FileSummaryEts }) ->
     [SrcObj = #file_summary {
diff --git a/src/rabbit_msg_store_gc.erl b/src/rabbit_msg_store_gc.erl
index 6e0461bb6c..3b98e1df8e 100644
--- a/src/rabbit_msg_store_gc.erl
+++ b/src/rabbit_msg_store_gc.erl
@@ -41,9 +41,9 @@
          terminate/2, code_change/3, prioritise_cast/2]).
 
 -record(state,
-        {parent,
-         scheduled,
-         msg_store_state
+        { parent,
+          pending_no_readers,
+          msg_store_state
         }).
 
 -include("rabbit.hrl").
@@ -84,9 +84,9 @@ set_maximum_since_use(Pid, Age) ->
 init([Parent, MsgStoreState]) ->
     ok = file_handle_cache:register_callback(?MODULE, set_maximum_since_use,
                                              [self()]),
-    {ok, #state { parent          = Parent,
-                  scheduled       = undefined,
-                  msg_store_state = MsgStoreState }, hibernate,
+    {ok, #state { parent             = Parent,
+                  pending_no_readers = dict:new(),
+                  msg_store_state    = MsgStoreState }, hibernate,
      {backoff, ?HIBERNATE_AFTER_MIN, ?HIBERNATE_AFTER_MIN, ?DESIRED_HIBERNATE}}.
 
 prioritise_cast({set_maximum_since_use, _Age}, _State) -> 8;
@@ -95,18 +95,21 @@ prioritise_cast(_Msg,                          _State) -> 0.
 handle_call(stop, _From, State) ->
     {stop, normal, ok, State}.
 
-handle_cast({gc, Source, Destination},
-            State = #state { scheduled = undefined }) ->
-    {noreply, attempt_gc(State #state { scheduled = {Source, Destination} }),
-     hibernate};
+handle_cast({gc, Source, Destination}, State) ->
+    {noreply, attempt_gc(Source, Destination, State), hibernate};
 
 handle_cast({no_readers, File},
-            State = #state { scheduled = {Source, Destination} })
-  when File =:= Source orelse File =:= Destination ->
-    {noreply, attempt_gc(State), hibernate};
-
-handle_cast({no_readers, _File}, State) ->
-    {noreply, State, hibernate};
+            State = #state { pending_no_readers = Pending }) ->
+    State1 = case dict:find(File, Pending) of
+                 error ->
+                     State;
+                 {ok, {Source, Destination}} ->
+                     attempt_gc(
+                       Source, Destination,
+                       State #state { pending_no_readers =
+                                          dict:erase(File, Pending) })
+             end,
+    {noreply, State1, hibernate};
 
 handle_cast({set_maximum_since_use, Age}, State) ->
     ok = file_handle_cache:set_maximum_since_use(Age),
@@ -121,17 +124,20 @@ terminate(_Reason, State) ->
 code_change(_OldVsn, State, _Extra) ->
     {ok, State}.
 
-attempt_gc(State = #state { parent          = Parent,
-                            scheduled       = {Source, Destination},
-                            msg_store_state = MsgStoreState }) ->
-    case lists:all(fun (File) ->
-                           rabbit_msg_store:has_no_readers(File, MsgStoreState)
-                   end, [Source, Destination]) of
-        true ->
+attempt_gc(Source, Destination,
+           State = #state { parent             = Parent,
+                            pending_no_readers = Pending,
+                            msg_store_state    = MsgStoreState }) ->
+    case lists:filter(fun (File) ->
+                              rabbit_msg_store:has_readers(File, MsgStoreState)
+                      end, [Source, Destination]) of
+        [] ->
             Reclaimed = rabbit_msg_store:gc(Source, Destination, MsgStoreState),
             ok = rabbit_msg_store:gc_done(Parent, Reclaimed, Source,
                                           Destination),
-            State #state { scheduled = undefined };
-        false ->
-            State
+            State;
+        [File | _] ->
+            State #state { pending_no_readers =
+                               dict:store(File, {Source, Destination}, Pending)
+                         }
     end.