QQ: Use new log_ext effect

This offloads the work of reading messages from on-disk segments
to the interacting process rather than doing this blocking, performance
affecting work in the ra server process.

QQ: ensure opened segments are closed after some time of inactivity

Processes that havea received messages that had to be read from disks
may keep a segment open indefinitely. This introduces a timer which
after some time of inactivity will close all opened segments to ensure
file desciptors are not kept open indefinitely.

Author: kjnilsson

MODULE.bazel

@@ -253,8 +253,8 @@ erlang_package.hex_package(
     name = "ra",
     build_file = "@rabbitmq-server//bazel:BUILD.ra",
     pkg = "ra",
-    sha256 = "1bbb663e0375543f61700793d90ad1ba800be653f0446b135792b4f187e925fe",
-    version = "2.16.0-pre.3",
+    sha256 = "a7eae50b0c1c0be4daf9b7ee97be796e6fda372ea6b2047c3aeac89cdc2011df",
+    version = "2.16.0-pre.10",
 )
 
 erlang_package.git_package(

deps/rabbit/src/rabbit_channel.erl

@@ -184,6 +184,7 @@
          messages_uncommitted,
          acks_uncommitted,
          pending_raft_commands,
+         cached_segments,
          prefetch_count,
          state,
          garbage_collection]).
@@ -570,8 +571,6 @@ prioritise_cast(Msg, _Len, _State) ->
     case Msg of
         {confirm,            _MsgSeqNos, _QPid} -> 5;
         {reject_publish,     _MsgSeqNos, _QPid} -> 5;
-        {queue_event, _, {confirm,            _MsgSeqNos, _QPid}} -> 5;
-        {queue_event, _, {reject_publish,     _MsgSeqNos, _QPid}} -> 5;
         _                                       -> 0
     end.
 
