CQ: Simplify handling of confirms from index writes

lhoguin · lhoguin · commit 179f68734744 · 2025-11-04T13:16:03.000+01:00
We avoid an extra unnecessary message to the queue.
diff --git a/deps/rabbit/src/rabbit_classic_queue_index_v2.erl b/deps/rabbit/src/rabbit_classic_queue_index_v2.erl
@@ -7,7 +7,7 @@
 
 -module(rabbit_classic_queue_index_v2).
 
--export([erase/1, init/2, reset_state/1, recover/5,
+-export([erase/1, init/1, reset_state/1, recover/4,
          terminate/3, delete_and_terminate/1,
          info/1, publish/7, ack/2, read/3]).
 
@@ -144,17 +144,11 @@
 
     %% File descriptors. We will keep up to 4 FDs
     %% at a time. See comments in reduce_fd_usage/2.
-    fds = #{} :: #{non_neg_integer() => file:fd()},
-
-    %% This fun must be called when messages that expect
-    %% confirms have either an ack or their entry
-    %% written to disk and file:sync/1 has been called.
-    on_sync :: on_sync_fun()
+    fds = #{} :: #{non_neg_integer() => file:fd()}
 }).
 
 -type state() :: #qi{}.
 
--type on_sync_fun() :: fun ((sets:set()) -> ok).
 -type contains_predicate() :: fun ((rabbit_types:msg_id()) -> boolean()).
 -type shutdown_terms() :: list() | 'non_clean_shutdown'.
 
@@ -168,24 +162,21 @@ erase(#resource{ virtual_host = VHost } = Name) ->
     Dir = queue_dir(VHostDir, Name),
     erase_index_dir(Dir).
 
--spec init(rabbit_amqqueue:name(), on_sync_fun()) -> state().
-
-%% We do not embed messages and as a result never need the OnSyncMsgFun.
+-spec init(rabbit_amqqueue:name()) -> state().
 
-init(#resource{ virtual_host = VHost } = Name, OnSyncFun) ->
-    ?DEBUG("~0p ~0p ~0p", [Name, OnSyncFun]),
+init(#resource{ virtual_host = VHost } = Name) ->
+    ?DEBUG("~0p", [Name]),
     VHostDir = rabbit_vhost:msg_store_dir_path(VHost),
     Dir = queue_dir(VHostDir, Name),
     false = rabbit_file:is_file(Dir), %% is_file == is file or dir
-    init1(Name, Dir, OnSyncFun).
+    init1(Name, Dir).
 
-init1(Name, Dir, OnSyncFun) ->
+init1(Name, Dir) ->
     ensure_queue_name_stub_file(Name, Dir),
     DirBin = rabbit_file:filename_to_binary(Dir),
     #qi{
         queue_name = Name,
-        dir = << DirBin/binary, "/" >>,
-        on_sync = OnSyncFun
+        dir = << DirBin/binary, "/" >>
     }.
 
 ensure_queue_name_stub_file(#resource{virtual_host = VHost, name = QName}, Dir) ->
@@ -197,15 +188,13 @@ ensure_queue_name_stub_file(#resource{virtual_host = VHost, name = QName}, Dir)
 -spec reset_state(State) -> State when State::state().
 
 reset_state(State = #qi{ queue_name     = Name,
-                         dir            = Dir,
-                         on_sync        = OnSyncFun }) ->
+                         dir            = Dir }) ->
     ?DEBUG("~0p", [State]),
     _ = delete_and_terminate(State),
-    init1(Name, rabbit_file:binary_to_filename(Dir), OnSyncFun).
+    init1(Name, rabbit_file:binary_to_filename(Dir)).
 
 -spec recover(rabbit_amqqueue:name(), shutdown_terms(), boolean(),
-                    contains_predicate(),
-                    on_sync_fun()) ->
+                    contains_predicate()) ->
                         {'undefined' | non_neg_integer(),
                          'undefined' | non_neg_integer(), state()}.
 
@@ -218,12 +207,11 @@ reset_state(State = #qi{ queue_name     = Name,
 -define(RECOVER_COUNTER_SIZE, 6).
 
 recover(#resource{ virtual_host = VHost, name = QueueName } = Name, Terms,
-        IsMsgStoreClean, ContainsCheckFun, OnSyncFun) ->
-    ?DEBUG("~0p ~0p ~0p ~0p ~0p", [Name, Terms, IsMsgStoreClean,
-                                   ContainsCheckFun, OnSyncFun]),
+        IsMsgStoreClean, ContainsCheckFun) ->
+    ?DEBUG("~0p ~0p ~0p ~0p", [Name, Terms, IsMsgStoreClean, ContainsCheckFun]),
     VHostDir = rabbit_vhost:msg_store_dir_path(VHost),
     Dir = queue_dir(VHostDir, Name),
-    State0 = init1(Name, Dir, OnSyncFun),
+    State0 = init1(Name, Dir),
     %% We go over all segments if either the index or the
     %% message store has/had to recover. Otherwise we just
     %% take our state from Terms.
