Speed up fanout exchange

Resolves #14531

 ## What?
Increase end-to-end message throughput for messages routed via the fanout exchange by ~42% (see benchmark below).
In addition to the fanout exchange, a similar speed up is achieved for the following exchange types:
* modulus hash
* random
* recent history

This applies only if Khepri is enabled.

 ## How?
Use an additional routing table (projection) whose table key is the source exchange.
Looking up the destinations happens then by an ETS table key.

Prior to this commit, CPUs were busy compiling the same match spec for every incoming message.

 ## Benchmark
1. Start RabbitMQ:
```
make run-broker RABBITMQ_SERVER_ADDITIONAL_ERL_ARGS="+S 5" \
    RABBITMQ_CONFIG_FILE="advanced.config" PLUGINS="rabbitmq_management"
```
where `advanced.config` contains:
```
[
 {rabbitmq_management_agent, [
  {disable_metrics_collector, true}
 ]}
].
```

2. Create a queue and binding:
```
deps/rabbitmq_management/bin/rabbitmqadmin declare queue queue_type=classic durable=true name=q1 && \
deps/rabbitmq_management/bin/rabbitmqadmin declare binding source=amq.fanout destination=q1
```

3. Create the load
```
java -jar target/perf-test.jar -p -e amq.fanout -u q1 -s 5 --autoack -z 60
```

Before this commit:
```
sending rate avg: 97394 msg/s
receiving rate avg: 97394 msg/s
```

After this commit:
```
sending rate avg: 138677 msg/s
receiving rate avg: 138677 msg/s
```

The CPU flamegraph shows that `rabbit_exchange:route/3` consumes the following CPU amounts:
* 13.5% before this commit
* 3.4% after this commit

 ## Downsides
Additional ETS memory usage for the new projection table.
However, the new table does not store any binding entries for the following
source exchange types:
* direct
* headers
* topic
* x-local-random

Author: ansd

deps/rabbit/src/rabbit_db_binding.erl

@@ -57,6 +57,7 @@
 -define(MNESIA_INDEX_TABLE, rabbit_index_route).
 -define(KHEPRI_BINDINGS_PROJECTION, rabbit_khepri_binding).
 -define(KHEPRI_INDEX_ROUTE_PROJECTION, rabbit_khepri_index_route).
+-define(KHEPRI_SOURCE_ROUTE_PROJECTION, rabbit_khepri_source_route).
 
 %% -------------------------------------------------------------------
 %% exists().
@@ -707,14 +708,19 @@ match_routing_key_in_mnesia(SrcName, RoutingKeys, UseIndex) ->
     end.
 
 match_routing_key_in_khepri(Src, ['_']) ->
-    try
-        MatchHead = #index_route{source_key  = {Src, '_'},
-                                 destination = '$1',
-                                 _           = '_'},
-        ets:select(?KHEPRI_INDEX_ROUTE_PROJECTION, [{MatchHead, [], ['$1']}])
-    catch
-        error:badarg ->
-            []
+    try ets:lookup_element(?KHEPRI_SOURCE_ROUTE_PROJECTION,
+                           Src, #source_route.destination, [])
+    catch error:badarg ->
+              %% The new projection doesn't exist.
+              %% Fall back to old & slow approach.
+              MatchHead = #index_route{source_key  = {Src, '_'},
+                                       destination = '$1',
+                                       _           = '_'},
+              try ets:select(?KHEPRI_INDEX_ROUTE_PROJECTION,
+                             [{MatchHead, [], ['$1']}])
+              catch error:badarg ->
+                        []
+              end
     end;
 match_routing_key_in_khepri(Src, RoutingKeys) ->
     lists:foldl(

deps/rabbit/src/rabbit_khepri.erl

@@ -1319,6 +1319,7 @@ register_projections() ->
                fun register_rabbit_user_permissions_projection/0,
                fun register_rabbit_bindings_projection/0,
                fun register_rabbit_index_route_projection/0,
+               fun register_rabbit_source_route_projection/0,
                fun register_rabbit_topic_graph_projection/0],
     rabbit_misc:for_each_while_ok(
       fun(RegisterFun) ->
@@ -1442,6 +1443,38 @@ register_rabbit_index_route_projection() ->
                     _RoutingKey = ?KHEPRI_WILDCARD_STAR),
     khepri:register_projection(?STORE_ID, PathPattern, Projection).
 
+register_rabbit_source_route_projection() ->
+    MapFun = fun(_Path, #binding{source = Source,
+                                 key = Key,
+                                 destination = Destination,
+                                 args = Args}) ->
+                     #source_route{source = Source,
+                                   key = Key,
+                                   destination = Destination,
+                                   args = Args}
+             end,
+    ProjectionFun = projection_fun_for_sets(MapFun),
+    Options = #{type => bag,
+                keypos => #source_route.source,
+                read_concurrency => true},
+    Projection = khepri_projection:new(
+                   rabbit_khepri_source_route, ProjectionFun, Options),
+    Exchange = #if_data_matches{
+                  pattern = #exchange{type = '$1', _ = '_'},
+                  conditions = [{'andalso',
+                                 {'=/=', '$1', direct},
+                                 {'=/=', '$1', headers},
+                                 {'=/=', '$1', topic},
+                                 {'=/=', '$1', 'x-local-random'}
+                                }]},
+    PathPattern = rabbit_db_binding:khepri_route_path(
+                    _VHost = ?KHEPRI_WILDCARD_STAR,
+                    Exchange,
+                    _Kind = ?KHEPRI_WILDCARD_STAR,
+                    _DstName = ?KHEPRI_WILDCARD_STAR,
+                    _RoutingKey = ?KHEPRI_WILDCARD_STAR),
+    khepri:register_projection(?STORE_ID, PathPattern, Projection).
+
 %% Routing information is stored in the Khepri store as a `set'.
 %% In order to turn these bindings into records in an ETS `bag', we use a
 %% `khepri_projection:extended_projection_fun()' to determine the changes

deps/rabbit_common/include/rabbit.hrl

@@ -94,6 +94,7 @@
 -record(route, {binding, value = const}).
 -record(reverse_route, {reverse_binding, value = const}).
 -record(index_route, {source_key, destination, args = []}).
+-record(source_route, {source, key, destination, args = []}).
 
 -record(binding, {source, key, destination, args = []}).
 -record(reverse_binding, {destination, key, source, args = []}).