From: Alexey Lebedeff <alebedev@mirantis.com>

Date: Wed, 9 Mar 2016 14:55:02 +0300

Subject: [PATCH] Avoid RPC roundtrips while listing items

- Emit info about particular items in parallel on every node, with

  results delivered directly to a `rabbitmqctl` instance.

- `rabbit_control_misc:wait_for_info_messages/5` can wait for results of

  more than one emitting map.

- Stop passing arround InfoItemKeys in

  `rabbit_control_misc:wait_for_info_messages/5`, the same information

  could be directly encoded in DisplayFun closure.

- Add `emit` to function names, to avoid confusion with regular ones

  which return result directly.

Part of https://github.com/rabbitmq/rabbitmq-server/pull/683

diff --git a/src/rabbit_amqqueue.erl b/src/rabbit_amqqueue.erl

index 27b352a..38300bd 100644

--- a/src/rabbit_amqqueue.erl

+++ b/src/rabbit_amqqueue.erl

@@ -25,10 +25,10 @@

          check_exclusive_access/2, with_exclusive_access_or_die/3,

          stat/1, deliver/2, requeue/3, ack/3, reject/4]).

 -export([list/0, list/1, info_keys/0, info/1, info/2, info_all/1, info_all/2,

-         info_all/6, info_local/1]).

+         emit_info_all/5, list_local/1, info_local/1]).

 -export([list_down/1]).

 -export([force_event_refresh/1, notify_policy_changed/1]).

--export([consumers/1, consumers_all/1,  consumers_all/3, consumer_info_keys/0]).

+-export([consumers/1, consumers_all/1,  emit_consumers_all/4, consumer_info_keys/0]).

 -export([basic_get/4, basic_consume/10, basic_cancel/4, notify_decorators/1]).

 -export([notify_sent/2, notify_sent_queue_down/1, resume/2]).

 -export([notify_down_all/2, notify_down_all/3, activate_limit_all/2, credit/5]).

@@ -41,7 +41,8 @@

 %% internal

 -export([internal_declare/2, internal_delete/1, run_backing_queue/3,

-         set_ram_duration_target/2, set_maximum_since_use/2]).

+         set_ram_duration_target/2, set_maximum_since_use/2,

+         emit_info_local/4, emit_info_down/4, emit_consumers_local/3]).

 -include("rabbit.hrl").

 -include_lib("stdlib/include/qlc.hrl").

@@ -117,10 +118,6 @@

 -spec info_all(rabbit_types:vhost()) -> [rabbit_types:infos()].

 -spec info_all(rabbit_types:vhost(), rabbit_types:info_keys()) ->

           [rabbit_types:infos()].

--spec info_all

-        (rabbit_types:vhost(), rabbit_types:info_keys(), boolean(), boolean(),

-         reference(), pid()) ->

-            'ok'.

 -spec force_event_refresh(reference()) -> 'ok'.

 -spec notify_policy_changed(rabbit_types:amqqueue()) -> 'ok'.

 -spec consumers(rabbit_types:amqqueue()) ->

@@ -130,7 +127,6 @@

 -spec consumers_all(rabbit_types:vhost()) ->

           [{name(), pid(), rabbit_types:ctag(), boolean(),

             non_neg_integer(), rabbit_framing:amqp_table()}].

--spec consumers_all(rabbit_types:vhost(), reference(), pid()) -> 'ok'.

 -spec stat(rabbit_types:amqqueue()) ->

           {'ok', non_neg_integer(), non_neg_integer()}.

 -spec delete_immediately(qpids()) -> 'ok'.

@@ -627,16 +623,23 @@ info_all(VHostPath, Items) ->

     map(list(VHostPath), fun (Q) -> info(Q, Items) end) ++

         map(list_down(VHostPath), fun (Q) -> info_down(Q, Items, down) end).

-info_all(VHostPath, Items, NeedOnline, NeedOffline, Ref, AggregatorPid) ->

-    NeedOnline andalso rabbit_control_misc:emitting_map_with_exit_handler(

-                         AggregatorPid, Ref, fun(Q) -> info(Q, Items) end, list(VHostPath),

-                         continue),

