Switch from pg_local to pg

pg_local is based on the pg2 module, which got removed
from Erlang/OTP years ago. It was replaced by the more
efficient pg module, so let's use it directly.

We use node-local scopes and single-item groups so that
(de)registration is fast, but we can list all connections/channels
(by listing all groups in the scope).

Author: mkuratczyk

deps/rabbit/src/rabbit.erl

@@ -31,7 +31,9 @@
          base_product_version/0,
          motd_file/0,
          motd/0,
-         pg_local_scope/1]).
+         pg_local_scope/1,
+         pg_scope_amqp091_channel/0,
+         pg_scope_amqp091_connection/0]).
 %% For CLI, testing and mgmt-agent.
 -export([set_log_level/1, log_locations/0, config_files/0]).
 -export([is_booted/1, is_booted/0, is_booting/1, is_booting/0]).
@@ -40,7 +42,11 @@
 %% Boot steps.
 -export([update_cluster_tags/0, maybe_insert_default_data/0, boot_delegate/0, recover/0,
          pg_local_amqp_session/0,
-         pg_local_amqp_connection/0, prevent_startup_if_node_was_reset/0]).
+         pg_local_amqp_connection/0,
+         pg_local_amqp091_channel/0,
+         pg_local_amqp091_connection/0,
+         start_pg_local/0,
+         prevent_startup_if_node_was_reset/0]).
 
 -rabbit_boot_step({pre_boot, [{description, "rabbit boot start"}]}).
 
@@ -157,6 +163,12 @@
                    [{description, "kernel ready"},
                     {requires,    external_infrastructure}]}).
 
