Switch from pg_local to pg

deps/rabbit/Makefile

@@ -249,7 +249,7 @@ define ct_master.erl
 endef
 
 PARALLEL_CT_SET_1_A = unit_rabbit_ssl unit_cluster_formation_locking_mocks unit_cluster_formation_sort_nodes unit_collections unit_config_value_encryption unit_connection_tracking
-PARALLEL_CT_SET_1_B = amqp_address amqp_auth amqp_filter_prop amqp_filter_sql amqp_filter_sql_unit amqp_dotnet server_named_queue_prefix_prop signal_handling single_active_consumer unit_access_control_authn_authz_context_propagation unit_access_control_credential_validation unit_amqp091_content_framing unit_amqp091_server_properties unit_app_management
+PARALLEL_CT_SET_1_B = amqp_address amqp_auth amqp_filter_prop amqp_filter_sql amqp_filter_sql_unit amqp_dotnet server_named_queue_prefix_prop signal_handling single_active_consumer unit_access_control_authn_authz_context_propagation unit_access_control_credential_validation unit_amqp091_content_framing unit_amqp091_server_properties unit_app_management rabbit_direct
 PARALLEL_CT_SET_1_C = amqp_proxy_protocol amqpl_consumer_ack bindings rabbit_db_maintenance rabbit_db_msup rabbit_db_policy rabbit_db_queue rabbit_db_topic_exchange cluster_limit cluster_minority default_queue_type_prop term_to_binary_compat_prop topic_permission unicode unit_access_control user_tags_count_limit
 PARALLEL_CT_SET_1_D = amqqueue_backward_compatibility channel_interceptor channel_operation_timeout classic_queue_prop config_schema peer_discovery_dns peer_discovery_tmp_hidden_node per_node_limit per_user_connection_channel_limit server_named_queue_prefix
 
@@ -263,7 +263,7 @@ PARALLEL_CT_SET_3_B = list_consumers_sanity_check list_queues_online_and_offline
 PARALLEL_CT_SET_3_C = cli_forget_cluster_node mc_unit message_size_limit
 PARALLEL_CT_SET_3_D = metrics mirrored_supervisor proxy_protocol runtime_parameters unit_rabbit_vm 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_int unit_default_queue_type
 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 queue_type rabbitmq_queues_cli_integration rabbitmq_streams_cli_integration rabbitmqctl_integration rabbitmqctl_shutdown routing rabbit_amqqueue

deps/rabbit/src/rabbit.erl

@@ -30,7 +30,10 @@
          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,
+         pg_scope_non_amqp_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]).
@@ -39,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,
+         pg_local_non_amqp_connection/0,
+         prevent_startup_if_node_was_reset/0]).
 
 -rabbit_boot_step({pre_boot, [{description, "rabbit boot start"}]}).
 
@@ -286,11 +293,29 @@
                     {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}]}).
+
+-rabbit_boot_step({pg_local_non_amqp_connection,
+                   [{description, "local-only pg scope for non-AMQP connections"},
+                    {mfa,         {rabbit, pg_local_non_amqp_connection, []}},
+                    {requires,    kernel_ready},
+                    {enables,     core_initialized}]}).
+
 %%---------------------------------------------------------------------------
 
 -include_lib("rabbit_common/include/rabbit.hrl").
@@ -1122,9 +1147,33 @@ 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),
+    persistent_term:put(pg_scope_amqp091_channel, PgScope),
+    rabbit_sup:start_child(pg_amqp091_channel, pg, [PgScope]).
+
+pg_local_amqp091_connection() ->
+    PgScope = pg_local_scope(amqp091_connection),
+    persistent_term:put(pg_scope_amqp091_connection, PgScope),
+    rabbit_sup:start_child(pg_amqp091_connection, pg, [PgScope]).
+
+pg_local_non_amqp_connection() ->
+    PgScope = pg_local_scope(non_amqp_connection),
+    persistent_term:put(pg_scope_non_amqp_connection, PgScope),
+    rabbit_sup:start_child(pg_non_amqp_connection, pg, [PgScope]).
+
 pg_local_scope(Prefix) ->
     list_to_atom(io_lib:format("~s_~s", [Prefix, node()])).
 
+pg_scope_amqp091_channel() ->
+    persistent_term:get(pg_scope_amqp091_channel).
+
+pg_scope_amqp091_connection() ->
+    persistent_term:get(pg_scope_amqp091_connection).
+
+pg_scope_non_amqp_connection() ->
+    persistent_term:get(pg_scope_non_amqp_connection).
+
 -spec update_cluster_tags() -> 'ok'.
 
 update_cluster_tags() ->

deps/rabbit/src/rabbit_channel.erl

@@ -307,7 +307,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().
 
@@ -425,6 +427,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, User, VHost,
       Capabilities, CollectorPid, LimiterPid, AmqpParams]) ->
     process_flag(trap_exit, true),
@@ -433,7 +439,7 @@ init([Channel, ReaderPid, WriterPid, ConnPid, ConnName, 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
@@ -732,7 +738,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()],