@@ -922,22 +910,14 @@ parse_entries(<< Status:8,
 
 %% ----
 %%
-%% Syncing and flushing to disk requested by the queue.
-%% Note: the v2 no longer calls fsync, it only flushes.
+%% Flushing to disk requested by the queue.
 
 -spec sync(State) -> State when State::state().
 
-sync(State0 = #qi{ confirms = Confirms,
-                   on_sync = OnSyncFun }) ->
+sync(State0 = #qi{ confirms = Confirms }) ->
     ?DEBUG("~0p", [State0]),
     State = flush_buffer(State0, full, segment_entry_count()),
-    _ = case sets:is_empty(Confirms) of
-        true ->
-            ok;
-        false ->
-            OnSyncFun(Confirms)
-    end,
-    State#qi{ confirms = sets:new([{version,2}]) }.
+    {Confirms, State#qi{ confirms = sets:new([{version,2}]) }}.
 
 -spec needs_sync(state()) -> 'false' | 'confirms'.
 
diff --git a/deps/rabbit/src/rabbit_variable_queue.erl b/deps/rabbit/src/rabbit_variable_queue.erl
@@ -22,7 +22,7 @@
 -export([start/2, stop/1]).
 
 %% exported for testing only
--export([start_msg_store/3, stop_msg_store/1, init/4]).
+-export([start_msg_store/3, stop_msg_store/1, init/3]).
 
 -include("mc.hrl").
 -include_lib("stdlib/include/qlc.hrl").
@@ -404,17 +404,16 @@ stop_msg_store(VHost) ->
     ok.
 
 init(Queue, Recover, Callback) ->
-    init(
+    init1(
       Queue, Recover,
       fun (MsgIds, ActionTaken) ->
               msgs_written_to_disk(Callback, MsgIds, ActionTaken)
-      end,
-      fun (MsgIds) -> msg_indices_written_to_disk(Callback, MsgIds) end).
+      end).
 
-init(Q, new, MsgOnDiskFun, MsgIdxOnDiskFun) when ?is_amqqueue(Q) ->
+init1(Q, new, MsgOnDiskFun) when ?is_amqqueue(Q) ->
     QueueName = amqqueue:get_name(Q),
     IsDurable = amqqueue:is_durable(Q),
-    IndexState = rabbit_classic_queue_index_v2:init(QueueName, MsgIdxOnDiskFun),
+    IndexState = rabbit_classic_queue_index_v2:init(QueueName),
     StoreState = rabbit_classic_queue_store_v2:init(QueueName),
     VHost = QueueName#resource.virtual_host,
     init(IsDurable, IndexState, StoreState, 0, 0, [],
@@ -427,7 +426,7 @@ init(Q, new, MsgOnDiskFun, MsgIdxOnDiskFun) when ?is_amqqueue(Q) ->
                                VHost), VHost);
 
 %% We can be recovering a transient queue if it crashed
-init(Q, Terms, MsgOnDiskFun, MsgIdxOnDiskFun) when ?is_amqqueue(Q) ->
+init1(Q, Terms, MsgOnDiskFun) when ?is_amqqueue(Q) ->
     QueueName = amqqueue:get_name(Q),
     IsDurable = amqqueue:is_durable(Q),
     {PRef, RecoveryTerms} = process_recovery_terms(Terms),
@@ -451,7 +450,7 @@ init(Q, Terms, MsgOnDiskFun, MsgIdxOnDiskFun) when ?is_amqqueue(Q) ->
           rabbit_vhost_msg_store:successfully_recovered_state(
               VHost,
               ?PERSISTENT_MSG_STORE),
-          ContainsCheckFun, MsgIdxOnDiskFun),
+          ContainsCheckFun),
     StoreState = rabbit_classic_queue_store_v2:init(QueueName),
     init(IsDurable, IndexState, StoreState,
          DeltaCount, DeltaBytes, RecoveryTerms,
@@ -714,12 +713,28 @@ sync(State = #vqstate { index_state = IndexState0,
                         store_state = StoreState0,
                         unconfirmed_simple = UCS,
                         confirmed   = C }) ->
-    IndexState = rabbit_classic_queue_index_v2:sync(IndexState0),
+    {MsgIdSet, IndexState} = rabbit_classic_queue_index_v2:sync(IndexState0),
     StoreState = rabbit_classic_queue_store_v2:sync(StoreState0),
-    State #vqstate { index_state = IndexState,
-                     store_state = StoreState,
-                     unconfirmed_simple = sets:new([{version,2}]),
-                     confirmed   = sets:union(C, UCS) }.
+    State1 = State #vqstate { index_state = IndexState,
+                              store_state = StoreState,
+                              unconfirmed_simple = sets:new([{version,2}]),
+                              confirmed   = sets:union(C, UCS) },
+    index_synced(MsgIdSet, State1).
+
+index_synced(MsgIdSet, State = #vqstate{
+        msgs_on_disk        = MOD,
+        msg_indices_on_disk = MIOD,
+        unconfirmed         = UC }) ->
+    case sets:is_empty(MsgIdSet) of
+        true ->
+            State;
+        false ->
+            Confirmed = sets:intersection(UC, MsgIdSet),
+            record_confirms(sets:intersection(MsgIdSet, MOD),
+                            State #vqstate {
+                              msg_indices_on_disk =
+                                  sets:union(MIOD, Confirmed) })
+    end.
 
 resume(State) -> a(timeout(State)).
 
