Reduce ETS copy overhead when delivering to target queues

 ## What?
This commit avoids copying the full amqqueue record from ETS per incoming message
and target queue.
The amqqueue record contains 21 elements and for some queue types,
especially streams, some elements are themselves nested terms.

 ## How?

In Khepri, use a new `rabbit_khepri_queue_target` projection which
contains a subset of the full amqqueue record.

This way all relevant information to deliver to a target queue can be
looked up in a single ets:lookup call.

Alternative approaches are described in erlang/otp#10211

 ## Benchmark
Fanout to 3 classic queues.
Variation: Fanout to 3 streams since streams have a larger amqqueue record.

Start broker:
```
make run-broker TEST_TMPDIR="$HOME/scratch/rabbit/test" \
    FULL=1 \
    RABBITMQ_SERVER_ADDITIONAL_ERL_ARGS="+S 5" \
    RABBITMQ_CONFIG_FILE="$HOME/scratch/rabbit/high-credit.config" \
    PLUGINS="rabbitmq_management"
```

`high-credit.config` contains:
```
[
 {rabbit, [
  %% Maximum incoming-window of AMQP 1.0 session.
  %% Default: 400
  {max_incoming_window, 5000},

  %% Maximum link-credit RabbitMQ grants to AMQP 1.0 sender.
  %% Default: 128
  {max_link_credit, 2000},

  %% Maximum link-credit RabbitMQ AMQP 1.0 session grants to sending queue.
  %% Default: 256
  {max_queue_credit, 5000},

  {loopback_users, []}
 ]},

 {rabbitmq_management_agent, [
  {disable_metrics_collector, true}
 ]}
].
```

Create the 3 classic queues and bindings to the fanout exchange:
```
deps/rabbitmq_management/bin/rabbitmqadmin declare queue queue_type=classic durable=true name=q1 && \
    deps/rabbitmq_management/bin/rabbitmqadmin declare queue queue_type=classic durable=true name=q2 && \
    deps/rabbitmq_management/bin/rabbitmqadmin declare queue queue_type=classic durable=true name=q3 && \
    deps/rabbitmq_management/bin/rabbitmqadmin declare binding source=amq.fanout destination=q1 && \
    deps/rabbitmq_management/bin/rabbitmqadmin declare binding source=amq.fanout destination=q2 && \
    deps/rabbitmq_management/bin/rabbitmqadmin declare binding source=amq.fanout destination=q3
```

Start the client:
```
quiver-arrow send //host.docker.internal//exchanges/amq.fanout --summary --count 1m --body-size 4
```

On main branch the rate is about ~67k msgs/s.
With this PR the rate is about ~70k msgs/s.

When faning out to 3 classic queues, the throughput increase is about 4%.
When faning out to 3 streams, the throughput increase is about 9%.

Author: ansd

deps/rabbit/src/amqqueue.erl

@@ -145,6 +145,8 @@
                           type_state :: #{}
                          }.
 
+-type target() :: #queue_target{}.
+
 -type ra_server_id() :: {Name :: atom(), Node :: node()}.
 
 -type amqqueue_pattern() :: amqqueue_v2_pattern().
@@ -175,6 +177,7 @@
               amqqueue_v2/0,
               amqqueue_pattern/0,
               amqqueue_v2_pattern/0,
