v9-0004-Add-a-GUC-max_apply_bgworkers_per_subscription-to.patch

From 2383a430c4e6f811bfe2ad0e40f5c4ba092c6bed Mon Sep 17 00:00:00 2001
From: wangw <wangw.fnst@fujitsu.com>
Date: Wed, 8 Jun 2022 11:52:54 +0800
Subject: [PATCH v9 4/4] Add a GUC "max_apply_bgworkers_per_subscription" to
 control parallelism.

This GUC controls how many apply background workers can be launched per
subscription.
---
 doc/src/sgml/config.sgml                      | 25 +++++++++++++++++++
 .../replication/logical/applybgwroker.c       |  9 +++++++
 src/backend/replication/logical/launcher.c    | 25 +++++++++++++++++++
 src/backend/utils/misc/guc.c                  | 12 +++++++++
 src/backend/utils/misc/postgresql.conf.sample |  1 +
 src/include/replication/logicallauncher.h     |  1 +
 src/include/replication/worker_internal.h     |  1 +
 7 files changed, 74 insertions(+)

diff --git a/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml
index 5b7ce6531d..ce7f6827f2 100644
--- a/doc/src/sgml/config.sgml
+++ b/doc/src/sgml/config.sgml
@@ -4970,6 +4970,31 @@ ANY <replaceable class="parameter">num_sync</replaceable> ( <replaceable class="
       </listitem>
      </varlistentry>
 
+     <varlistentry id="guc-max-apply-bgworkers-per-subscription" xreflabel="max_apply_bgworkers_per_subscription">
+      <term><varname>max_apply_bgworkers_per_subscription</varname> (<type>integer</type>)
+      <indexterm>
+       <primary><varname>max_apply_bgworkers_per_subscription</varname> configuration parameter</primary>
+      </indexterm>
+      </term>
+      <listitem>
+       <para>
+        Maximum number of apply background workers per subscription. This
+        parameter controls the amount of parallelism of the streaming of
+        in-progress transactions if we set subscription option
+        <literal>streaming</literal> to <literal>apply</literal>.
+       </para>
+       <para>
+        The apply background workers are taken from the pool defined by
+        <varname>max_logical_replication_workers</varname>.
+       </para>
+       <para>
+        The default value is 3. This parameter can only be set in the
+        <filename>postgresql.conf</filename> file or on the server command
+        line.
+       </para>
+      </listitem>
+     </varlistentry>
+
      </variablelist>
     </sect2>
 
diff --git a/src/backend/replication/logical/applybgwroker.c b/src/backend/replication/logical/applybgwroker.c
index 0f9081d62f..9d08261906 100644
--- a/src/backend/replication/logical/applybgwroker.c
+++ b/src/backend/replication/logical/applybgwroker.c
@@ -21,6 +21,7 @@
 #include "mb/pg_wchar.h"
 #include "pgstat.h"
 #include "postmaster/interrupt.h"
+#include "replication/logicallauncher.h"
 #include "replication/logicalworker.h"
 #include "replication/origin.h"
 #include "replication/walreceiver.h"
@@ -571,9 +572,17 @@ apply_bgworker_setup(void)
 	MemoryContext oldcontext;
 	bool		launched;
 	ApplyBgworkerState *wstate;
+	int			napplyworkers;
 
 	elog(DEBUG1, "setting up apply worker #%u", list_length(ApplyWorkersList) + 1);
 
+	/* Check If there are free worker slot(s) */
+	LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
+	napplyworkers = logicalrep_apply_background_worker_count(MyLogicalRepWorker->subid);
+	LWLockRelease(LogicalRepWorkerLock);
+	if (napplyworkers >= max_apply_bgworkers_per_subscription)
+		return NULL;
+
 	oldcontext = MemoryContextSwitchTo(ApplyContext);
 
 	wstate = (ApplyBgworkerState *) palloc0(sizeof(ApplyBgworkerState));
diff --git a/src/backend/replication/logical/launcher.c b/src/backend/replication/logical/launcher.c
index 201f51232e..dd68651314 100644
--- a/src/backend/replication/logical/launcher.c
+++ b/src/backend/replication/logical/launcher.c
@@ -54,6 +54,7 @@
 
 int			max_logical_replication_workers = 4;
 int			max_sync_workers_per_subscription = 2;
+int			max_apply_bgworkers_per_subscription = 3;
 
 LogicalRepWorker *MyLogicalRepWorker = NULL;
 
@@ -736,6 +737,30 @@ logicalrep_sync_worker_count(Oid subid)
 	return res;
 }
 