@@ -1903,19 +1918,6 @@ msgs_written_to_disk(Callback, MsgIdSet, written) ->
                                            sets:union(MOD, Confirmed) })
              end).
 
-%% @todo Having to call run_backing_queue is probably reducing performance...
-msg_indices_written_to_disk(Callback, MsgIdSet) ->
-    Callback(?MODULE,
-             fun (?MODULE, State = #vqstate { msgs_on_disk        = MOD,
-                                              msg_indices_on_disk = MIOD,
-                                              unconfirmed         = UC }) ->
-                     Confirmed = sets:intersection(UC, MsgIdSet),
-                     record_confirms(sets:intersection(MsgIdSet, MOD),
-                                     State #vqstate {
-                                       msg_indices_on_disk =
-                                           sets:union(MIOD, Confirmed) })
-             end).
-
 %%----------------------------------------------------------------------------
 %% Internal plumbing for requeue
 %%----------------------------------------------------------------------------
diff --git a/deps/rabbit/test/backing_queue_SUITE.erl b/deps/rabbit/test/backing_queue_SUITE.erl
@@ -844,7 +844,7 @@ bq_queue_index1(_Config) ->
                       Qi13
               end,
               {_DeletedSegments, Qi16} = IndexMod:ack(SeqIdsB, Qi15),
-              Qi17 = IndexMod:sync(Qi16),
+              {_Confirms, Qi17} = IndexMod:sync(Qi16),
               %% Everything will have gone now because #pubs == #acks
               {NextSeqIdB, NextSeqIdB, Qi18} = IndexMod:bounds(Qi17, NextSeqIdB),
               %% should get length back as 0 because all persistent
@@ -865,7 +865,7 @@ bq_queue_index1(_Config) ->
                   _ -> Qi1
               end,
               {_DeletedSegments, Qi3} = IndexMod:ack(SeqIdsC, Qi2),
-              Qi4 = IndexMod:sync(Qi3),
+              {_Confirms, Qi4} = IndexMod:sync(Qi3),
               {Qi5, _SeqIdsMsgIdsC1} = queue_index_publish([SegmentSize],
                                                            false, Qi4),
               Qi5
@@ -883,7 +883,8 @@ bq_queue_index1(_Config) ->
               {Qi3, _SeqIdsMsgIdsC3} = queue_index_publish([SegmentSize],
                                                            false, Qi2),
               {_DeletedSegments, Qi4} = IndexMod:ack(SeqIdsC, Qi3),
-              IndexMod:sync(Qi4)
+              {_Confirms, Qi5} = IndexMod:sync(Qi4),
+              Qi5
       end),
 
     %% c) just fill up several segments of all pubs, then +acks
@@ -896,7 +897,8 @@ bq_queue_index1(_Config) ->
                   _ -> Qi1
               end,
               {_DeletedSegments, Qi3} = IndexMod:ack(SeqIdsD, Qi2),
-              IndexMod:sync(Qi3)
+              {_Confirms, Qi4} = IndexMod:sync(Qi3),
+              Qi4
       end),
 
     %% d) get messages in all states to a segment, then flush, then do
@@ -910,7 +912,7 @@ bq_queue_index1(_Config) ->
                   _ -> Qi1
               end,
               {_DeletedSegments3, Qi3} = IndexMod:ack([0], Qi2),
-              Qi4 = IndexMod:sync(Qi3),
+              {_Confirms, Qi4} = IndexMod:sync(Qi3),
               {Qi5, [Eight,Six|_]} = queue_index_publish([3,6,8], false, Qi4),
               Qi6 = case IndexMod of
                   rabbit_queue_index -> IndexMod:deliver([2,3,5,6], Qi5);
@@ -1641,8 +1643,7 @@ init_test_queue(QName) ->
             QName, [], false,
             fun (MsgId) ->
                     rabbit_msg_store:contains(MsgId, PersistentClient)
-            end,
-            fun nop/1),
+            end),
     ok = rabbit_msg_store:client_delete_and_terminate(PersistentClient),
     Res.
 
@@ -1702,9 +1703,6 @@ queue_index_publish(SeqIds, Persistent, Qi) ->
     ok = rabbit_msg_store:client_delete_and_terminate(MSCState),
     {A, B}.
 
-nop(_) -> ok.
-nop(_, _) -> ok.
-
 msg_store_client_init(MsgStore, Ref) ->
     rabbit_vhost_msg_store:client_init(?VHOST, MsgStore, Ref,  undefined).
 
@@ -1714,7 +1712,7 @@ variable_queue_init(Q, Recover) ->
              true  -> non_clean_shutdown;
              false -> new;
              Terms -> Terms
-         end, fun nop/2, fun nop/1).
+         end, fun(_, _) -> ok end).
 
 variable_queue_read_terms(QName) ->
     #resource { kind = queue,
