Optimisation for 'delegate'

This is copied from https://github.com/rabbitmq/rabbitmq-common/pull/349


If a message is sent to only one queue(in most application scenarios), passing through the 'delegate' is meaningless. Otherwise, it increases the delay of the message and the possibility of 'delegate' congestion.

Here are some test data：
node1: Pentium(R) Dual-Core CPU E5300 @ 2.60GHz
node2: Pentium(R) Dual-Core CPU E5300 @ 2.60GHz

Join node1 and node2 to a cluster. Create 100 queues on node2, and start 100 consumers to receive messages from these queues.
Start 100 publishers on node1 to send messages to the queues of node2. Each publisher will send 10k messages at the rate of 100/s(10k/s theoretically in total), and all the messages for all publishers is 1 million.

Before optimisation:
{1,[{msg_time,812312(=<1ms),177922(=<5ms),9507(=<50ms),221(=<500ms),38(=<1000ms),0,0,0,0,1061,1069,0,0}]}

After optimisation:
{1,[{msg_time,902854(=< 1ms),93993(=<5ms),3038(=<50ms),96(=<500ms),19(=<1000ms),0,0,0,0,1049,1060,0,0}]}

Additional information:

Time counted here is the stay time of a message in the cluster, that is, Time(leaving from node2 at) - Time(reaching node1 at).
"812312(=<1ms)" is the number of messages with time consumption less than or equal to 1ms.
Overall, the optimisation is effective.

1 files changed, 59 insertions, 2 deletions

diff --git a/deps/rabbit_common/src/delegate.erl b/deps/rabbit_common/src/delegate.erl
index bdea11534d..e6e76d2ea0 100644
--- a/deps/rabbit_common/src/delegate.erl
+++ b/deps/rabbit_common/src/delegate.erl
@@ -36,6 +36,17 @@
 %% the pool is configurable, the aim is to make sure we don't have too
 %% few delegates and thus limit performance on many-CPU machines.
 
