Speed up AMQP connection and session (de)registration

 ## What?

Prior to this commit connecting 40k AMQP clients with 5 sessions each,
i.e. 200k sessions in total, took 7m55s.

After to this commit the same scenario takes 1m37s.

Additionally, prior to this commit, disconnecting all connections and sessions
at once caused the pg process to become overloaded taking ~14 minutes to
process its mailbox.

After this commit, these same deregistrations take less than 5 seconds.

To repro:
```go

package main

import (
	"context"
	"log"
	"time"

	"github.com/Azure/go-amqp"
)

func main() {
	for i := 0; i < 40_000; i++ {
		if i%1000 == 0 {
			log.Printf("opened %d connections", i)
		}
		conn, err := amqp.Dial(
			context.TODO(),
			"amqp://localhost",
			&amqp.ConnOptions{SASLType: amqp.SASLTypeAnonymous()})
		if err != nil {
			log.Fatal("open connection:", err)
		}
		for j := 0; j < 5; j++ {
			_, err = conn.NewSession(context.TODO(), nil)
			if err != nil {
				log.Fatal("begin session:", err)
			}
		}
	}
	log.Println("opened all connections")
	time.Sleep(5 * time.Hour)
}
```

 ## How?

This commit uses separate pg scopes (that is processes and ETS tables) to register
AMQP connections and AMQP sessions. Since each Pid is now its own group,
registration and deregistration is fast.

(cherry picked from commit 93d1ac9)

Author: ansd

deps/rabbit/src/rabbit.erl

@@ -29,14 +29,17 @@
          base_product_name/0,
          base_product_version/0,
          motd_file/0,
-         motd/0]).
+         motd/0,
+         pg_local_scope/1]).
 %% 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]).
 
 %%---------------------------------------------------------------------------
 %% Boot steps.
--export([maybe_insert_default_data/0, boot_delegate/0, recover/0, pg_local/0]).
+-export([maybe_insert_default_data/0, boot_delegate/0, recover/0,
+         pg_local_amqp_session/0,
+         pg_local_amqp_connection/0]).
 
 %% for tests
 -export([validate_msg_store_io_batch_size_and_credit_disc_bound/2]).
@@ -263,9 +266,15 @@
         {mfa,         {rabbit_vhosts, boot, []}},
         {requires,    notify_cluster}]}).
 
--rabbit_boot_step({pg_local,
-                   [{description, "local-only pg scope"},
-                    {mfa,         {rabbit, pg_local, []}},
+-rabbit_boot_step({pg_local_amqp_session,
+                   [{description, "local-only pg scope for AMQP sessions"},
+                    {mfa,         {rabbit, pg_local_amqp_session, []}},
+                    {requires,    kernel_ready},
+                    {enables,     core_initialized}]}).
+
+-rabbit_boot_step({pg_local_amqp_connection,
+                   [{description, "local-only pg scope for AMQP connections"},
+                    {mfa,         {rabbit, pg_local_amqp_connection, []}},
                     {requires,    kernel_ready},
                     {enables,     core_initialized}]}).
 
@@ -1115,11 +1124,18 @@ boot_delegate() ->
 -spec recover() -> 'ok'.
 
 recover() ->
-    ok = rabbit_vhost:recover(),
-    ok.
+    ok = rabbit_vhost:recover().
+
+pg_local_amqp_session() ->
+    PgScope = pg_local_scope(amqp_session),
+    rabbit_sup:start_child(pg_amqp_session, pg, [PgScope]).
+
+pg_local_amqp_connection() ->
+    PgScope = pg_local_scope(amqp_connection),
+    rabbit_sup:start_child(pg_amqp_connection, pg, [PgScope]).
 
-pg_local() ->
-    rabbit_sup:start_child(pg, [node()]).
+pg_local_scope(Prefix) ->
+    list_to_atom(io_lib:format("~s_~s", [Prefix, node()])).
 
 -spec maybe_insert_default_data() -> 'ok'.
 

deps/rabbit/src/rabbit_amqp1_0.erl

@@ -6,8 +6,6 @@
 %%
 -module(rabbit_amqp1_0).
 
--define(PROCESS_GROUP_NAME, rabbit_amqp10_connections).
-
 -export([list_local/0,
          register_connection/1]).
 
@@ -36,8 +34,11 @@ emit_connection_info_local(Items, Ref, AggregatorPid) ->
 
 -spec list_local() -> [pid()].
 list_local() ->
-    pg:get_local_members(node(), ?PROCESS_GROUP_NAME).
+    pg:which_groups(pg_scope()).
 
 -spec register_connection(pid()) -> ok.
 register_connection(Pid) ->
-    ok = pg:join(node(), ?PROCESS_GROUP_NAME, Pid).
+    ok = pg:join(pg_scope(), Pid, Pid).
+
+pg_scope() ->
+    rabbit:pg_local_scope(amqp_connection).

deps/rabbit/src/rabbit_amqp_session.erl

@@ -69,7 +69,6 @@
                    ?V_1_0_SYMBOL_MODIFIED]).
 -define(DEFAULT_EXCHANGE_NAME, <<>>).
 -define(PROTOCOL, amqp10).
--define(PROCESS_GROUP_NAME, amqp_sessions).
 -define(MAX_PERMISSION_CACHE_SIZE, 12).
 -define(HIBERNATE_AFTER, 6_000).
 -define(CREDIT_REPLY_TIMEOUT, 30_000).
@@ -373,8 +372,8 @@ init({ReaderPid, WriterPid, ChannelNum, MaxFrameSize, User, Vhost, ConnName,
                     handle_max = HandleMax0}}) ->
     process_flag(trap_exit, true),
     process_flag(message_queue_data, off_heap),
-    ok = pg:join(node(), ?PROCESS_GROUP_NAME, self()),
 
+    ok = pg:join(pg_scope(), self(), self()),
     Alarms0 = rabbit_alarm:register(self(), {?MODULE, conserve_resources, []}),
     Alarms = sets:from_list(Alarms0, [{version, 2}]),
 
@@ -439,7 +438,7 @@ terminate(_Reason, #state{incoming_links = IncomingLinks,
 
 -spec list_local() -> [pid()].
 list_local() ->
-    pg:get_local_members(node(), ?PROCESS_GROUP_NAME).
+    pg:which_groups(pg_scope()).
 
 -spec conserve_resources(pid(),
                          rabbit_alarm:resource_alarm_source(),
@@ -3432,6 +3431,9 @@ is_valid_max(Val) ->
     Val > 0 andalso
     Val =< ?UINT_MAX.
 
+pg_scope() ->
+    rabbit:pg_local_scope(amqp_session).
+
 -spec cap_credit(rabbit_queue_type:credit()) ->
     rabbit_queue_type:credit().
 cap_credit(DesiredCredit) ->

deps/rabbitmq_mqtt/src/rabbit_mqtt_sup.erl

@@ -29,7 +29,7 @@ init([{Listeners, SslListeners0}]) ->
           end,
     %% Use separate process group scope per RabbitMQ node. This achieves a local-only
     %% process group which requires less memory with millions of connections.
-    PgScope = list_to_atom(io_lib:format("~s_~s", [?PG_SCOPE, node()])),
+    PgScope = rabbit:pg_local_scope(?PG_SCOPE),
     persistent_term:put(?PG_SCOPE, PgScope),
     {ok,
      {#{strategy => one_for_all,