wip

Author: kjnilsson

deps/rabbit/src/rabbit_channel.erl

@@ -366,8 +366,9 @@ declare_fast_reply_to_v1(EncodedBin) ->
 -spec list() -> [pid()].
 
 list() ->
-    Nodes = rabbit_nodes:list_running(),
-    rabbit_misc:append_rpc_all_nodes(Nodes, rabbit_channel, list_local, [], ?RPC_TIMEOUT).
+    Nodes = rabbit_presence:list_present(),
+    rabbit_misc:append_rpc_all_nodes(Nodes, rabbit_channel, list_local,
+                                     [], ?RPC_TIMEOUT).
 
 -spec list_local() -> [pid()].
 

deps/rabbit/src/rabbit_queue_type_util.erl

@@ -12,7 +12,8 @@
          check_auto_delete/1,
          check_exclusive/1,
          check_non_durable/1,
-         run_checks/2]).
+         run_checks/2,
+         erpc_call/5]).
 
 -include_lib("rabbit_common/include/rabbit.hrl").
 -include("amqqueue.hrl").
@@ -70,3 +71,30 @@ run_checks([C | Checks], Q) ->
         Err ->
             Err
     end.
+
+-spec erpc_call(node(), module(), atom(), list(), non_neg_integer()) ->
+    term() | {error, term()}.
+erpc_call(Node, M, F, A, _Timeout)
+  when Node =:= node()  ->
+    %% Only timeout 'infinity' optimises the local call in OTP 23-25 avoiding a new process being spawned:
+    %% https://github.com/erlang/otp/blob/47f121af8ee55a0dbe2a8c9ab85031ba052bad6b/lib/kernel/src/erpc.erl#L121
+    try erpc:call(Node, M, F, A, infinity) of
+        Result ->
+            Result
+    catch
+        error:Err ->
+            {error, Err}
+    end;
+erpc_call(Node, M, F, A, Timeout) ->
+    case lists:member(Node, nodes()) of
+        true ->
+            try erpc:call(Node, M, F, A, Timeout) of
+                Result ->
+                    Result
+            catch
+                error:Err ->
+                    {error, Err}
+            end;
+        false ->
+            {error, noconnection}
+    end.

deps/rabbit/src/rabbit_quorum_queue.erl

@@ -78,7 +78,9 @@
          force_all_queues_shrink_member_to_current_member/0]).
 
 -import(rabbit_queue_type_util, [args_policy_lookup/3,
-                                 qname_to_internal_name/1]).
+                                 qname_to_internal_name/1,
+                                 erpc_call/5
+                                ]).
 
 -include_lib("stdlib/include/qlc.hrl").
 -include_lib("rabbit_common/include/rabbit.hrl").
@@ -1676,8 +1678,7 @@ format(Q, Ctx) when ?is_amqqueue(Q) ->
                   #{running_nodes := Running0} ->
                       Running0;
                   _ ->
-                      %% WARN: slow
-                      rabbit_nodes:list_running()
+                      rabbit_presence:list_present()
               end,
     Online = [N || N <- Nodes, lists:member(N, Running)],
     {_, LeaderNode} = amqqueue:get_pid(Q),
@@ -1840,31 +1841,6 @@ notify_decorators(QName, F, A) ->
             ok
     end.
 
-erpc_call(Node, M, F, A, _Timeout)
-  when Node =:= node()  ->
-    %% Only timeout 'infinity' optimises the local call in OTP 23-25 avoiding a new process being spawned:
-    %% https://github.com/erlang/otp/blob/47f121af8ee55a0dbe2a8c9ab85031ba052bad6b/lib/kernel/src/erpc.erl#L121
-    try erpc:call(Node, M, F, A, infinity) of
-        Result ->
-            Result
-    catch
-        error:Err ->
-            {error, Err}
-    end;
-erpc_call(Node, M, F, A, Timeout) ->
-    case lists:member(Node, nodes()) of
-        true ->
-            try erpc:call(Node, M, F, A, Timeout) of
-                Result ->
-                    Result
-            catch
-                error:Err ->
-                    {error, Err}
-            end;
-        false ->
-            {error, noconnection}
-    end.
-
 is_stateful() -> true.
 
 force_shrink_member_to_current_member(VHost, Name) ->

deps/rabbit/src/rabbit_stream_coordinator.erl

@@ -57,6 +57,11 @@
 -export([eval_listeners/3,
          state/0]).
 