@@ -661,10 +660,14 @@ handle_cast(terminate, State = #ch{cfg = #conf{writer_pid = WriterPid}}) ->
     ok = rabbit_writer:flush(WriterPid),
     {stop, normal, State};
 
-handle_cast({command, #'basic.consume_ok'{consumer_tag = CTag} = Msg}, State) ->
+handle_cast({command, #'basic.consume_ok'{consumer_tag = CTag} = Msg},
+            #ch{consumer_mapping = CMap} = State)
+  when is_map_key(CTag, CMap) ->
     ok = send(Msg, State),
     noreply(consumer_monitor(CTag, State));
-
+handle_cast({command, #'basic.consume_ok'{}}, State) ->
+    %% a consumer was not found so just ignore this
+    noreply(State);
 handle_cast({command, Msg}, State) ->
     ok = send(Msg, State),
     noreply(State);
@@ -2281,6 +2284,8 @@ i(acks_uncommitted,        #ch{tx = {_Msgs, Acks}})       -> ack_len(Acks);
 i(acks_uncommitted,        #ch{})                         -> 0;
 i(pending_raft_commands,   #ch{queue_states = QS}) ->
     pending_raft_commands(QS);
+i(cached_segments,   #ch{queue_states = QS}) ->
+    cached_segments(QS);
 i(state,                   #ch{cfg = #conf{state = running}}) -> credit_flow:state();
 i(state,                   #ch{cfg = #conf{state = State}}) -> State;
 i(prefetch_count,          #ch{cfg = #conf{consumer_prefetch = C}})    -> C;
@@ -2309,6 +2314,17 @@ pending_raft_commands(QStates) ->
           end,
     rabbit_queue_type:fold_state(Fun, 0, QStates).
 
+cached_segments(QStates) ->
+    Fun = fun(_, V, Acc) ->
+                  case rabbit_queue_type:state_info(V) of
+                      #{cached_segments := P} ->
+                          Acc + P;
+                      _ ->
+                          Acc
+                  end
+          end,
+    rabbit_queue_type:fold_state(Fun, 0, QStates).
+
 name(#ch{cfg = #conf{conn_name = ConnName, channel = Channel}}) ->
     list_to_binary(rabbit_misc:format("~ts (~tp)", [ConnName, Channel])).
 

deps/rabbit/src/rabbit_fifo.erl

@@ -89,7 +89,10 @@
          make_purge/0,
          make_purge_nodes/1,
          make_update_config/1,
-         make_garbage_collection/0
+         make_garbage_collection/0,
+
+         exec_read/3
+
         ]).
 
 -ifdef(TEST).
@@ -2076,30 +2079,27 @@ delivery_effect(ConsumerKey, [{MsgId, ?MSG(Idx,  Header)}],
     {CTag, CPid} = consumer_id(ConsumerKey, State),
     {send_msg, CPid, {delivery, CTag, [{MsgId, {Header, RawMsg}}]},
      ?DELIVERY_SEND_MSG_OPTS};
-delivery_effect(ConsumerKey, Msgs,
-                #?STATE{cfg = #cfg{resource = QR}} = State) ->
+delivery_effect(ConsumerKey, Msgs, #?STATE{} = State) ->
     {CTag, CPid} = consumer_id(ConsumerKey, State),
-    {RaftIdxs, Num} = lists:foldr(fun ({_, ?MSG(I, _)}, {Acc, N}) ->
-                                          {[I | Acc], N+1}
-                                  end, {[], 0}, Msgs),
-    {log, RaftIdxs,
-     fun (Commands)
-           when length(Commands) < Num ->
-             %% the mandatory length/1 guard is a bit :(
-             rabbit_log:info("~ts: requested read consumer tag '~ts' of ~b "
-                             "indexes ~w but only ~b were returned. "
-                             "This is most likely a stale read request "
-                             "and can be ignored",
-                             [rabbit_misc:rs(QR), CTag, Num, RaftIdxs,
-                              length(Commands)]),
-             [];
-         (Commands) ->
-             DelMsgs = lists:zipwith(
-                         fun (Cmd, {MsgId, ?MSG(_Idx,  Header)}) ->
-                                 {MsgId, {Header, get_msg(Cmd)}}
-                         end, Commands, Msgs),
-             [{send_msg, CPid, {delivery, CTag, DelMsgs},
-               ?DELIVERY_SEND_MSG_OPTS}]
+    {RaftIdxs, _Num} = lists:foldr(fun ({_, ?MSG(I, _)}, {Acc, N}) ->
+                                           {[I | Acc], N+1}
+                                   end, {[], 0}, Msgs),
+    {log_ext, RaftIdxs,
+     fun (ReadPlan) ->
+             case node(CPid) == node() of
+                 true ->
+                     [{send_msg, CPid, {delivery, CTag, ReadPlan, Msgs},
+                       ?DELIVERY_SEND_MSG_OPTS}];
+                 false ->
+                     %% if we got there we need to read the data on this node
+                     %% and send it to the consumer pid as it isn't availble
+                     %% locally
+                     {DelMsgs, Flru} = exec_read(undefined, ReadPlan, Msgs),
+                     %% we need to evict all cached items here
+                     _ = ra_flru:evict_all(Flru),
+                     [{send_msg, CPid, {delivery, CTag, DelMsgs},
+                       ?DELIVERY_SEND_MSG_OPTS}]
+             end
      end,
      {local, node(CPid)}}.
 
@@ -3014,3 +3014,21 @@ incr_msg(Msg0, DelFailed, Anns) ->
         false ->
             Msg2
     end.
+
+exec_read(Flru0, ReadPlan, Msgs) ->
+    try ra_log_read_plan:execute(ReadPlan, Flru0) of
+        {Entries, Flru} ->
+            %% return a list in original order
+            {lists:map(fun ({MsgId, ?MSG(Idx,  Header)}) ->
+                               Cmd = maps:get(Idx, Entries),
+                               {MsgId, {Header, get_msg(Cmd)}}
+                       end, Msgs), Flru}
+    catch exit:{missing_key, _}
+            when Flru0 =/= undefined ->
+              %% this segment has most likely been appended to but the
+              %% cached index doesn't know about new items and need to be
+              %% re-generated
+              _ = ra_flru:evict_all(Flru0),
+              %% retry without segment cache
+              exec_read(undefined, ReadPlan, Msgs)
+    end.

deps/rabbit/src/rabbit_fifo_client.erl

@@ -30,6 +30,7 @@
          purge/1,
          update_machine_state/2,
          pending_size/1,
+         num_cached_segments/1,
          stat/1,
          stat/2,
          query_single_active_consumer/1,
@@ -40,8 +41,12 @@
 -define(TIMER_TIME, 10000).
 -define(COMMAND_TIMEOUT, 30000).
 -define(UNLIMITED_PREFETCH_COUNT, 2000). %% something large for ra
+%% controls the timer for closing cached segments
+-define(CACHE_SEG_TIMEOUT, 5000).
 
 -type seq() :: non_neg_integer().
+-type milliseconds() :: non_neg_integer().
+
 
 -record(consumer, {key :: rabbit_fifo:consumer_key(),
                    % status = up :: up | cancelled,
@@ -69,7 +74,11 @@
                 pending = #{} :: #{seq() =>
                                    {term(), rabbit_fifo:command()}},
                 consumers = #{} :: #{rabbit_types:ctag() => #consumer{}},
-                timer_state :: term()
+                timer_state :: term(),
+                cached_segments :: undefined |
+                                  {undefined | reference(),
+                                   LastSeenMs :: milliseconds(),
+                                   ra_flru:state()}
                }).
 
 -opaque state() :: #state{}.
@@ -132,9 +141,15 @@ enqueue(QName, Correlation, Msg,
             %% it is safe to reject the message as we never attempted
             %% to send it
             {reject_publish, State0};
+        {error, {shutdown, delete}} ->
+            rabbit_log:debug("~ts: QQ ~ts tried to register enqueuer during delete shutdown",
+                             [?MODULE, rabbit_misc:rs(QName)]),
+            {reject_publish, State0};
         {timeout, _} ->
             {reject_publish, State0};
         Err ->
+            rabbit_log:debug("~ts: QQ ~ts error when registering enqueuer ~p",
+                             [?MODULE, rabbit_misc:rs(QName), Err]),
             exit(Err)
     end;
 enqueue(_QName, _Correlation, _Msg,
@@ -167,7 +182,7 @@ enqueue(QName, Correlation, Msg,
 %% @param QueueName Name of the queue.
 %% @param Msg an arbitrary erlang term representing the message.
 %% @param State the current {@module} state.
-%% @returns
+%% @return's
 %% `{ok, State, Actions}' if the command was successfully sent.
 %% {@module} assigns a sequence number to every raft command it issues. The
 %% SequenceNumber can be correlated to the applied sequence numbers returned
@@ -510,6 +525,15 @@ purge(Server) ->
 pending_size(#state{pending = Pend}) ->
     maps:size(Pend).
 
+-spec num_cached_segments(state()) -> non_neg_integer().
+num_cached_segments(#state{cached_segments = CachedSegments}) ->
+    case CachedSegments of
+        undefined ->
+            0;
+        {_, _, Cached} ->
+            ra_flru:size(Cached)
+    end.
+
 -spec stat(ra:server_id()) ->
     {ok, non_neg_integer(), non_neg_integer()}
     | {error | timeout, term()}.
@@ -633,7 +657,8 @@ handle_ra_event(QName, From, {applied, Seqs},
         _ ->
             {ok, State1, Actions}
     end;
-handle_ra_event(QName, From, {machine, {delivery, _ConsumerTag, _} = Del}, State0) ->
+handle_ra_event(QName, From, {machine, Del}, State0)
+      when element(1, Del) == delivery ->
     handle_delivery(QName, From, Del, State0);
 handle_ra_event(_QName, _From, {machine, Action}, State)
   when element(1, Action) =:= credit_reply orelse
@@ -643,28 +668,31 @@ handle_ra_event(_QName, _, {machine, {queue_status, Status}},
                 #state{} = State) ->
     %% just set the queue status
     {ok, State#state{queue_status = Status}, []};
-handle_ra_event(_QName, Leader, {machine, leader_change},
+handle_ra_event(QName, Leader, {machine, leader_change},
                 #state{leader = OldLeader,
                        pending = Pending} = State0) ->
     %% we need to update leader
     %% and resend any pending commands
-    rabbit_log:debug("~ts: Detected QQ leader change from ~w to ~w, "
+    rabbit_log:debug("~ts: ~s Detected QQ leader change from ~w to ~w, "
                      "resending ~b pending commands",
-                     [?MODULE, OldLeader, Leader, maps:size(Pending)]),
+                     [rabbit_misc:rs(QName), ?MODULE, OldLeader,
+                      Leader, maps:size(Pending)]),
     State = resend_all_pending(State0#state{leader = Leader}),
     {ok, State, []};
 handle_ra_event(_QName, _From, {rejected, {not_leader, Leader, _Seq}},
                 #state{leader = Leader} = State) ->
     {ok, State, []};
-handle_ra_event(_QName, _From, {rejected, {not_leader, Leader, _Seq}},
+handle_ra_event(QName, _From, {rejected, {not_leader, Leader, _Seq}},
                 #state{leader = OldLeader,
                        pending = Pending} = State0) ->
-    rabbit_log:debug("~ts: Detected QQ leader change (rejection) from ~w to ~w, "
+    rabbit_log:debug("~ts: ~s Detected QQ leader change (rejection) from ~w to ~w, "
                      "resending ~b pending commands",
-                     [?MODULE, OldLeader, Leader, maps:size(Pending)]),
+                     [rabbit_misc:rs(QName), ?MODULE, OldLeader,
+                      Leader, maps:size(Pending)]),
     State = resend_all_pending(State0#state{leader = Leader}),
     {ok, cancel_timer(State), []};
-handle_ra_event(_QName, _From, {rejected, {not_leader, _UndefinedMaybe, _Seq}}, State0) ->
+handle_ra_event(_QName, _From,
+                {rejected, {not_leader, _UndefinedMaybe, _Seq}}, State0) ->
     % TODO: how should these be handled? re-sent on timer or try random
     {ok, State0, []};
 handle_ra_event(QName, _, timeout, #state{cfg = #cfg{servers = Servers}} = State0) ->
@@ -676,6 +704,30 @@ handle_ra_event(QName, _, timeout, #state{cfg = #cfg{servers = Servers}} = State
             State = resend_all_pending(State0#state{leader = Leader}),
             {ok, State, []}
     end;
+handle_ra_event(QName, Leader, close_cached_segments,
+                #state{cached_segments = CachedSegments} = State) ->
+    {ok,
+     case CachedSegments of
+         undefined ->
+             %% timer didn't get cancelled so just ignore this
+             State;
+         {_TRef, Last, Cache} ->
+             case now_ms() > Last + ?CACHE_SEG_TIMEOUT of
+                 true ->
+                     rabbit_log:debug("~ts: closing_cached_segments",
+                                      [rabbit_misc:rs(QName)]),
+                     %% its been long enough, evict all
+                     _ = ra_flru:evict_all(Cache),
+                     State#state{cached_segments = undefined};
+                 false ->
+                     %% set another timer
+                     Ref = erlang:send_after(?CACHE_SEG_TIMEOUT, self(),
+                                             {'$gen_cast',
+                                              {queue_event, QName,
+                                               {Leader, close_cached_segments}}}),
+                     State#state{cached_segments = {Ref, Last, Cache}}
+             end
+     end, []};
 handle_ra_event(_QName, _Leader, {machine, eol}, State) ->
     {eol, [{unblock, cluster_name(State)}]}.
 
@@ -839,7 +891,39 @@ handle_delivery(_QName, _Leader, {delivery, Tag, [_ | _] = IdMsgs},
     %% we should return all messages.
     MsgIntIds = [Id || {Id, _} <- IdMsgs],
     {State1, Deliveries} = return(Tag, MsgIntIds, State0),
-    {ok, State1, Deliveries}.
+    {ok, State1, Deliveries};
+handle_delivery(QName, Leader, {delivery, Tag, ReadPlan, Msgs},
+                #state{cached_segments = CachedSegments} = State) ->
+    {TRef, Cached0} = case CachedSegments of
+                          undefined ->
+                              {undefined, undefined};
+                          {R, _, C} ->
+                              {R, C}
+                      end,
+    {MsgIds, Cached1} = rabbit_fifo:exec_read(Cached0, ReadPlan, Msgs),
+    %% if there are cached segments after a read and there
+    %% is no current timer set, set a timer
+    %% send a message to evict cache after some time
+    Cached = case ra_flru:size(Cached1) > 0 of
+                 true when TRef == undefined ->
+                     Ref = erlang:send_after(?CACHE_SEG_TIMEOUT, self(),
+                                             {'$gen_cast',
+                                              {queue_event, QName,
+                                               {Leader, close_cached_segments}}}),
+                     {Ref, now_ms(), Cached1};
+                 true ->
+                     {TRef, now_ms(), Cached1};
+                 false ->
+                     if TRef =/= undefined ->
+                            _ = erlang:cancel_timer(TRef, [{async, true}]),
+                            ok;
+                        true  ->
+                            ok
+                     end,
+                     undefined
+             end,
+    handle_delivery(QName, Leader, {delivery, Tag, MsgIds},
+                    State#state{cached_segments = Cached}).
 
 transform_msgs(QName, QRef, Msgs) ->
     lists:map(
@@ -1008,3 +1092,6 @@ send_pending(Cid, #state{unsent_commands = Unsent} = State0) ->
                                               normal, S0)
                          end, State0, Commands),
     State1#state{unsent_commands = maps:remove(Cid, Unsent)}.
+
+now_ms() ->
+    erlang:system_time(millisecond).

deps/rabbit/src/rabbit_fifo_dlx.erl

@@ -154,6 +154,8 @@ discard(Msgs, Reason, undefined, State) ->
               [Reason, rabbit_quorum_queue, disabled, length(Msgs)]}]};
 discard(Msgs0, Reason, {at_most_once, {Mod, Fun, Args}}, State) ->
     Idxs = [I || ?MSG(I, _) <- Msgs0],
+    %% TODO: this could be turned into a log_ext effect instead to avoid
+    %% reading from disk inside the qq process
     Effect = {log, Idxs,
               fun (Log) ->
                       Lookup = maps:from_list(lists:zip(Idxs, Log)),

deps/rabbit/src/rabbit_quorum_queue.erl

@@ -1094,7 +1094,8 @@ deliver(QSs, Msg0, Options) ->
 
 
 state_info(S) ->
-    #{pending_raft_commands => rabbit_fifo_client:pending_size(S)}.
+    #{pending_raft_commands => rabbit_fifo_client:pending_size(S),
+      cached_segments => rabbit_fifo_client:num_cached_segments(S)}.
 
 -spec infos(rabbit_types:r('queue')) -> rabbit_types:infos().
 infos(QName) ->