+-rabbit_boot_step({pg_local,
+                   [{description, "local-only pg scope"},
+                    {mfa,         {rabbit, start_pg_local, []}},
+                    {requires,    kernel_ready},
+                    {enables,     core_initialized}]}).
+
 -rabbit_boot_step({guid_generator,
                    [{description, "guid generator"},
                     {mfa,         {rabbit_sup, start_restartable_child,
@@ -292,11 +304,23 @@
                     {enables,     core_initialized}]}).
 
 -rabbit_boot_step({pg_local_amqp_connection,
-                   [{description, "local-only pg scope for AMQP connections"},
+                   [{description, "local-only pg scope for AMQP 1.0 connections"},
                     {mfa,         {rabbit, pg_local_amqp_connection, []}},
                     {requires,    kernel_ready},
                     {enables,     core_initialized}]}).
 
+-rabbit_boot_step({pg_local_amqp091_channel,
+                   [{description, "local-only pg scope for AMQP 0-9-1 channels"},
+                    {mfa,         {rabbit, pg_local_amqp091_channel, []}},
+                    {requires,    kernel_ready},
+                    {enables,     core_initialized}]}).
+
+-rabbit_boot_step({pg_local_amqp091_connection,
+                   [{description, "local-only pg scope for AMQP 0-9-1 connections"},
+                    {mfa,         {rabbit, pg_local_amqp091_connection, []}},
+                    {requires,    kernel_ready},
+                    {enables,     core_initialized}]}).
+
 %%---------------------------------------------------------------------------
 
 -include_lib("rabbit_common/include/rabbit.hrl").
@@ -1146,6 +1170,9 @@ boot_delegate() ->
 recover() ->
     ok = rabbit_vhost:recover().
 
+start_pg_local() ->
+    rabbit_sup:start_child(pg_local_scope, pg, [node()]).
+
 pg_local_amqp_session() ->
     PgScope = pg_local_scope(amqp_session),
     rabbit_sup:start_child(pg_amqp_session, pg, [PgScope]).
@@ -1154,9 +1181,23 @@ pg_local_amqp_connection() ->
     PgScope = pg_local_scope(amqp_connection),
     rabbit_sup:start_child(pg_amqp_connection, pg, [PgScope]).
 
+pg_local_amqp091_channel() ->
+    PgScope = pg_local_scope(amqp091_channel),
+    rabbit_sup:start_child(pg_amqp091_channel, pg, [PgScope]).
+
+pg_local_amqp091_connection() ->
+    PgScope = pg_local_scope(amqp091_connection),
+    rabbit_sup:start_child(pg_amqp091_connection, pg, [PgScope]).
+
 pg_local_scope(Prefix) ->
     list_to_atom(io_lib:format("~s_~s", [Prefix, node()])).
 
+pg_scope_amqp091_channel() ->
+    pg_local_scope(amqp091_channel).
+
+pg_scope_amqp091_connection() ->
+    pg_local_scope(amqp091_connection).
+
 -spec update_cluster_tags() -> 'ok'.
 
 update_cluster_tags() ->

deps/rabbit/src/rabbit_channel.erl

@@ -304,9 +304,9 @@ send_command(Pid, Msg) ->
 %% Delete this function when feature flag rabbitmq_4.2.0 becomes required.
 -spec deliver_reply_local(pid(), binary(), mc:state()) -> ok.
 deliver_reply_local(Pid, Key, Message) ->
-    case pg_local:in_group(rabbit_channels, Pid) of
-        true  -> gen_server2:cast(Pid, {deliver_reply, Key, Message});
-        false -> ok
+    case pg:get_local_members(pg_scope(), Pid) of
+        [] -> ok;
+        _  -> gen_server2:cast(Pid, {deliver_reply, Key, Message})
     end.
 
 -spec list() -> [pid()].
@@ -318,7 +318,9 @@ list() ->
 -spec list_local() -> [pid()].
 
 list_local() ->
-    pg_local:get_members(rabbit_channels).
+    try pg:which_groups(pg_scope())
+    catch error:badarg -> []
+    end.
 
 -spec info_keys() -> rabbit_types:info_keys().
 
@@ -436,6 +438,10 @@ update_user_state(Pid, UserState) when is_pid(Pid) ->
 
 %%---------------------------------------------------------------------------
 
+-spec pg_scope() -> atom().
+pg_scope() ->
+    rabbit:pg_scope_amqp091_channel().
+
 init([Channel, ReaderPid, WriterPid, ConnPid, ConnName, Protocol, User, VHost,
       Capabilities, CollectorPid, LimiterPid, AmqpParams]) ->
     process_flag(trap_exit, true),
@@ -444,7 +450,7 @@ init([Channel, ReaderPid, WriterPid, ConnPid, ConnName, Protocol, User, VHost,
 
     ?LG_PROCESS_TYPE(channel),
     ?store_proc_name({ConnName, Channel}),
-    ok = pg_local:join(rabbit_channels, self()),
+    ok = pg:join(pg_scope(), self(), self()),
     Flow = case rabbit_misc:get_env(rabbit, classic_queue_flow_control, true) of
              true   -> flow;
              false  -> noflow
@@ -783,7 +789,7 @@ terminate(_Reason,
                       queue_states = QueueCtxs}) ->
     rabbit_queue_type:close(QueueCtxs),
     {_Res, _State1} = notify_queues(State),
-    pg_local:leave(rabbit_channels, self()),
+    pg:leave(pg_scope(), self(), self()),
     rabbit_event:if_enabled(State, #ch.stats_timer,
                             fun() -> emit_stats(State) end),
     [delete_stats(Tag) || {Tag, _} <- get()],

deps/rabbit/src/rabbit_networking.erl

@@ -461,13 +461,18 @@ node_client_listeners(Node) ->
                          end, Xs)
     end.
 
+pg_scope_amqp091_connection() ->
+    rabbit:pg_scope_amqp091_connection().
+
 -spec register_connection(pid()) -> ok.
 
-register_connection(Pid) -> pg_local:join(rabbit_connections, Pid).
+register_connection(Pid) ->
+    pg:join(pg_scope_amqp091_connection(), Pid, Pid).
 
 -spec unregister_connection(pid()) -> ok.
 
-unregister_connection(Pid) -> pg_local:leave(rabbit_connections, Pid).
+unregister_connection(Pid) ->
+    pg:leave(pg_scope_amqp091_connection(), Pid, Pid).
 
 -spec connections() -> [rabbit_types:connection()].
 connections() ->
@@ -476,17 +481,19 @@ connections() ->
 
 -spec local_connections() -> [rabbit_types:connection()].
 local_connections() ->
-    Amqp091Pids = pg_local:get_members(rabbit_connections),
+    Amqp091Pids = try pg:which_groups(pg_scope_amqp091_connection())
+                  catch error:badarg -> []
+                  end,
     Amqp10Pids = rabbit_amqp1_0:list_local(),
     Amqp10Pids ++ Amqp091Pids.
 
 -spec register_non_amqp_connection(pid()) -> ok.
 
-register_non_amqp_connection(Pid) -> pg_local:join(rabbit_non_amqp_connections, Pid).
+register_non_amqp_connection(Pid) -> pg:join(node(), rabbit_non_amqp_connections, Pid).
 
 -spec unregister_non_amqp_connection(pid()) -> ok.
 
-unregister_non_amqp_connection(Pid) -> pg_local:leave(rabbit_non_amqp_connections, Pid).
+unregister_non_amqp_connection(Pid) -> pg:leave(node(), rabbit_non_amqp_connections, Pid).
 
 -spec non_amqp_connections() -> [rabbit_types:connection()].
 
@@ -496,7 +503,7 @@ non_amqp_connections() ->
 
 -spec local_non_amqp_connections() -> [rabbit_types:connection()].
 local_non_amqp_connections() ->
-  pg_local:get_members(rabbit_non_amqp_connections).
+  pg:get_local_members(node(), rabbit_non_amqp_connections).
 
 -spec connection_info(rabbit_types:connection(), rabbit_types:info_keys()) ->
           rabbit_types:infos().

deps/rabbit/src/rabbit_volatile_queue.erl

@@ -235,7 +235,7 @@ local_call(Pid, Request) ->
 
 is_local(Pid) ->
     rabbit_amqp_session:is_local(Pid) orelse
-    pg_local:in_group(rabbit_channels, Pid).
+    pg:get_local_members(rabbit:pg_scope_amqp091_channel(), Pid) =/= [].
 
 handle_event(QName, {deliver, Msg}, #?STATE{name = QName,
                                             ctag = Ctag,

deps/rabbit/test/proxy_protocol_SUITE.erl

@@ -110,15 +110,25 @@ proxy_protocol_v2_local(Config) ->
     ok.
 
 connection_name() ->
-    ?awaitMatch([_], pg_local:get_members(rabbit_connections), 30000),
-    [Pid] = pg_local:get_members(rabbit_connections),
+    Scope = rabbit:pg_scope_amqp091_connection(),
+    GetGroups = fun() ->
+                    try pg:which_groups(Scope)
+                    catch error:badarg -> []
+                    end
+                end,
+    ?awaitMatch([_], GetGroups(), 30000),
+    [Pid] = GetGroups(),
     {dictionary, Dict} = process_info(Pid, dictionary),
     {process_name, {rabbit_reader, ConnectionName}} = lists:keyfind(process_name, 1, Dict),
     ConnectionName.
 
 wait_for_connection_close(Config) ->
     ?awaitMatch(
        [],
-       rabbit_ct_broker_helpers:rpc(
-         Config, 0, pg_local, get_members, [rabbit_connnections]),
+       begin
+           Scope = rabbit_ct_broker_helpers:rpc(Config, 0, rabbit, pg_scope_amqp091_connection, []),
+           try rabbit_ct_broker_helpers:rpc(Config, 0, pg, which_groups, [Scope])
+           catch error:badarg -> []
+           end
+       end,
        30000).

deps/rabbit/test/quorum_queue_SUITE.erl

@@ -2214,8 +2214,17 @@ cleanup_queue_state_on_channel_after_publish(Config) ->
     [NCh1, NCh2] = rpc:call(Server, rabbit_channel, list, []),
     %% Check the channel state contains the state for the quorum queue on
     %% channel 1 and 2
-    wait_for_cleanup(Server, NCh1, 0),
-    wait_for_cleanup(Server, NCh2, 1),
+    %% Note: pg:get_local_members doesn't guarantee order, so we need to identify
+    %% which channel has queue state
+    {ChWithoutState, ChWithState} = case length(rpc:call(Server,
+                                                         rabbit_channel,
+                                                         list_queue_states,
+                                                         [NCh1])) of
+        0 -> {NCh1, NCh2};
+        1 -> {NCh2, NCh1}
+    end,
+    wait_for_cleanup(Server, ChWithoutState, 0),
+    wait_for_cleanup(Server, ChWithState, 1),
     %% then delete the queue and wait for the process to terminate
     ?assertMatch(#'queue.delete_ok'{},
                  amqp_channel:call(Ch1, #'queue.delete'{queue = QQ})),
@@ -2225,8 +2234,8 @@ cleanup_queue_state_on_channel_after_publish(Config) ->
                                           [?SUPNAME]))
       end, 30000),
     %% Check that all queue states have been cleaned
-    wait_for_cleanup(Server, NCh2, 0),
-    wait_for_cleanup(Server, NCh1, 0).
+    wait_for_cleanup(Server, ChWithState, 0),
+    wait_for_cleanup(Server, ChWithoutState, 0).
 
 cleanup_queue_state_on_channel_after_subscribe(Config) ->
     %% Declare/delete the queue and publish in one channel, while consuming on a