+/*
+ * Count the number of registered (not necessarily running) apply background
+ * worker for a subscription.
+ */
+int
+logicalrep_apply_background_worker_count(Oid subid)
+{
+	int			i;
+	int			res = 0;
+
+	Assert(LWLockHeldByMe(LogicalRepWorkerLock));
+
+	/* Search for attached worker for a given subscription id. */
+	for (i = 0; i < max_logical_replication_workers; i++)
+	{
+		LogicalRepWorker *w = &LogicalRepCtx->workers[i];
+
+		if (w->subid == subid && w->subworker)
+			res++;
+	}
+
+	return res;
+}
+
 /*
  * ApplyLauncherShmemSize
  *		Compute space needed for replication launcher shared memory
diff --git a/src/backend/utils/misc/guc.c b/src/backend/utils/misc/guc.c
index 55d41ae7d6..539ec07ff5 100644
--- a/src/backend/utils/misc/guc.c
+++ b/src/backend/utils/misc/guc.c
@@ -3220,6 +3220,18 @@ static struct config_int ConfigureNamesInt[] =
 		NULL, NULL, NULL
 	},
 
+	{
+		{"max_apply_bgworkers_per_subscription",
+			PGC_SIGHUP,
+			REPLICATION_SUBSCRIBERS,
+			gettext_noop("Maximum number of apply backgrand workers per subscription."),
+			NULL,
+		},
+		&max_apply_bgworkers_per_subscription,
+		3, 0, MAX_BACKENDS,
+		NULL, NULL, NULL
+	},
+
 	{
 		{"log_rotation_age", PGC_SIGHUP, LOGGING_WHERE,
 			gettext_noop("Sets the amount of time to wait before forcing "
diff --git a/src/backend/utils/misc/postgresql.conf.sample b/src/backend/utils/misc/postgresql.conf.sample
index 48ad80cf2e..b71340549d 100644
--- a/src/backend/utils/misc/postgresql.conf.sample
+++ b/src/backend/utils/misc/postgresql.conf.sample
@@ -360,6 +360,7 @@
 #max_logical_replication_workers = 4	# taken from max_worker_processes
 					# (change requires restart)
 #max_sync_workers_per_subscription = 2	# taken from max_logical_replication_workers
+#max_apply_bgworkers_per_subscription = 3	# taken from max_logical_replication_workers
 
 
 #------------------------------------------------------------------------------
diff --git a/src/include/replication/logicallauncher.h b/src/include/replication/logicallauncher.h
index f1e2821e25..ac8ef94381 100644
--- a/src/include/replication/logicallauncher.h
+++ b/src/include/replication/logicallauncher.h
@@ -14,6 +14,7 @@
 
 extern PGDLLIMPORT int max_logical_replication_workers;
 extern PGDLLIMPORT int max_sync_workers_per_subscription;
+extern PGDLLIMPORT int max_apply_bgworkers_per_subscription;
 
 extern void ApplyLauncherRegister(void);
 extern void ApplyLauncherMain(Datum main_arg);
diff --git a/src/include/replication/worker_internal.h b/src/include/replication/worker_internal.h
index deb80cdb19..dd49d8e2ec 100644
--- a/src/include/replication/worker_internal.h
+++ b/src/include/replication/worker_internal.h
@@ -161,6 +161,7 @@ extern void logicalrep_worker_wakeup(Oid subid, Oid relid);
 extern void logicalrep_worker_wakeup_ptr(LogicalRepWorker *worker);
 
 extern int	logicalrep_sync_worker_count(Oid subid);
+extern int	logicalrep_apply_background_worker_count(Oid subid);
 
 extern void ReplicationOriginNameForTablesync(Oid suboid, Oid relid,
 											  char *originname, int szorgname);
-- 
2.23.0.windows.1