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/Makefile

@@ -277,7 +277,7 @@ PARALLEL_CT_SET_3_B = cluster_upgrade list_consumers_sanity_check list_queues_on
 PARALLEL_CT_SET_3_C = cli_forget_cluster_node feature_flags_v2 mc_unit message_size_limit metadata_store_migration
 PARALLEL_CT_SET_3_D = metadata_store_phase1 metrics mirrored_supervisor peer_discovery_classic_config proxy_protocol runtime_parameters unit_stats_and_metrics unit_supervisor2 unit_vm_memory_monitor
 
-PARALLEL_CT_SET_4_A = clustering_events rabbit_local_random_exchange rabbit_msg_interceptor rabbitmq_4_0_deprecations unit_pg_local unit_plugin_directories unit_plugin_versioning unit_policy_validators unit_priority_queue
+PARALLEL_CT_SET_4_A = clustering_events rabbit_local_random_exchange rabbit_msg_interceptor rabbitmq_4_0_deprecations unit_plugin_directories unit_plugin_versioning unit_policy_validators unit_priority_queue
 PARALLEL_CT_SET_4_B = per_user_connection_tracking per_vhost_connection_limit rabbit_fifo_dlx_integration rabbit_fifo_int
 PARALLEL_CT_SET_4_C = msg_size_metrics unit_msg_size_metrics per_vhost_msg_store per_vhost_queue_limit priority_queue upgrade_preparation vhost
 PARALLEL_CT_SET_4_D = per_user_connection_channel_tracking product_info publisher_confirms_parallel queue_type rabbitmq_queues_cli_integration rabbitmqctl_integration rabbitmqctl_shutdown routing rabbit_amqqueue

deps/rabbit/src/pg_local.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.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_channel.erl

@@ -41,7 +41,7 @@ force_event_refresh(Ref) ->
 -spec list_local() -> [pid()].
 
 list_local() ->
-    pg_local:get_members(rabbit_direct).
+    pg:get_local_members(node(), rabbit_direct).
 
 -spec list() -> [pid()].
 
@@ -204,7 +204,7 @@ connect1(User = #user{username = Username}, VHost, Protocol, Pid, Infos) ->
             AuthzContext = proplists:get_value(variable_map, Infos, #{}),
             try rabbit_access_control:check_vhost_access(User, VHost,
                                                {ip, PeerHost}, AuthzContext) of
-                ok -> ok = pg_local:join(rabbit_direct, Pid),
+                ok -> ok = pg:join(node(), rabbit_direct, Pid),
                       rabbit_core_metrics:connection_created(Pid, Infos),
                       rabbit_event:notify(connection_created, Infos),
                       _ = rabbit_alarm:register(
@@ -252,7 +252,7 @@ start_channel(Number, ClientChannelPid, ConnPid, ConnName, Protocol,
 -spec disconnect(pid(), rabbit_event:event_props()) -> 'ok'.
 
 disconnect(Pid, Infos) ->
-    pg_local:leave(rabbit_direct, Pid),
+    pg:leave(node(), rabbit_direct, Pid),
     rabbit_core_metrics:connection_closed(Pid),
     rabbit_event:notify(connection_closed, Infos).