Merge pull request #14507 from rabbitmq/fix-direct-reply-to-regression

Speed up Direct Reply-To

Author: ansd

deps/rabbit/src/rabbit_channel.erl

@@ -300,7 +300,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(
@@ -323,7 +323,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},
@@ -336,6 +336,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).