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_element call.

Alternative approaches are described in erlang/otp#10211

 ## Benchmark

Fanout to 3 streams

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 streams and bindings to the fanout exchange:
```
deps/rabbitmq_management/bin/rabbitmqadmin declare queue queue_type=stream durable=true name=ssssssssssssssssssssssssssssssssssssssssssssssssssssssssssss1 && \
    deps/rabbitmq_management/bin/rabbitmqadmin declare queue queue_type=stream durable=true name=ssssssssssssssssssssssssssssssssssssssssssssssssssssssssssss2 && \
    deps/rabbitmq_management/bin/rabbitmqadmin declare queue queue_type=stream durable=true name=ssssssssssssssssssssssssssssssssssssssssssssssssssssssssssss3 && \
    deps/rabbitmq_management/bin/rabbitmqadmin declare binding source=amq.fanout destination=ssssssssssssssssssssssssssssssssssssssssssssssssssssssssssss1 && \
    deps/rabbitmq_management/bin/rabbitmqadmin declare binding source=amq.fanout destination=ssssssssssssssssssssssssssssssssssssssssssssssssssssssssssss2 && \
    deps/rabbitmq_management/bin/rabbitmqadmin declare binding source=amq.fanout destination=ssssssssssssssssssssssssssssssssssssssssssssssssssssssssssss3

```

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

`main` branch:
```
Count ............................................. 1,000,000 messages
Duration ............................................... 16.3 seconds
Message rate ......................................... 61,237 messages/s
```

with this PR:
```
Count ............................................. 1,000,000 messages
Duration ............................................... 14.2 seconds
Message rate ......................................... 70,309 messages/s
```

Hence, this PR increases the throughput when sending to 3 streams via AMQP by ~14%.

Author: ansd

deps/rabbit/src/amqqueue.erl

@@ -14,6 +14,7 @@
          new/9,
          new_with_version/9,
          new_with_version/10,
+         new_target/4,
          fields/0,
          fields/1,
          field_vhost/0,
@@ -39,6 +40,7 @@
          % options
          get_options/1,
          set_options/2,
+         get_extra_bcc/1,
          % pid
          get_pid/1,
          set_pid/2,
@@ -77,7 +79,8 @@
          qnode/1,
          to_printable/1,
          to_printable/2,
-         macros/0]).
+         macros/0
+        ]).
 
 -define(record_version, amqqueue_v2).
 -define(is_backwards_compat_classic(T),
@@ -120,6 +123,17 @@
           type_state = #{} :: map() | ets:match_pattern()
          }).
 
+%% A subset of the amqqueue record containing just the necessary fields
+%% to deliver a message to a target queue.
+-record(queue_target,
+        {name :: rabbit_amqqueue:name(),
+         type :: rabbit_queue_type:queue_type(),
+         pid :: pid() | ra_server_id() | none,
+         extra_bcc :: rabbit_misc:resource_name() | none
+        }).
+
+-opaque target() :: #queue_target{}.
+
 -type amqqueue() :: amqqueue_v2().
 -type amqqueue_v2() :: #amqqueue{
                           name :: rabbit_amqqueue:name(),
@@ -175,6 +189,7 @@
               amqqueue_v2/0,
               amqqueue_pattern/0,
               amqqueue_v2_pattern/0,
+              target/0,
               ra_server_id/0]).
 
 -spec new(rabbit_amqqueue:name(),
@@ -328,6 +343,17 @@ new_with_version(?record_version,
               options         = Options,
               type            = ensure_type_compat(Type)}.
 
+-spec new_target(rabbit_amqqueue:name(),
+                 rabbit_queue_type:queue_type(),
+                 pid() | ra_server_id() | none,
+                 rabbit_misc:resource_name() | none) ->
+    target().
+new_target(Name, Type, Pid, ExtraBcc) ->
+    #queue_target{name = Name,
+                  type = Type,
+                  pid = Pid,
+                  extra_bcc = ExtraBcc}.
+
 -spec is_amqqueue(any()) -> boolean().
 
 is_amqqueue(#amqqueue{}) -> true.
@@ -361,15 +387,21 @@ set_arguments(#amqqueue{} = Queue, Args) ->
 % options
 
 -spec get_options(amqqueue()) -> amqqueue_options().
-
 get_options(#amqqueue{options = Options}) ->
     Options.
 
 -spec set_options(amqqueue(), amqqueue_options()) -> amqqueue().
-
 set_options(#amqqueue{} = Queue, Options) ->
     Queue#amqqueue{options = Options}.
 
+-spec get_extra_bcc(amqqueue() | target()) ->
+    rabbit_misc:resource_name() | none.
+get_extra_bcc(#amqqueue{options = #{extra_bcc := Name}}) ->
+    Name;
+get_extra_bcc(#amqqueue{})  ->
+    none;
+get_extra_bcc(#queue_target{extra_bcc = Name}) ->
+    Name.
 
 % decorators
 
@@ -418,9 +450,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 +462,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 +522,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.
 
@@ -629,8 +664,6 @@ to_printable(QName = #resource{name = Name, virtual_host = VHost}, Type) ->
        <<"virtual_host">> => VHost,
        <<"type">> => Type}.
 
-% private
-
 macros() ->
     io:format(
       "-define(is_~ts(Q), is_record(Q, amqqueue, ~b)).~n~n",

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,60 @@ 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 amqqueue: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 amqqueue: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()) ->
+-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
@@ -2126,7 +2126,12 @@ deliver_to_queues(XName,
             rabbit_misc:protocol_error(
               resource_error,
               "Stream coordinator unavailable for ~ts",
-              [rabbit_misc:rs(Resource)])
+              [rabbit_misc:rs(Resource)]);
+        {error, Reason} ->
+            rabbit_misc:protocol_error(
+              resource_error,
+              "failed to deliver message: ~tp",
+              [Reason])
     end.
 
 process_routing_mandatory(_Mandatory = true,

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,57 @@ 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() -> lookup_targets(Names) end
+       }).
+
+lookup_targets(Names) ->
+    lists:filtermap(fun({Name, RouteInfos})
+                          when is_map(RouteInfos) ->
+                            case lookup_target(Name) of
+                                not_found -> false;
+                                Target -> {true, {Target, RouteInfos}}
+                            end;
+                       (Name) ->
+                            case lookup_target(Name) of
+                                not_found -> false;
+                                Target -> {true, Target}
+                            end
+                    end, Names).
+
+lookup_target(#resource{name = NameBin} = Name) ->
+    case rabbit_volatile_queue:is(NameBin) of
+        true ->
+            %% This queue is not stored in the database.
+            %% We create it on the fly.
+            case rabbit_volatile_queue:new(Name) of
+                error -> not_found;
+                Q -> Q
+            end;
+        false ->
+            try
+                ets:lookup_element(?KHEPRI_TARGET_PROJECTION, Name, 2, not_found) of
+                {Type, Pid, ExtraBcc} ->
+                    amqqueue:new_target(Name, Type, Pid, ExtraBcc);
+                not_found ->
+                    not_found
+            catch
+                error:badarg ->
+                    not_found
+            end
+    end.
+
 %% -------------------------------------------------------------------
 %% get_many().
 %% -------------------------------------------------------------------

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.