-    NeedOffline andalso rabbit_control_misc:emitting_map_with_exit_handler(

-                          AggregatorPid, Ref, fun(Q) -> info_down(Q, Items, down) end,

-                          list_down(VHostPath),

-                          continue),

-    %% Previous maps are incomplete, finalize emission

-    rabbit_control_misc:emitting_map(AggregatorPid, Ref, fun(_) -> no_op end, []).

+emit_info_local(VHostPath, Items, Ref, AggregatorPid) ->

+    rabbit_control_misc:emitting_map_with_exit_handler(

+      AggregatorPid, Ref, fun(Q) -> info(Q, Items) end, list_local(VHostPath)).

+

+emit_info_all(Nodes, VHostPath, Items, Ref, AggregatorPid) ->

+    Pids = [ spawn_link(Node, rabbit_amqqueue, emit_info_local, [VHostPath, Items, Ref, AggregatorPid]) || Node <- Nodes ],

+    rabbit_control_misc:await_emitters_termination(Pids).

+

+emit_info_down(VHostPath, Items, Ref, AggregatorPid) ->

+    rabbit_control_misc:emitting_map_with_exit_handler(

+      AggregatorPid, Ref, fun(Q) -> info_down(Q, Items, down) end,

+      list_down(VHostPath)).

+

+list_local(VHostPath) ->

+    [ Q || #amqqueue{state = State, pid=QPid} = Q <- list(VHostPath),

+           State =/= crashed,

+           node() =:= node(QPid) ].

 info_local(VHostPath) ->

     map(list_local(VHostPath), fun (Q) -> info(Q, [name]) end).

@@ -664,12 +667,17 @@ consumers_all(VHostPath) ->

       map(list(VHostPath),

           fun(Q) -> get_queue_consumer_info(Q, ConsumerInfoKeys) end)).

-consumers_all(VHostPath, Ref, AggregatorPid) ->

+emit_consumers_all(Nodes, VHostPath, Ref, AggregatorPid) ->

+    Pids = [ spawn_link(Node, rabbit_amqqueue, emit_consumers_local, [VHostPath, Ref, AggregatorPid]) || Node <- Nodes ],

+    rabbit_control_misc:await_emitters_termination(Pids),

+    ok.

+

+emit_consumers_local(VHostPath, Ref, AggregatorPid) ->

     ConsumerInfoKeys = consumer_info_keys(),

     rabbit_control_misc:emitting_map(

       AggregatorPid, Ref,

       fun(Q) -> get_queue_consumer_info(Q, ConsumerInfoKeys) end,

-      list(VHostPath)).

+      list_local(VHostPath)).

 get_queue_consumer_info(Q, ConsumerInfoKeys) ->

     [lists:zip(ConsumerInfoKeys,

diff --git a/src/rabbit_channel.erl b/src/rabbit_channel.erl

index ab7d38d..837a892 100644

--- a/src/rabbit_channel.erl

+++ b/src/rabbit_channel.erl

@@ -56,7 +56,7 @@

 -export([send_command/2, deliver/4, deliver_reply/2,

          send_credit_reply/2, send_drained/2]).

 -export([list/0, info_keys/0, info/1, info/2, info_all/0, info_all/1,

-         info_all/3, info_local/1]).

+         emit_info_all/4, info_local/1]).

 -export([refresh_config_local/0, ready_for_close/1]).

 -export([force_event_refresh/1]).

@@ -64,7 +64,7 @@

          handle_info/2, handle_pre_hibernate/1, prioritise_call/4,

          prioritise_cast/3, prioritise_info/3, format_message_queue/2]).

 %% Internal

--export([list_local/0, deliver_reply_local/3]).