+              target/0,
               ra_server_id/0]).
 
 -spec new(rabbit_amqqueue:name(),
@@ -418,9 +421,10 @@ set_operator_policy(#amqqueue{} = Queue, Policy) ->
 
 % name
 
--spec get_name(amqqueue()) -> rabbit_amqqueue:name().
+-spec get_name(amqqueue() | target()) -> rabbit_amqqueue:name().
 
-get_name(#amqqueue{name = Name}) -> Name.
+get_name(#amqqueue{name = Name}) -> Name;
+get_name(#queue_target{name = Name}) -> Name.
 
 -spec set_name(amqqueue(), rabbit_amqqueue:name()) -> amqqueue().
 
@@ -429,9 +433,10 @@ set_name(#amqqueue{} = Queue, Name) ->
 
 % pid
 
--spec get_pid(amqqueue_v2()) -> pid() | ra_server_id() | none.
+-spec get_pid(amqqueue_v2() | target()) -> pid() | ra_server_id() | none.
 
-get_pid(#amqqueue{pid = Pid}) -> Pid.
+get_pid(#amqqueue{pid = Pid}) -> Pid;
+get_pid(#queue_target{pid = Pid}) -> Pid.
 
 -spec set_pid(amqqueue_v2(), pid() | ra_server_id() | none) -> amqqueue_v2().
 
@@ -488,9 +493,10 @@ set_state(#amqqueue{} = Queue, State) ->
 
 %% New in v2.
 
--spec get_type(amqqueue()) -> atom().
+-spec get_type(amqqueue() | target()) -> atom().
 
-get_type(#amqqueue{type = Type}) -> Type.
+get_type(#amqqueue{type = Type}) -> Type;
+get_type(#queue_target{type = Type}) -> Type.
 
 -spec get_vhost(amqqueue()) -> rabbit_types:vhost() | undefined.
 

deps/rabbit/src/rabbit_amqp_session.erl

@@ -2528,7 +2528,7 @@ incoming_link_transfer(
             QNames = rabbit_exchange:route(X, Mc2, #{return_binding_keys => true}),
             rabbit_trace:tap_in(Mc2, QNames, ConnName, ChannelNum, Username, Trace),
             Opts = #{correlation => {HandleInt, DeliveryId}},
-            Qs0 = rabbit_amqqueue:lookup_many(QNames),
+            Qs0 = rabbit_db_queue:get_targets(QNames),
             Qs = rabbit_amqqueue:prepend_extra_bcc(Qs0),
             Mc = ensure_mc_cluster_compat(Mc2),
             case rabbit_queue_type:deliver(Qs, Mc, Opts, QStates0) of
@@ -2674,8 +2674,8 @@ process_routing_confirm([], _SenderSettles = false, DeliveryId, U) ->
     Disposition = released(DeliveryId),
     {U, [Disposition]};
 process_routing_confirm([_|_] = Qs, SenderSettles, DeliveryId, U0) ->
-    QNames = rabbit_amqqueue:queue_names(Qs),
     false = maps:is_key(DeliveryId, U0),
+    QNames = rabbit_amqqueue:queue_names(Qs),
     Map = maps:from_keys(QNames, ok),
     U = U0#{DeliveryId => {Map, SenderSettles, false}},
     rabbit_global_counters:messages_routed(?PROTOCOL, map_size(Map)),

deps/rabbit/src/rabbit_amqqueue.erl

@@ -2051,65 +2051,70 @@ get_quorum_nodes(Q) ->
     end.
 
 -spec prepend_extra_bcc(Qs) ->
-    Qs when Qs :: [amqqueue:amqqueue() |
-                   {amqqueue:amqqueue(), route_infos()}].
+    Qs when Qs :: [amqqueue:amqqueue() | amqqueue:target() |
+                   {amqqueue:amqqueue() | amqqueue:target(), route_infos()}].
 prepend_extra_bcc([]) ->
     [];
 prepend_extra_bcc([Q0] = Qs) ->
     Q = queue(Q0),
-    case amqqueue:get_options(Q) of
-        #{extra_bcc := BCCName} ->
-            case get_bcc_queue(Q, BCCName) of
+    case get_extra_bcc(Q) of
+        none ->
+            Qs;
+        Name ->
+            case lookup_extra_bcc(Q, Name) of
                 {ok, BCCQueue} ->
                     [BCCQueue | Qs];
                 {error, not_found} ->
                     Qs
-            end;
-        _ ->
-            Qs
+            end
     end;
 prepend_extra_bcc(Qs) ->
-    BCCQueues =
-        lists:filtermap(
-          fun(Q0) ->
-                  Q = queue(Q0),
-                  case amqqueue:get_options(Q) of
-                      #{extra_bcc := BCCName} ->
-                          case get_bcc_queue(Q, BCCName) of
-                              {ok, BCCQ} ->
-                                  {true, BCCQ};
-                              {error, not_found} ->
-                                  false
-                          end;
-                      _ ->
-                          false
-                  end
-          end, Qs),
-    lists:usort(BCCQueues) ++ Qs.
+    ExtraQs = lists:filtermap(
+                fun(Q0) ->
+                        Q = queue(Q0),
+                        case get_extra_bcc(Q) of
+                            none ->
+                                false;
+                            Name ->
+                                case lookup_extra_bcc(Q, Name) of
+                                    {ok, BCCQ} ->
+                                        {true, BCCQ};
+                                    {error, not_found} ->
+                                        false
+                                end
+                        end
+                end, Qs),
+    lists:usort(ExtraQs) ++ Qs.
 
 -spec queue(Q | {Q, route_infos()}) ->
-    Q when Q :: amqqueue:amqqueue().
-queue(Q)
-  when ?is_amqqueue(Q) ->
+    Q when Q :: amqqueue:amqqueue() | amqqueue:target().
+queue({Q, RouteInfos}) when is_map(RouteInfos) ->
     Q;
-queue({Q, RouteInfos})
-  when ?is_amqqueue(Q) andalso is_map(RouteInfos) ->
+queue(Q) ->
     Q.
 
 -spec queue_names([Q | {Q, route_infos()}]) ->
-    [name()] when Q :: amqqueue:amqqueue().
-queue_names(Queues)
-  when is_list(Queues) ->
-    lists:map(fun(Q) when ?is_amqqueue(Q) ->
+    [name()] when Q :: amqqueue:amqqueue() | amqqueue:target().
+queue_names(Queues) ->
+    lists:map(fun({Q, RouteInfos}) when is_map(RouteInfos) ->
                       amqqueue:get_name(Q);
-                 ({Q, RouteInfos})
-                   when ?is_amqqueue(Q) andalso is_map(RouteInfos) ->
+                 (Q) ->
                       amqqueue:get_name(Q)
               end, Queues).
 
--spec get_bcc_queue(amqqueue:amqqueue(), binary()) ->
+get_extra_bcc(Q) when ?is_amqqueue(Q) ->
+    case amqqueue:get_options(Q) of
+        #{extra_bcc := Name} ->
+            Name;
+        _ ->
+            none
+    end;
+get_extra_bcc(#queue_target{extra_bcc = Name}) ->
+    Name.
+
+-spec lookup_extra_bcc(amqqueue:amqqueue() | amqqueue:target(), binary()) ->
     {ok, amqqueue:amqqueue()} | {error, not_found}.
-get_bcc_queue(Q, BCCName) ->
+lookup_extra_bcc(Q, BCCName) ->
     #resource{virtual_host = VHost} = amqqueue:get_name(Q),
     BCCQueueName = rabbit_misc:r(VHost, queue, BCCName),
     lookup(BCCQueueName).

deps/rabbit/src/rabbit_channel.erl

@@ -1222,7 +1222,7 @@ handle_method(#'basic.publish'{exchange    = ExchangeNameBin,
             check_user_id_header(Message0, User),
             Message = rabbit_msg_interceptor:intercept_incoming(Message0, MsgIcptCtx),
             QNames = rabbit_exchange:route(Exchange, Message, #{return_binding_keys => true}),
-            Queues = rabbit_amqqueue:lookup_many(QNames),
+            Queues = rabbit_db_queue:get_targets(QNames),
             rabbit_trace:tap_in(Message, QNames, ConnName, ChannelNum,
                                 Username, TraceState),
             %% TODO: call delivery_to_queues with plain args

deps/rabbit/src/rabbit_classic_queue.erl

@@ -287,16 +287,17 @@ format(Q, _Ctx) when ?is_amqqueue(Q) ->
      {state, State},
      {node, node(amqqueue:get_pid(Q))}].
 
--spec init(amqqueue:amqqueue()) -> {ok, state()}.
-init(Q) when ?amqqueue_is_classic(Q) ->
+-spec init(amqqueue:amqqueue() | amqqueue:target()) ->
+    {ok, state()}.
+init(Q) ->
     {ok, #?STATE{pid = amqqueue:get_pid(Q)}}.
 
 -spec close(state()) -> ok.
 close(_State) ->
     ok.
 
--spec update(amqqueue:amqqueue(), state()) -> state().
-update(Q, #?STATE{pid = Pid} = State) when ?amqqueue_is_classic(Q) ->
+-spec update(amqqueue:amqqueue() | amqqueue:target(), state()) -> state().
+update(Q, #?STATE{pid = Pid} = State) ->
     case amqqueue:get_pid(Q) of
         Pid ->
             State;
@@ -473,7 +474,7 @@ settlement_action(Type, QRef, MsgSeqs, Acc) ->
 
 supports_stateful_delivery() -> true.
 
--spec deliver([{amqqueue:amqqueue(), state()}],
+-spec deliver([{amqqueue:amqqueue() | amqqueue:target(), state()}],
               Delivery :: mc:state(),
               rabbit_queue_type:delivery_options()) ->
     {[{amqqueue:amqqueue(), state()}], rabbit_queue_type:actions()}.

deps/rabbit/src/rabbit_db_queue.erl

@@ -19,6 +19,7 @@
 -export([
          get/1,
          get_many/1,
+         get_targets/1,
          get_all/0,
          get_all/1,
          get_all_by_type/1,
@@ -85,6 +86,7 @@
 -define(MNESIA_DURABLE_TABLE, rabbit_durable_queue).
 
 -define(KHEPRI_PROJECTION, rabbit_khepri_queue).
+-define(KHEPRI_TARGET_PROJECTION, rabbit_khepri_queue_target).
 
 %% -------------------------------------------------------------------
 %% get_all().
@@ -468,6 +470,21 @@ internal_delete_in_mnesia(QueueName, OnlyDurable, Reason) ->
     %% after the transaction.
     rabbit_db_binding:delete_for_destination_in_mnesia(QueueName, OnlyDurable).
 
+%% -------------------------------------------------------------------
+%% get_targets().
+%% -------------------------------------------------------------------
+
+%% Queue target optimisation is only available in Khepri.
+%% Mnesia falls back looking up the full amqqueue record.
+-spec get_targets(rabbit_exchange:route_return()) ->
+    [amqqueue:target() | amqqueue:amqqueue() |
+     {amqqueue:target() | amqqueue:amqqueue(), rabbit_exchange:route_infos()}].
+get_targets(Names) ->
+    rabbit_khepri:handle_fallback(
+      #{mnesia => fun() -> get_many_in_ets(?MNESIA_TABLE, Names) end,
+        khepri => fun() -> get_many_in_khepri(?KHEPRI_TARGET_PROJECTION, Names) end
+       }).
+
 %% -------------------------------------------------------------------
 %% get_many().
 %% -------------------------------------------------------------------
@@ -477,12 +494,12 @@ internal_delete_in_mnesia(QueueName, OnlyDurable, Reason) ->
 get_many(Names) when is_list(Names) ->
     rabbit_khepri:handle_fallback(
       #{mnesia => fun() -> get_many_in_ets(?MNESIA_TABLE, Names) end,
-        khepri => fun() -> get_many_in_khepri(Names) end
+        khepri => fun() -> get_many_in_khepri(?KHEPRI_PROJECTION, Names) end
        }).
 
-get_many_in_khepri(Names) ->
+get_many_in_khepri(Table, Names) ->
     try
-        get_many_in_ets(?KHEPRI_PROJECTION, Names)
+        get_many_in_ets(Table, Names)
     catch
         error:badarg ->
             []

deps/rabbit/src/rabbit_dead_letter.erl

@@ -44,7 +44,7 @@ publish(Msg0, Reason, #exchange{name = XName} = DLX, RK,
     Routed0 = rabbit_exchange:route(DLX, DLMsg, #{return_binding_keys => true}),
     {Cycles, Routed} = detect_cycles(Reason, DLMsg, Routed0),
     lists:foreach(fun log_cycle_once/1, Cycles),
-    Qs0 = rabbit_amqqueue:lookup_many(Routed),
+    Qs0 = rabbit_db_queue:get_targets(Routed),
     Qs = rabbit_amqqueue:prepend_extra_bcc(Qs0),
     _ = rabbit_queue_type:deliver(Qs, DLMsg, #{}, stateless),
     ok.

deps/rabbit/src/rabbit_fifo_dlx_worker.erl

@@ -342,7 +342,7 @@ forward(ConsumedMsg, ConsumedMsgId, ConsumedQRef, DLX, Reason,
                 {Cycles, RouteToQs1} = rabbit_dead_letter:detect_cycles(
                                          Reason, Msg, RouteToQs0),
                 State1 = log_cycles(Cycles, [RKey], State0),
-                RouteToQs2 = rabbit_amqqueue:lookup_many(RouteToQs1),
+                RouteToQs2 = rabbit_db_queue:get_targets(RouteToQs1),
                 RouteToQs = rabbit_amqqueue:prepend_extra_bcc(RouteToQs2),
                 State2 = case RouteToQs of
                              [] ->
@@ -496,7 +496,7 @@ redeliver0(#pending{delivery = Msg0,
     %% queues that do not exist. Therefore, filter out non-existent target queues.
     RouteToQs0 = queue_names(
                    rabbit_amqqueue:prepend_extra_bcc(
-                     rabbit_amqqueue:lookup_many(
+                     rabbit_db_queue:get_targets(
                        rabbit_exchange:route(DLX, Msg)))),
     case {RouteToQs0, Settled} of
         {[], [_|_]} ->
@@ -529,7 +529,10 @@ redeliver0(#pending{delivery = Msg0,
                                          rejected = []},
                     State = State0#state{pendings = maps:update(OutSeq, Pend, Pendings)},
                     Options = #{correlation => OutSeq},
-                    deliver_to_queues(Msg, Options, rabbit_amqqueue:lookup_many(RouteToQs), State)
+                    deliver_to_queues(Msg,
+                                      Options,
+                                      rabbit_db_queue:get_targets(RouteToQs),
+                                      State)
             end
     end.
 
@@ -569,8 +572,7 @@ cancel_timer(#state{timer = TRef} = State)
 cancel_timer(State) ->
     State.
 
-queue_names(Qs)
-  when is_list(Qs) ->
+queue_names(Qs) ->
     lists:map(fun amqqueue:get_name/1, Qs).
 
 format_status(#{state := #state{

deps/rabbit/src/rabbit_khepri.erl

@@ -1310,6 +1310,7 @@ delete_or_fail(Path) ->
 register_projections() ->
     RegFuns = [fun register_rabbit_exchange_projection/0,
                fun register_rabbit_queue_projection/0,
+               fun register_rabbit_queue_target_projection/0,
                fun register_rabbit_vhost_projection/0,
                fun register_rabbit_users_projection/0,
                fun register_rabbit_global_runtime_parameters_projection/0,
@@ -1351,7 +1352,26 @@ register_rabbit_queue_projection() ->
                     _VHost = ?KHEPRI_WILDCARD_STAR,
                     _Name = ?KHEPRI_WILDCARD_STAR),
     KeyPos = 2, %% #amqqueue.name
-    register_simple_projection(Name, PathPattern, KeyPos, true).
+    register_simple_projection(Name, PathPattern, KeyPos, false).
+
+register_rabbit_queue_target_projection() ->
+    PathPattern = rabbit_db_queue:khepri_queue_path(
+                    _VHost = ?KHEPRI_WILDCARD_STAR,
+                    _Name = ?KHEPRI_WILDCARD_STAR),
+    Fun = fun(_Path, Q) ->
+                  BCC = case amqqueue:get_options(Q) of
+                            #{extra_bcc := Name} -> Name;
+                            _ -> none
+                        end,
+                  #queue_target{name = amqqueue:get_name(Q),
+                                type = amqqueue:get_type(Q),
+                                pid = amqqueue:get_pid(Q),
+                                extra_bcc = BCC}
+          end,
+    Opts = #{keypos => #queue_target.name,
+             read_concurrency => true},
+    Projection = khepri_projection:new(rabbit_khepri_queue_target, Fun, Opts),
+    khepri:register_projection(?STORE_ID, PathPattern, Projection).
 
 register_rabbit_vhost_projection() ->
     Name = rabbit_khepri_vhost,