+%% Optimisation for 'delegate'
+%% If a message is sent to only one queue(in most application scenarios),
+%% passing through the 'delegate' is meaningless.
+%% Hardcoding "?DEFAULT_NAME and/or gen_server2" is to avoid affecting those
+%%  operations that must go through the 'delegate', such as:
+%%  1. "delegate:invoke(Pids, {erlang, process_info, [memory]})", "erlang, process_info" 
+%%      must be called inside the target node. 
+%%  2. "{Results, Errors} = delegate:invoke(MemberPids, ?DELEGATE_PREFIX, FunOrMFA)",
+%%      For some reason, the operation specifically specifies a delegate name rather than
+%%      ?DEFAULT_NAME.
+
 -behaviour(gen_server2).
 
 -export([start_link/1, start_link/2, invoke_no_result/2,
@@ -77,11 +88,21 @@ start_link(Name, Num) ->
     Name1 = delegate_name(Name, Num),
     gen_server2:start_link({local, Name1}, ?MODULE, [Name1], []).
 
+invoke(Pid, FunOrMFA = {gen_server2, _F, _A}) when is_pid(Pid) ->  %% optimisation
+    case safe_invoke(Pid, FunOrMFA) of
+        {ok,    _, Result} -> Result;
+        {error, _, {Class, Reason, StackTrace}}  -> erlang:raise(Class, Reason, StackTrace)
+    end;
 invoke(Pid, FunOrMFA) ->
     invoke(Pid, ?DEFAULT_NAME, FunOrMFA).
 
 invoke(Pid, _Name, FunOrMFA) when is_pid(Pid) andalso node(Pid) =:= node() ->
     apply1(FunOrMFA, Pid);
+invoke(Pid, ?DEFAULT_NAME, FunOrMFA = {gen_server2, _F, _A}) when is_pid(Pid) ->  %% optimisation
+    case safe_invoke(Pid, FunOrMFA) of
+        {ok,    _, Result} -> Result;
+        {error, _, {Class, Reason, StackTrace}}  -> erlang:raise(Class, Reason, StackTrace)
+    end;
 invoke(Pid, Name, FunOrMFA) when is_pid(Pid) ->
     case invoke([Pid], Name, FunOrMFA) of
         {[{Pid, Result}], []} ->
@@ -92,13 +113,24 @@ invoke(Pid, Name, FunOrMFA) when is_pid(Pid) ->
 
 invoke([], _Name, _FunOrMFA) -> %% optimisation
     {[], []};
+invoke([Pid], ?DEFAULT_NAME, FunOrMFA = {gen_server2, _F, _A}) when is_pid(Pid) -> %% optimisation
+    case safe_invoke(Pid, FunOrMFA) of
+        {ok,    _, Result} -> {[{Pid, Result}], []};
+        {error, _, Error}  -> {[], [{Pid, Error}]}
+    end;
 invoke([Pid], _Name, FunOrMFA) when node(Pid) =:= node() -> %% optimisation
     case safe_invoke(Pid, FunOrMFA) of
         {ok,    _, Result} -> {[{Pid, Result}], []};
         {error, _, Error}  -> {[], [{Pid, Error}]}
     end;
+invoke(Pids, Name = ?DEFAULT_NAME, FunOrMFA = {gen_server2, _F, _A}) when is_list(Pids) ->
+    {LocalCallPids, Grouped} = group_local_call_pids_by_node(Pids),
+    invoke(Pids, Name, FunOrMFA, LocalCallPids, Grouped);
 invoke(Pids, Name, FunOrMFA) when is_list(Pids) ->
     {LocalPids, Grouped} = group_pids_by_node(Pids),
+    invoke(Pids, Name, FunOrMFA, LocalPids, Grouped).
+
+invoke(Pids, Name, FunOrMFA, LocalCallPids, Grouped) when is_list(Pids) ->
     %% The use of multi_call is only safe because the timeout is
     %% infinity, and thus there is no process spawned in order to do
     %% the sending. Thus calls can't overtake preceding calls/casts.
@@ -112,7 +144,7 @@ invoke(Pids, Name, FunOrMFA) when is_list(Pids) ->
     BadPids = [{Pid, {exit, {nodedown, BadNode}, []}} ||
                   BadNode <- BadNodes,
                   Pid     <- maps:get(BadNode, Grouped)],
-    ResultsNoNode = lists:append([safe_invoke(LocalPids, FunOrMFA) |
+    ResultsNoNode = lists:append([safe_invoke(LocalCallPids, FunOrMFA) |
                                   [Results || {_Node, Results} <- Replies]]),
     lists:foldl(
       fun ({ok,    Pid, Result}, {Good, Bad}) -> {[{Pid, Result} | Good], Bad};
@@ -134,6 +166,9 @@ demonitor(Ref) when is_reference(Ref) ->
 demonitor({Name, Pid}) ->
     gen_server2:cast(Name, {demonitor, self(), Pid}).
 
+invoke_no_result(Pid, FunOrMFA = {gen_server2, _F, _A}) when is_pid(Pid) ->
+    _ = safe_invoke(Pid, FunOrMFA), %% we don't care about any error
+    ok;
 invoke_no_result(Pid, FunOrMFA) when is_pid(Pid) andalso node(Pid) =:= node() ->
     %% Optimization, avoids calling invoke_no_result/3.
     %%
@@ -154,6 +189,9 @@ invoke_no_result(Pid, FunOrMFA) when is_pid(Pid) ->
     ok;
 invoke_no_result([], _FunOrMFA) -> %% optimisation
     ok;
+invoke_no_result([Pid], FunOrMFA = {gen_server2, _F, _A}) when is_pid(Pid) -> %% optimisation
+    _ = safe_invoke(Pid, FunOrMFA), %% must not die
+    ok;
 invoke_no_result([Pid], FunOrMFA) when node(Pid) =:= node() -> %% optimisation
     _ = safe_invoke(Pid, FunOrMFA), %% must not die
     ok;
@@ -163,15 +201,21 @@ invoke_no_result([Pid], FunOrMFA) ->
                        {invoke, FunOrMFA,
                         maps:from_list([{RemoteNode, [Pid]}])}),
     ok;
+invoke_no_result(Pids, FunOrMFA = {gen_server2, _F, _A}) when is_list(Pids) ->
+    {LocalCallPids, Grouped} = group_local_call_pids_by_node(Pids),
+    invoke_no_result(Pids, FunOrMFA, LocalCallPids, Grouped);
 invoke_no_result(Pids, FunOrMFA) when is_list(Pids) ->
     {LocalPids, Grouped} = group_pids_by_node(Pids),
+    invoke_no_result(Pids, FunOrMFA, LocalPids, Grouped).
+
+invoke_no_result(Pids, FunOrMFA, LocalCallPids, Grouped) when is_list(Pids) ->
     case maps:keys(Grouped) of
         []          -> ok;
         RemoteNodes -> gen_server2:abcast(
                          RemoteNodes, delegate(self(), ?DEFAULT_NAME, RemoteNodes),
                          {invoke, FunOrMFA, Grouped})
     end,
-    _ = safe_invoke(LocalPids, FunOrMFA), %% must not die
+    _ = safe_invoke(LocalCallPids, FunOrMFA), %% must not die
     ok.
 
 %%----------------------------------------------------------------------------
@@ -187,6 +231,19 @@ group_pids_by_node(Pids) ->
                  node(Pid), fun (List) -> [Pid | List] end, [Pid], Remote)}
       end, {[], maps:new()}, Pids).
 
+group_local_call_pids_by_node(Pids) ->
+    {LocalPids0, Grouped0} = group_pids_by_node(Pids),
+    maps:fold(fun(K, V, {AccIn, MapsIn}) -> 
+        case V of
+            %% just one Pid for the node
+            [SinglePid] -> {[SinglePid | AccIn], MapsIn};
+            %% If the value is a list of more than one pid, 
+            %% the (K,V) will be put into the new map which will be called 
+            %% through delegate to reduce inter-node communication.
+            _ -> {AccIn, maps:update_with(K, fun(V1) -> V1 end, V, MapsIn)}
+        end
+    end, {LocalPids0, maps:new()}, Grouped0).
+
 delegate_name(Name, Hash) ->
     list_to_atom(Name ++ integer_to_list(Hash)).