+-export([list_local/0, emit_info_local/3, deliver_reply_local/3]).

 -export([get_vhost/1, get_user/1]).

 -record(ch, {

@@ -220,7 +220,6 @@

 -spec info(pid(), rabbit_types:info_keys()) -> rabbit_types:infos().

 -spec info_all() -> [rabbit_types:infos()].

 -spec info_all(rabbit_types:info_keys()) -> [rabbit_types:infos()].

--spec info_all(rabbit_types:info_keys(), reference(), pid()) -> 'ok'.

 -spec refresh_config_local() -> 'ok'.

 -spec ready_for_close(pid()) -> 'ok'.

 -spec force_event_refresh(reference()) -> 'ok'.

@@ -329,9 +328,16 @@ info_all(Items) ->

 info_local(Items) ->

     rabbit_misc:filter_exit_map(fun (C) -> info(C, Items) end, list_local()).

-info_all(Items, Ref, AggregatorPid) ->

+emit_info_all(Nodes, Items, Ref, AggregatorPid) ->

+    Pids = [ spawn_link(Node, rabbit_channel, emit_info_local, [Items, Ref, AggregatorPid]) || Node <- Nodes ],

+    rabbit_control_misc:await_emitters_termination(Pids).

+

+emit_info_local(Items, Ref, AggregatorPid) ->

+    emit_info(list_local(), Items, Ref, AggregatorPid).

+

+emit_info(PidList, InfoItems, Ref, AggregatorPid) ->

     rabbit_control_misc:emitting_map_with_exit_handler(

-      AggregatorPid, Ref, fun(C) -> info(C, Items) end, list()).

+      AggregatorPid, Ref, fun(C) -> info(C, InfoItems) end, PidList).

 refresh_config_local() ->

     rabbit_misc:upmap(

diff --git a/src/rabbit_control_misc.erl b/src/rabbit_control_misc.erl

index 2e1f6cc..3b0c60b 100644

--- a/src/rabbit_control_misc.erl

+++ b/src/rabbit_control_misc.erl

@@ -17,7 +17,8 @@

 -module(rabbit_control_misc).

 -export([emitting_map/4, emitting_map/5, emitting_map_with_exit_handler/4,

-         emitting_map_with_exit_handler/5, wait_for_info_messages/5,

+         emitting_map_with_exit_handler/5, wait_for_info_messages/6,

+         spawn_emitter_caller/7, await_emitters_termination/1,

          print_cmd_result/2]).

 -spec emitting_map(pid(), reference(), fun(), list()) -> 'ok'.

@@ -25,7 +26,14 @@

 -spec emitting_map_with_exit_handler

         (pid(), reference(), fun(), list()) -> 'ok'.

 -spec emitting_map_with_exit_handler

-        (pid(), reference(), fun(), list(), atom()) -> 'ok'.

+        (pid(), reference(), fun(), list(), 'continue') -> 'ok'.

+

+-type fold_fun() :: fun ((term(), term()) -> term()).

+

+-spec wait_for_info_messages (pid(), reference(), fold_fun(), term(), timeout(), non_neg_integer()) -> {'ok', term()} | {'error', term()}.

+-spec spawn_emitter_caller (node(), module(), atom(), [term()], reference(), pid(), timeout()) -> 'ok'.

+-spec await_emitters_termination ([pid()]) -> 'ok'.

+

 -spec print_cmd_result(atom(), term()) -> 'ok'.

 emitting_map(AggregatorPid, Ref, Fun, List) ->

@@ -65,27 +73,108 @@ step_with_exit_handler(AggregatorPid, Ref, Fun, Item) ->

             ok

     end.

-wait_for_info_messages(Pid, Ref, ArgAtoms, DisplayFun, Timeout) ->

-    _ = notify_if_timeout(Pid, Ref, Timeout),

-    wait_for_info_messages(Ref, ArgAtoms, DisplayFun).

+%% Invokes RPC for async info collection in separate (but linked to

+%% the caller) process. Separate process waits for RPC to finish and

+%% in case of errors sends them in wait_for_info_messages/5-compatible

+%% form to aggregator process. Calling process is then expected to

+%% do blocking call of wait_for_info_messages/5.

+%%

+%% Remote function MUST use calls to emitting_map/4 (and other

+%% emitting_map's) to properly deliver requested information to an

+%% aggregator process.

+%%

+%% If for performance reasons several parallel emitting_map's need to

+%% be run, remote function MUST NOT return until all this

+%% emitting_map's are done. And during all this time remote RPC

+%% process MUST be linked to emitting

+%% processes. await_emitters_termination/1 helper can be used as a

+%% last statement of remote function to ensure this behaviour.

+spawn_emitter_caller(Node, Mod, Fun, Args, Ref, Pid, Timeout) ->

+    spawn_monitor(

+      fun () ->

+              case rpc_call_emitter(Node, Mod, Fun, Args, Ref, Pid, Timeout) of

+                  {error, _} = Error        ->

+                      Pid ! {Ref, error, Error};

+                  {bad_argument, _} = Error ->

+                      Pid ! {Ref, error, Error};

+                  {badrpc, _} = Error       ->

+                      Pid ! {Ref, error, Error};

+                  _                         ->

+                      ok

+              end

+      end),

+    ok.

+

+rpc_call_emitter(Node, Mod, Fun, Args, Ref, Pid, Timeout) ->

+    rabbit_misc:rpc_call(Node, Mod, Fun, Args++[Ref, Pid], Timeout).

+

+%% Agregator process expects correct numbers of explicits ACKs about

+%% finished emission process. While everything is linked, we still

+%% need somehow to wait for termination of all emitters before

+%% returning from RPC call - otherwise links will be just broken with

+%% reason 'normal' and we can miss some errors, and subsequentially

+%% hang.

+await_emitters_termination(Pids) ->

+    Monitors = [erlang:monitor(process, Pid) || Pid <- Pids],

+    collect_monitors(Monitors).

-wait_for_info_messages(Ref, InfoItemKeys, DisplayFun) when is_reference(Ref) ->

+collect_monitors([]) ->

+    ok;

+collect_monitors([Monitor|Rest]) ->

     receive

-        {Ref,  finished}         ->

-            ok;

-        {Ref,  {timeout, T}}     ->

+        {'DOWN', Monitor, _Pid, normal} ->

+            collect_monitors(Rest);

+        {'DOWN', Monitor, _Pid, noproc} ->

+            %% There is a link and a monitor to a process. Matching

+            %% this clause means that process has gracefully

+            %% terminated even before we've started monitoring.

+            collect_monitors(Rest);

+        {'DOWN', _, Pid, Reason} ->

+            exit({emitter_exit, Pid, Reason})

+    end.

+

+%% Wait for result of one or more calls to emitting_map-family

+%% functions.

+%%

+%% Number of expected acknowledgments is specified by ChunkCount

+%% argument. Most common usage will be with ChunkCount equals to

+%% number of live nodes, but it's not mandatory - thus more generic

+%% name of 'ChunkCount' was chosen.

+wait_for_info_messages(Pid, Ref, Fun, Acc0, Timeout, ChunkCount) ->

+    notify_if_timeout(Pid, Ref, Timeout),

+    wait_for_info_messages(Ref, Fun, Acc0, ChunkCount).

+

+wait_for_info_messages(Ref, Fun, Acc0, ChunksLeft) ->

+    receive

+        {Ref, finished} when ChunksLeft =:= 1 ->

+            {ok, Acc0};

+        {Ref, finished} ->

+            wait_for_info_messages(Ref, Fun, Acc0, ChunksLeft - 1);

+        {Ref, {timeout, T}} ->

             exit({error, {timeout, (T / 1000)}});

-        {Ref,  []}               ->

-            wait_for_info_messages(Ref, InfoItemKeys, DisplayFun);

-        {Ref,  Result, continue} ->

-            DisplayFun(Result, InfoItemKeys),

-            wait_for_info_messages(Ref, InfoItemKeys, DisplayFun);

-        {error, Error}           ->

-            Error;

-        _                        ->

-            wait_for_info_messages(Ref, InfoItemKeys, DisplayFun)

+        {Ref, []} ->

+            wait_for_info_messages(Ref, Fun, Acc0, ChunksLeft);

+        {Ref, Result, continue} ->

+            wait_for_info_messages(Ref, Fun, Fun(Result, Acc0), ChunksLeft);

+        {Ref, error, Error} ->

+            {error, simplify_emission_error(Error)};

+        {'DOWN', _MRef, process, _Pid, normal} ->

+            wait_for_info_messages(Ref, Fun, Acc0, ChunksLeft);

+        {'DOWN', _MRef, process, _Pid, Reason} ->

+            {error, simplify_emission_error(Reason)};

+        _Msg ->

+            wait_for_info_messages(Ref, Fun, Acc0, ChunksLeft)

     end.

+simplify_emission_error({badrpc, {'EXIT', {{nocatch, EmissionError}, _Stacktrace}}}) ->

+    EmissionError;

+simplify_emission_error({{nocatch, EmissionError}, _Stacktrace}) ->

+    EmissionError;

+simplify_emission_error(Anything) ->

+    {error, Anything}.

+

+notify_if_timeout(_, _, infinity) ->

+    ok;

 notify_if_timeout(Pid, Ref, Timeout) ->

     timer:send_after(Timeout, Pid, {Ref, {timeout, Timeout}}).

diff --git a/src/rabbit_misc.erl b/src/rabbit_misc.erl

index 8965c59..9341ea9 100644

--- a/src/rabbit_misc.erl

+++ b/src/rabbit_misc.erl

@@ -75,7 +75,7 @@

 -export([get_env/3]).

 -export([get_channel_operation_timeout/0]).

 -export([random/1]).

--export([rpc_call/4, rpc_call/5, rpc_call/7]).

+-export([rpc_call/4, rpc_call/5]).

 -export([report_default_thread_pool_size/0]).

 -export([get_gc_info/1]).

@@ -264,8 +264,6 @@

 -spec random(non_neg_integer()) -> non_neg_integer().

 -spec rpc_call(node(), atom(), atom(), [any()]) -> any().

 -spec rpc_call(node(), atom(), atom(), [any()], number()) -> any().

--spec rpc_call

-        (node(), atom(), atom(), [any()], reference(), pid(), number()) -> any().

 -spec report_default_thread_pool_size() -> 'ok'.

 -spec get_gc_info(pid()) -> integer().

@@ -1184,9 +1182,6 @@ rpc_call(Node, Mod, Fun, Args, Timeout) ->

                            rpc:call(Node, Mod, Fun, Args, Timeout)

     end.

-rpc_call(Node, Mod, Fun, Args, Ref, Pid, Timeout) ->

-    rpc_call(Node, Mod, Fun, Args++[Ref, Pid], Timeout).

-

 guess_number_of_cpu_cores() ->

     case erlang:system_info(logical_processors_available) of

         unknown -> % Happens on Mac OS X.

diff --git a/src/rabbit_networking.erl b/src/rabbit_networking.erl

index 5bf30ff..63e3ed0 100644

--- a/src/rabbit_networking.erl

+++ b/src/rabbit_networking.erl

@@ -33,7 +33,8 @@

          node_listeners/1, register_connection/1, unregister_connection/1,

          connections/0, connection_info_keys/0,

          connection_info/1, connection_info/2,

-         connection_info_all/0, connection_info_all/1, connection_info_all/3,

+         connection_info_all/0, connection_info_all/1,

+         emit_connection_info_all/4, emit_connection_info_local/3,

          close_connection/2, force_connection_event_refresh/1, tcp_host/1]).

 %% Used by TCP-based transports, e.g. STOMP adapter

@@ -89,8 +90,6 @@

 -spec connection_info_all() -> [rabbit_types:infos()].

 -spec connection_info_all(rabbit_types:info_keys()) ->

           [rabbit_types:infos()].

--spec connection_info_all(rabbit_types:info_keys(), reference(), pid()) ->

-          'ok'.

 -spec close_connection(pid(), string()) -> 'ok'.

 -spec force_connection_event_refresh(reference()) -> 'ok'.

@@ -365,10 +364,15 @@ connection_info(Pid, Items) -> rabbit_reader:info(Pid, Items).

 connection_info_all() -> cmap(fun (Q) -> connection_info(Q) end).

 connection_info_all(Items) -> cmap(fun (Q) -> connection_info(Q, Items) end).

-connection_info_all(Items, Ref, AggregatorPid) ->

+emit_connection_info_all(Nodes, Items, Ref, AggregatorPid) ->

+    Pids = [ spawn_link(Node, rabbit_networking, emit_connection_info_local, [Items, Ref, AggregatorPid]) || Node <- Nodes ],

+    rabbit_control_misc:await_emitters_termination(Pids),

+    ok.

+

+emit_connection_info_local(Items, Ref, AggregatorPid) ->

     rabbit_control_misc:emitting_map_with_exit_handler(

       AggregatorPid, Ref, fun(Q) -> connection_info(Q, Items) end,

-      connections()).

+      connections_local()).

 close_connection(Pid, Explanation) ->

     rabbit_log:info("Closing connection ~p because ~p~n", [Pid, Explanation]),