Merge pull request #11112 from rabbitmq/loic-faster-cq-shared-store-gc

4.x: Additional CQv2 message store optimisations

Author: lhoguin

deps/rabbit/BUILD.bazel

@@ -39,7 +39,6 @@ _APP_ENV = """[
 	    {vm_memory_calculation_strategy, rss},
 	    {memory_monitor_interval, 2500},
 	    {disk_free_limit, 50000000}, %% 50MB
-	    {msg_store_index_module, rabbit_msg_store_ets_index},
 	    {backing_queue_module, rabbit_variable_queue},
 	    %% 0 ("no limit") would make a better default, but that
 	    %% breaks the QPid Java client

deps/rabbit/Makefile

@@ -19,7 +19,6 @@ define PROJECT_ENV
 	    {vm_memory_calculation_strategy, rss},
 	    {memory_monitor_interval, 2500},
 	    {disk_free_limit, 50000000}, %% 50MB
-	    {msg_store_index_module, rabbit_msg_store_ets_index},
 	    {backing_queue_module, rabbit_variable_queue},
 	    %% 0 ("no limit") would make a better default, but that
 	    %% breaks the QPid Java client

deps/rabbit/app.bzl

@@ -170,7 +170,6 @@ def all_beam_files(name = "all_beam_files"):
             "src/rabbit_mirror_queue_misc.erl",
             "src/rabbit_mnesia.erl",
             "src/rabbit_msg_store.erl",
-            "src/rabbit_msg_store_ets_index.erl",
             "src/rabbit_msg_store_gc.erl",
             "src/rabbit_networking.erl",
             "src/rabbit_networking_store.erl",
@@ -431,7 +430,6 @@ def all_test_beam_files(name = "all_test_beam_files"):
             "src/rabbit_mirror_queue_misc.erl",
             "src/rabbit_mnesia.erl",
             "src/rabbit_msg_store.erl",
-            "src/rabbit_msg_store_ets_index.erl",
             "src/rabbit_msg_store_gc.erl",
             "src/rabbit_networking.erl",
             "src/rabbit_networking_store.erl",
@@ -711,7 +709,6 @@ def all_srcs(name = "all_srcs"):
             "src/rabbit_mirror_queue_misc.erl",
             "src/rabbit_mnesia.erl",
             "src/rabbit_msg_store.erl",
-            "src/rabbit_msg_store_ets_index.erl",
             "src/rabbit_msg_store_gc.erl",
             "src/rabbit_networking.erl",
             "src/rabbit_networking_store.erl",

deps/rabbit/src/rabbit_classic_queue_index_v2.erl

@@ -1125,8 +1125,11 @@ queue_index_walker({next, Gatherer}) when is_pid(Gatherer) ->
         empty ->
             ok = gatherer:stop(Gatherer),
             finished;
+        %% From v1 index walker. @todo Remove when no longer possible to convert from v1.
         {value, {MsgId, Count}} ->
-            {MsgId, Count, {next, Gatherer}}
+            {MsgId, Count, {next, Gatherer}};
+        {value, MsgIds} ->
+            {MsgIds, {next, Gatherer}}
     end.
 
 queue_index_walker_reader(#resource{ virtual_host = VHost } = Name, Gatherer) ->
@@ -1153,27 +1156,30 @@ queue_index_walker_segment(F, Gatherer) ->
         {ok, <<?MAGIC:32,?VERSION:8,
                FromSeqId:64/unsigned,ToSeqId:64/unsigned,
                _/bits>>} ->
-            queue_index_walker_segment(Fd, Gatherer, 0, ToSeqId - FromSeqId);
+            queue_index_walker_segment(Fd, Gatherer, 0, ToSeqId - FromSeqId, []);
         _ ->
             %% Invalid segment file. Skip.
             ok
     end,
     ok = file:close(Fd).
 
-queue_index_walker_segment(_, _, N, N) ->
+queue_index_walker_segment(_, Gatherer, N, N, Acc) ->
     %% We reached the end of the segment file.
+    gatherer:sync_in(Gatherer, Acc),
     ok;
-queue_index_walker_segment(Fd, Gatherer, N, Total) ->
+queue_index_walker_segment(Fd, Gatherer, N, Total, Acc) ->
     case file:read(Fd, ?ENTRY_SIZE) of
         %% We found a non-ack persistent entry. Gather it.
         {ok, <<1,_:7,1:1,_,1,Id:16/binary,_/bits>>} ->
-            gatherer:sync_in(Gatherer, {Id, 1}),
-            queue_index_walker_segment(Fd, Gatherer, N + 1, Total);
+            queue_index_walker_segment(Fd, Gatherer, N + 1, Total, [Id|Acc]);
         %% We found an ack, a transient entry or a non-entry. Skip it.
         {ok, _} ->
-            queue_index_walker_segment(Fd, Gatherer, N + 1, Total);
+            queue_index_walker_segment(Fd, Gatherer, N + 1, Total, Acc);
         %% We reached the end of a partial segment file.
+        eof when Acc =:= [] ->
+            ok;
         eof ->
+            gatherer:sync_in(Gatherer, Acc),
             ok
     end.