Switch from pg_local to pg

pg_local was 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 the node() as the local scope, so we maintain
isolation between nodes and avoid local groups from
being synchronized between nodes.

Author: mkuratczyk

deps/rabbit/src/rabbit.erl

@@ -40,7 +40,9 @@
 %% 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,
+         start_pg_local/0,
+         prevent_startup_if_node_was_reset/0]).
 
 -rabbit_boot_step({pre_boot, [{description, "rabbit boot start"}]}).
 
@@ -157,6 +159,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,
@@ -1146,6 +1154,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]).

deps/rabbit/src/rabbit_channel.erl

@@ -304,7 +304,7 @@ 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
+    case lists:member(Pid, pg:get_local_members(node(), rabbit_channels)) of
         true  -> gen_server2:cast(Pid, {deliver_reply, Key, Message});
         false -> ok
     end.
@@ -318,7 +318,7 @@ list() ->
 -spec list_local() -> [pid()].
 
 list_local() ->
-    pg_local:get_members(rabbit_channels).
+    pg:get_local_members(node(), rabbit_channels).
 
 -spec info_keys() -> rabbit_types:info_keys().
 
@@ -444,7 +444,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(node(), rabbit_channels, self()),
     Flow = case rabbit_misc:get_env(rabbit, classic_queue_flow_control, true) of
              true   -> flow;
              false  -> noflow
@@ -783,7 +783,7 @@ terminate(_Reason,
                       queue_states = QueueCtxs}) ->
     rabbit_queue_type:close(QueueCtxs),
     {_Res, _State1} = notify_queues(State),
-    pg_local:leave(rabbit_channels, self()),
+    pg:leave(node(), rabbit_channels, 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

@@ -463,11 +463,11 @@ node_client_listeners(Node) ->
 
 -spec register_connection(pid()) -> ok.
 
-register_connection(Pid) -> pg_local:join(rabbit_connections, Pid).
+register_connection(Pid) -> pg:join(node(), rabbit_connections, Pid).
 
 -spec unregister_connection(pid()) -> ok.
 
-unregister_connection(Pid) -> pg_local:leave(rabbit_connections, Pid).
+unregister_connection(Pid) -> pg:leave(node(), rabbit_connections, Pid).
 
 -spec connections() -> [rabbit_types:connection()].
 connections() ->
@@ -476,17 +476,17 @@ connections() ->
 
 -spec local_connections() -> [rabbit_types:connection()].
 local_connections() ->
-    Amqp091Pids = pg_local:get_members(rabbit_connections),
+    Amqp091Pids = pg:get_local_members(node(), rabbit_connections),
     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 +496,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).
+    lists:member(Pid, pg:get_local_members(node(), rabbit_channels)).
 
 handle_event(QName, {deliver, Msg}, #?STATE{name = QName,
                                             ctag = Ctag,

deps/rabbit/test/proxy_protocol_SUITE.erl

@@ -110,8 +110,8 @@ proxy_protocol_v2_local(Config) ->
     ok.
 
 connection_name() ->
-    ?awaitMatch([_], pg_local:get_members(rabbit_connections), 30000),
-    [Pid] = pg_local:get_members(rabbit_connections),
+    ?awaitMatch([_], pg:get_local_members(node(), rabbit_connections), 30000),
+    [Pid] = pg:get_local_members(node(), rabbit_connections),
     {dictionary, Dict} = process_info(Pid, dictionary),
     {process_name, {rabbit_reader, ConnectionName}} = lists:keyfind(process_name, 1, Dict),
     ConnectionName.
@@ -120,5 +120,5 @@ wait_for_connection_close(Config) ->
     ?awaitMatch(
        [],
        rabbit_ct_broker_helpers:rpc(
-         Config, 0, pg_local, get_members, [rabbit_connnections]),
+         Config, 0, pg, get_local_members, [node(), rabbit_connections]),
        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

deps/rabbitmq_stream/src/rabbit_stream.erl

@@ -144,7 +144,7 @@ kill_connection(ConnectionName) ->
                      after 1000 -> ok
                      end
                   end,
-                  pg_local:get_members(rabbit_stream_connections)).
+                  pg:get_local_members(node(), rabbit_stream_connections)).
 
 emit_connection_info_all(Nodes, Items, Ref, AggregatorPid) ->
     Pids =

deps/rabbitmq_stream/src/rabbit_stream_reader.erl

@@ -473,7 +473,7 @@ transition_to_opened(Transport,
     % TODO remove registration to rabbit_stream_connections
     % just meant to be able to close the connection remotely
     % should be possible once the connections are available in ctl list_connections
-    pg_local:join(rabbit_stream_connections, self()),
+    pg:join(node(), rabbit_stream_connections, self()),
     Connection1 =
         rabbit_event:init_stats_timer(NewConnection,
                                       #stream_connection.stats_timer),