+
+-import(rabbit_queue_type_util, [
+                                 erpc_call/5
+                                ]).
+
 -rabbit_boot_step({?MODULE,
                    [{description, "Restart stream coordinator"},
                     {mfa,         {?MODULE, recover, []}},
@@ -396,34 +401,44 @@ process_command([Server | Servers], Cmd) ->
 
 ensure_coordinator_started() ->
     Local = {?MODULE, node()},
-    AllNodes = all_coord_members(),
+    ExpectedMembers = expected_coord_members(),
     case whereis(?MODULE) of
         undefined ->
             global:set_lock(?STREAM_COORDINATOR_STARTUP),
             Nodes = case ra:restart_server(?RA_SYSTEM, Local) of
                         {error, Reason} when Reason == not_started orelse
                                              Reason == name_not_registered ->
-                            OtherNodes = all_coord_members() -- [Local],
+                            OtherNodes = ExpectedMembers -- [Local],
+                            %% this could potentially be slow if some expected
+                            %% members are on nodes that have recently terminated
+                            %% and have left a dangling TCP connection
+                            %% I suspect this rarely happens as the local coordinator
+                            %% server is started in recover/0
                             case lists:filter(
                                    fun({_, N}) ->
-                                           erpc:call(N, erlang, whereis, [?MODULE]) =/= undefined
+                                           is_pid(erpc_call(N, erlang,
+                                                            whereis, [?MODULE],
+                                                            1000))
                                    end, OtherNodes) of
                                 [] ->
                                     start_coordinator_cluster();
                                 _ ->
                                     OtherNodes
                             end;
                         ok ->
-                            AllNodes;
+                            %% TODO: it may be better to do a leader call
+                            %% here as the local member will not have caught up
+                            %% yet
+                            locally_known_members();
                         {error, {already_started, _}} ->
-                            AllNodes;
+                            locally_known_members();
                         _ ->
-                            AllNodes
+                            locally_known_members()
                     end,
             global:del_lock(?STREAM_COORDINATOR_STARTUP),
             Nodes;
         _ ->
-            AllNodes
+            locally_known_members()
     end.
 
 start_coordinator_cluster() ->
@@ -440,9 +455,15 @@ start_coordinator_cluster() ->
             []
     end.
 
-all_coord_members() ->
-    Nodes = rabbit_nodes:list_running() -- [node()],
-    [{?MODULE, Node} || Node <- [node() | Nodes]].
+present_coord_members() ->
+    Local = {?MODULE, node()},
+    Nodes = rabbit_presence:list_present(),
+    [Local] ++ [{?MODULE, Node} || Node <- [node() | Nodes]].
+
+expected_coord_members() ->
+    Local = {?MODULE, node()},
+    Nodes = rabbit_nodes:list_members(),
+    [Local] ++ [{?MODULE, Node} || Node <- [node() | Nodes]].
 
 version() -> 4.
 
@@ -681,6 +702,15 @@ all_member_nodes(Streams) ->
 tick(_Ts, _State) ->
     [{aux, maybe_resize_coordinator_cluster}].
 
+locally_known_members() ->
+    %% TODO: use ra_leaderboard and fallback if leaderboard not populated
+    case ra:members({local, {?MODULE, node()}}) of
+        {_, Members, _} ->
+            Members;
+        Err ->
+            exit({error_fetching_locally_known_coordinator_members, Err})
+    end.
+
 maybe_resize_coordinator_cluster() ->
     spawn(fun() ->
                   case ra:members({?MODULE, node()}) of

deps/rabbit/test/rabbit_stream_queue_SUITE.erl

@@ -645,9 +645,9 @@ grow_coordinator_cluster(Config) ->
     ?assertEqual({'queue.declare_ok', Q, 0, 0},
                  declare(Config, Server0, Q, [{<<"x-queue-type">>, longstr, <<"stream">>}])),
 
-    ok = rabbit_control_helper:command(stop_app, Server1),
+    % ok = rabbit_control_helper:command(stop_app, Server1),
     ok = rabbit_control_helper:command(join_cluster, Server1, [atom_to_list(Server0)], []),
-    rabbit_control_helper:command(start_app, Server1),
+    % rabbit_control_helper:command(start_app, Server1),
     %% at this point there _probably_ won't be a stream coordinator member on
     %% Server1
 
@@ -673,7 +673,8 @@ grow_coordinator_cluster(Config) ->
                       false
               end
       end, 60000),
-    rabbit_ct_broker_helpers:rpc(Config, 0, ?MODULE, delete_testcase_queue, [Q]).
+    % rabbit_ct_broker_helpers:rpc(Config, 0, ?MODULE, delete_testcase_queue, [Q]),
+    ok.
 
 shrink_coordinator_cluster(Config) ->
     [Server0, Server1, Server2] =

deps/rabbitmq_management/src/rabbit_mgmt_db.erl

@@ -258,7 +258,7 @@ handle_pre_hibernate(State) ->
     %% rabbit_mgmt_db is hibernating the odds are rabbit_event is
     %% quiescing in some way too).
     _ = rpc:multicall(
-      rabbit_nodes:list_running(), rabbit_mgmt_db_handler, gc, []),
+          rabbit_presence:list_present(), rabbit_mgmt_db_handler, gc, []),
     {hibernate, State}.
 
 format_message_queue(Opt, MQ) -> rabbit_misc:format_message_queue(Opt, MQ).

deps/rabbitmq_management/src/rabbit_mgmt_wm_queues.erl

@@ -72,10 +72,9 @@ is_authorized(ReqData, {Mode, Context}) ->
 %% Exported functions
 
 basic(ReqData) ->
-    %% rabbit_nodes:list_running/1 is a potentially slow function that performs
-    %% a cluster wide query with a reasonably long (10s) timeout.
-    %% TODO: replace with faster approximate function
-    Running = rabbit_nodes:list_running(),
+    %% rabbit_presence:list_present/1 is an approximate, "good enough" view of
+    %% the current active running cluster members.
+    Running = rabbit_presence:list_present(),
     Ctx = #{running_nodes => Running},
     FmtQ = fun (Q) -> rabbit_mgmt_format:queue(Q, Ctx) end,
     case rabbit_mgmt_util:disable_stats(ReqData) of