Speed up Direct Reply-To

Speed up Direct Reply-To in AMQP 0.9.1 because prior to this PR sending each RPC reply
was bottlenecked by the slowest network link between the RabbitMQ node the responder
connected to and all other RabbitMQ nodes. This regression was introduced in RabbitMQ 3.12.0
via d656371

(cherry picked from commit 9f2ff6c)
(cherry picked from commit 376d118)

Author: ansd

deps/rabbit/src/rabbit_channel.erl

@@ -302,7 +302,7 @@ send_command(Pid, Msg) ->
 
 -spec deliver_reply(binary(), mc:state()) -> 'ok'.
 deliver_reply(<<"amq.rabbitmq.reply-to.", EncodedBin/binary>>, Message) ->
-    Nodes = rabbit_nodes:all_running_with_hashes(),
+    Nodes = nodes_with_hashes(),
     case rabbit_direct_reply_to:decode_reply_to(EncodedBin, Nodes) of
         {ok, Pid, Key} ->
             delegate:invoke_no_result(
@@ -325,7 +325,7 @@ deliver_reply_local(Pid, Key, Message) ->
 declare_fast_reply_to(<<"amq.rabbitmq.reply-to">>) ->
     exists;
 declare_fast_reply_to(<<"amq.rabbitmq.reply-to.", EncodedBin/binary>>) ->
-    Nodes = rabbit_nodes:all_running_with_hashes(),
+    Nodes = nodes_with_hashes(),
     case rabbit_direct_reply_to:decode_reply_to(EncodedBin, Nodes) of
         {ok, Pid, Key} ->
             Msg = {declare_fast_reply_to, Key},
@@ -338,6 +338,9 @@ declare_fast_reply_to(<<"amq.rabbitmq.reply-to.", EncodedBin/binary>>) ->
 declare_fast_reply_to(_) ->
     not_found.
 
+nodes_with_hashes() ->
+    #{erlang:phash2(Node) => Node || Node <- rabbit_nodes:list_members()}.
+
 -spec list() -> [pid()].
 
 list() ->

deps/rabbit/src/rabbit_direct_reply_to.erl

@@ -20,7 +20,7 @@ compute_key_and_suffix(Pid) ->
     PidParts0 = #{node := Node} = pid_recomposition:decompose(Pid),
     %% Note: we hash the entire node name. This is sufficient for our needs of shortening node name
     %% in the TTB-encoded pid, and helps avoid doing the node name split for every single cluster member
-    %% in rabbit_nodes:all_running_with_hashes/0.
+    %% in rabbit_channel:nodes_with_hashes/0.
     %%
     %% We also use a synthetic node prefix because the hash alone will be sufficient to
     NodeHash = erlang:phash2(Node),

deps/rabbit/src/rabbit_nodes.erl

@@ -28,7 +28,7 @@
          await_running_count/2, is_single_node_cluster/0,
          boot/0]).
 -export([persistent_cluster_id/0, seed_internal_cluster_id/0, seed_user_provided_cluster_name/0]).
--export([all/0, all_running_with_hashes/0, target_cluster_size_hint/0, reached_target_cluster_size/0,
+-export([all/0, target_cluster_size_hint/0, reached_target_cluster_size/0,
          if_reached_target_cluster_size/2]).
 -export([lock_id/1, lock_retries/0]).
 -export([me_in_nodes/1, nodes_incl_me/1, nodes_excl_me/1]).
@@ -614,10 +614,6 @@ await_running_count_with_retries(TargetCount, Retries) ->
             await_running_count_with_retries(TargetCount, Retries - 1)
     end.
 
--spec all_running_with_hashes() -> #{non_neg_integer() => node()}.
-all_running_with_hashes() ->
-    maps:from_list([{erlang:phash2(Node), Node} || Node <- list_running()]).
-
 -spec target_cluster_size_hint() -> non_neg_integer().
 target_cluster_size_hint() ->
     cluster_formation_key_or_default(target_cluster_size_hint, ?DEFAULT_TARGET_CLUSTER_SIZE).