--- /dev/null
+From: Alexey Lebedeff <alebedev@mirantis.com>
+Date: Wed, 9 Mar 2016 18:09:04 +0300
+Subject: [PATCH] Avoid RPC roundtrips in list commands
+
+Current implementation of various `list_XXX` commands require cross-node
+roundtrip for every processed item - because `rabbitmqctl` target node
+is responsible for gathering global list of all items of
+interest (channels etc.) and then processing them one by one.
+
+For example, listing 10000 channels evenly distributed across 3 nodes
+where network has 1ms delay takes more than 10 seconds on my
+machine. And with the proposed change listing will take almost the same
+time as it'll take to gather this info locally. E.g. in the case above
+listing now takes 0.7 second on the same machine with same 1ms delay.
+
+It works by invoking emitting_map on every node, where it should send
+info about only local items to aggregator, in an async fashion - as no
+reply from aggregator is needed.
+
+diff --git a/src/rabbit_control_main.erl b/src/rabbit_control_main.erl
+index ea9d6a2..e6b168a 100644
+--- a/src/rabbit_control_main.erl
++++ b/src/rabbit_control_main.erl
+@@ -23,7 +23,7 @@
+ sync_queue/1, cancel_sync_queue/1, become/1,
+ purge_queue/1]).
+
+--import(rabbit_misc, [rpc_call/4, rpc_call/5, rpc_call/7]).
++-import(rabbit_misc, [rpc_call/4, rpc_call/5]).
+
+ -define(EXTERNAL_CHECK_INTERVAL, 1000).
+
+@@ -595,56 +595,74 @@ action(purge_queue, Node, [Q], Opts, Inform, Timeout) ->
+
+ action(list_users, Node, [], _Opts, Inform, Timeout) ->
+ Inform("Listing users", []),
+- call(Node, {rabbit_auth_backend_internal, list_users, []},
+- rabbit_auth_backend_internal:user_info_keys(), true, Timeout);
++ call_emitter(Node, {rabbit_auth_backend_internal, list_users, []},
++ rabbit_auth_backend_internal:user_info_keys(),
++ [{timeout, Timeout}, to_bin_utf8]);
+
+ action(list_permissions, Node, [], Opts, Inform, Timeout) ->
+ VHost = proplists:get_value(?VHOST_OPT, Opts),
+ Inform("Listing permissions in vhost \"~s\"", [VHost]),
+- call(Node, {rabbit_auth_backend_internal, list_vhost_permissions, [VHost]},
+- rabbit_auth_backend_internal:vhost_perms_info_keys(), true, Timeout,
+- true);
++ call_emitter(Node, {rabbit_auth_backend_internal, list_vhost_permissions, [VHost]},
++ rabbit_auth_backend_internal:vhost_perms_info_keys(),
++ [{timeout, Timeout}, to_bin_utf8, is_escaped]);
+
+ action(list_parameters, Node, [], Opts, Inform, Timeout) ->
+ VHostArg = list_to_binary(proplists:get_value(?VHOST_OPT, Opts)),
+ Inform("Listing runtime parameters", []),
+- call(Node, {rabbit_runtime_parameters, list_formatted, [VHostArg]},
+- rabbit_runtime_parameters:info_keys(), Timeout);
++ call_emitter(Node, {rabbit_runtime_parameters, list_formatted, [VHostArg]},
++ rabbit_runtime_parameters:info_keys(),
++ [{timeout, Timeout}]);
+
+ action(list_policies, Node, [], Opts, Inform, Timeout) ->
+ VHostArg = list_to_binary(proplists:get_value(?VHOST_OPT, Opts)),
+ Inform("Listing policies", []),
+- call(Node, {rabbit_policy, list_formatted, [VHostArg]},
+- rabbit_policy:info_keys(), Timeout);
++ call_emitter(Node, {rabbit_policy, list_formatted, [VHostArg]},
++ rabbit_policy:info_keys(),
++ [{timeout, Timeout}]);
+
+ action(list_vhosts, Node, Args, _Opts, Inform, Timeout) ->
+ Inform("Listing vhosts", []),
+ ArgAtoms = default_if_empty(Args, [name]),
+- call(Node, {rabbit_vhost, info_all, []}, ArgAtoms, true, Timeout);
++ call_emitter(Node, {rabbit_vhost, info_all, []}, ArgAtoms,
++ [{timeout, Timeout}, to_bin_utf8]);
+
+ action(list_user_permissions, _Node, _Args = [], _Opts, _Inform, _Timeout) ->
+ {error_string,
+ "list_user_permissions expects a username argument, but none provided."};
+ action(list_user_permissions, Node, Args = [_Username], _Opts, Inform, Timeout) ->
+ Inform("Listing permissions for user ~p", Args),
+- call(Node, {rabbit_auth_backend_internal, list_user_permissions, Args},
+- rabbit_auth_backend_internal:user_perms_info_keys(), true, Timeout,
+- true);
++ call_emitter(Node, {rabbit_auth_backend_internal, list_user_permissions, Args},
++ rabbit_auth_backend_internal:user_perms_info_keys(),
++ [{timeout, Timeout}, to_bin_utf8, is_escaped]);
+
+ action(list_queues, Node, Args, Opts, Inform, Timeout) ->
+- [Online, Offline] = rabbit_cli:filter_opts(Opts, [?ONLINE_OPT, ?OFFLINE_OPT]),
+ Inform("Listing queues", []),
++ %% User options
++ [Online, Offline] = rabbit_cli:filter_opts(Opts, [?ONLINE_OPT, ?OFFLINE_OPT]),
+ VHostArg = list_to_binary(proplists:get_value(?VHOST_OPT, Opts)),
+ ArgAtoms = default_if_empty(Args, [name, messages]),
+- call(Node, {rabbit_amqqueue, info_all, [VHostArg, ArgAtoms, Online, Offline]},
+- ArgAtoms, Timeout);
++
++ %% Data for emission
++ Nodes = nodes_in_cluster(Node, Timeout),
++ OnlineChunks = if Online -> length(Nodes); true -> 0 end,
++ OfflineChunks = if Offline -> 1; true -> 0 end,
++ ChunksOpt = {chunks, OnlineChunks + OfflineChunks},
++ TimeoutOpt = {timeout, Timeout},
++ EmissionRef = make_ref(),
++ EmissionRefOpt = {ref, EmissionRef},
++
++ _ = Online andalso start_emission(Node, {rabbit_amqqueue, emit_info_all, [Nodes, VHostArg, ArgAtoms]},
++ [TimeoutOpt, EmissionRefOpt]),
++ _ = Offline andalso start_emission(Node, {rabbit_amqqueue, emit_info_down, [VHostArg, ArgAtoms]},
++ [TimeoutOpt, EmissionRefOpt]),
++ display_emission_result(EmissionRef, ArgAtoms, [ChunksOpt, TimeoutOpt]);
+
+ action(list_exchanges, Node, Args, Opts, Inform, Timeout) ->
+ Inform("Listing exchanges", []),
+ VHostArg = list_to_binary(proplists:get_value(?VHOST_OPT, Opts)),
+ ArgAtoms = default_if_empty(Args, [name, type]),
+- call(Node, {rabbit_exchange, info_all, [VHostArg, ArgAtoms]},
+- ArgAtoms, Timeout);
++ call_emitter(Node, {rabbit_exchange, info_all, [VHostArg, ArgAtoms]},
++ ArgAtoms, [{timeout, Timeout}]);
+
+ action(list_bindings, Node, Args, Opts, Inform, Timeout) ->
+ Inform("Listing bindings", []),
+@@ -652,27 +670,31 @@ action(list_bindings, Node, Args, Opts, Inform, Timeout) ->
+ ArgAtoms = default_if_empty(Args, [source_name, source_kind,
+ destination_name, destination_kind,
+ routing_key, arguments]),
+- call(Node, {rabbit_binding, info_all, [VHostArg, ArgAtoms]},
+- ArgAtoms, Timeout);
++ call_emitter(Node, {rabbit_binding, info_all, [VHostArg, ArgAtoms]},
++ ArgAtoms, [{timeout, Timeout}]);
+
+ action(list_connections, Node, Args, _Opts, Inform, Timeout) ->
+ Inform("Listing connections", []),
+ ArgAtoms = default_if_empty(Args, [user, peer_host, peer_port, state]),
+- call(Node, {rabbit_networking, connection_info_all, [ArgAtoms]},
+- ArgAtoms, Timeout);
++ Nodes = nodes_in_cluster(Node, Timeout),
++ call_emitter(Node, {rabbit_networking, emit_connection_info_all, [Nodes, ArgAtoms]},
++ ArgAtoms, [{timeout, Timeout}, {chunks, length(Nodes)}]);
+
+ action(list_channels, Node, Args, _Opts, Inform, Timeout) ->
+ Inform("Listing channels", []),
+ ArgAtoms = default_if_empty(Args, [pid, user, consumer_count,
+ messages_unacknowledged]),
+- call(Node, {rabbit_channel, info_all, [ArgAtoms]},
+- ArgAtoms, Timeout);
++ Nodes = nodes_in_cluster(Node, Timeout),
++ call_emitter(Node, {rabbit_channel, emit_info_all, [Nodes, ArgAtoms]}, ArgAtoms,
++ [{timeout, Timeout}, {chunks, length(Nodes)}]);
+
+ action(list_consumers, Node, _Args, Opts, Inform, Timeout) ->
+ Inform("Listing consumers", []),
+ VHostArg = list_to_binary(proplists:get_value(?VHOST_OPT, Opts)),
+- call(Node, {rabbit_amqqueue, consumers_all, [VHostArg]},
+- rabbit_amqqueue:consumer_info_keys(), Timeout);
++ Nodes = nodes_in_cluster(Node, Timeout),
++ call_emitter(Node, {rabbit_amqqueue, emit_consumers_all, [Nodes, VHostArg]},
++ rabbit_amqqueue:consumer_info_keys(),
++ [{timeout, Timeout}, {chunks, length(Nodes)}]);
+
+ action(node_health_check, Node, _Args, _Opts, Inform, Timeout) ->
+ Inform("Checking health of node ~p", [Node]),
+@@ -788,17 +810,18 @@ display_info_message_row(IsEscaped, Result, InfoItemKeys) ->
+ {X, Value} -> Value
+ end, IsEscaped) || X <- InfoItemKeys]).
+
+-display_info_message(IsEscaped) ->
++display_info_message(IsEscaped, InfoItemKeys) ->
+ fun ([], _) ->
+ ok;
+- ([FirstResult|_] = List, InfoItemKeys) when is_list(FirstResult) ->
++ ([FirstResult|_] = List, _) when is_list(FirstResult) ->
+ lists:foreach(fun(Result) ->
+ display_info_message_row(IsEscaped, Result, InfoItemKeys)
+ end,
+ List),
+ ok;
+- (Result, InfoItemKeys) ->
+- display_info_message_row(IsEscaped, Result, InfoItemKeys)
++ (Result, _) ->
++ display_info_message_row(IsEscaped, Result, InfoItemKeys),
++ ok
+ end.
+
+ display_info_list(Results, InfoItemKeys) when is_list(Results) ->
+@@ -855,7 +878,10 @@ display_call_result(Node, MFA) ->
+ end.
+
+ unsafe_rpc(Node, Mod, Fun, Args) ->
+- case rpc_call(Node, Mod, Fun, Args) of
++ unsafe_rpc(Node, Mod, Fun, Args, ?RPC_TIMEOUT).
++
++unsafe_rpc(Node, Mod, Fun, Args, Timeout) ->
++ case rpc_call(Node, Mod, Fun, Args, Timeout) of
+ {badrpc, _} = Res -> throw(Res);
+ Normal -> Normal
+ end.
+@@ -874,33 +900,42 @@ ensure_app_running(Node) ->
+ call(Node, {Mod, Fun, Args}) ->
+ rpc_call(Node, Mod, Fun, lists:map(fun list_to_binary_utf8/1, Args)).
+
+-call(Node, {Mod, Fun, Args}, InfoKeys, Timeout) ->
+- call(Node, {Mod, Fun, Args}, InfoKeys, false, Timeout, false).
++call_emitter(Node, {Mod, Fun, Args}, InfoKeys, Opts) ->
++ Ref = start_emission(Node, {Mod, Fun, Args}, Opts),
++ display_emission_result(Ref, InfoKeys, Opts).
++
++start_emission(Node, {Mod, Fun, Args}, Opts) ->
++ ToBinUtf8 = proplists:get_value(to_bin_utf8, Opts, false),
++ Timeout = proplists:get_value(timeout, Opts, infinity),
++ Ref = proplists:get_value(ref, Opts, make_ref()),
++ rabbit_control_misc:spawn_emitter_caller(
++ Node, Mod, Fun, prepare_call_args(Args, ToBinUtf8),
++ Ref, self(), Timeout),
++ Ref.
++
++display_emission_result(Ref, InfoKeys, Opts) ->
++ IsEscaped = proplists:get_value(is_escaped, Opts, false),
++ Chunks = proplists:get_value(chunks, Opts, 1),
++ Timeout = proplists:get_value(timeout, Opts, infinity),
++ EmissionStatus = rabbit_control_misc:wait_for_info_messages(
++ self(), Ref, display_info_message(IsEscaped, InfoKeys), ok, Timeout, Chunks),
++ emission_to_action_result(EmissionStatus).
++
++%% Convert rabbit_control_misc:wait_for_info_messages/6 return value
++%% into form expected by rabbit_cli:main/3.
++emission_to_action_result({ok, ok}) ->
++ ok;
++emission_to_action_result({error, Error}) ->
++ Error.
+
+-call(Node, {Mod, Fun, Args}, InfoKeys, ToBinUtf8, Timeout) ->
+- call(Node, {Mod, Fun, Args}, InfoKeys, ToBinUtf8, Timeout, false).
++prepare_call_args(Args, ToBinUtf8) ->
++ case ToBinUtf8 of
++ true -> valid_utf8_args(Args);
++ false -> Args
++ end.
+
+-call(Node, {Mod, Fun, Args}, InfoKeys, ToBinUtf8, Timeout, IsEscaped) ->
+- Args0 = case ToBinUtf8 of
+- true -> lists:map(fun list_to_binary_utf8/1, Args);
+- false -> Args
+- end,
+- Ref = make_ref(),
+- Pid = self(),
+- spawn_link(
+- fun () ->
+- case rabbit_cli:rpc_call(Node, Mod, Fun, Args0,
+- Ref, Pid, Timeout) of
+- {error, _} = Error ->
+- Pid ! {error, Error};
+- {bad_argument, _} = Error ->
+- Pid ! {error, Error};
+- _ ->
+- ok
+- end
+- end),
+- rabbit_control_misc:wait_for_info_messages(
+- Pid, Ref, InfoKeys, display_info_message(IsEscaped), Timeout).
++valid_utf8_args(Args) ->
++ lists:map(fun list_to_binary_utf8/1, Args).
+
+ list_to_binary_utf8(L) ->
+ B = list_to_binary(L),
+@@ -950,7 +985,10 @@ split_list([_]) -> exit(even_list_needed);
+ split_list([A, B | T]) -> [{A, B} | split_list(T)].
+
+ nodes_in_cluster(Node) ->
+- unsafe_rpc(Node, rabbit_mnesia, cluster_nodes, [running]).
++ unsafe_rpc(Node, rabbit_mnesia, cluster_nodes, [running], ?RPC_TIMEOUT).
++
++nodes_in_cluster(Node, Timeout) ->
++ unsafe_rpc(Node, rabbit_mnesia, cluster_nodes, [running], Timeout).
+
+ alarms_by_node(Name) ->
+ case rpc_call(Name, rabbit, status, []) of
Code Review / pti / rtp.git / blobdiff