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

From a975b33b26147b039524c26789cab4eefd18e4aa Mon Sep 17 00:00:00 2001
From: wangw <wangw.fnst@fujitsu.com>
Date: Mon, 30 May 2022 13:42:41 +0800
Subject: [PATCH v7 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 +++++++++++++++++++
 src/backend/replication/logical/launcher.c    | 25 +++++++++++++++++++
 src/backend/replication/logical/worker.c      |  9 +++++++
 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/launcher.c b/src/backend/replication/logical/launcher.c
index 58337bef91..dd75e16cbf 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;
 
@@ -735,6 +736,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/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index 40c811aa0e..1ba4df44b6 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -180,6 +180,7 @@
 #include "postmaster/walwriter.h"
 #include "replication/decode.h"
 #include "replication/logical.h"
+#include "replication/logicallauncher.h"
 #include "replication/logicalproto.h"
 #include "replication/logicalrelation.h"
 #include "replication/logicalworker.h"
@@ -4989,9 +4990,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/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 1feafccaf6..f1b9a242c6 100644
--- a/src/include/replication/worker_internal.h
+++ b/src/include/replication/worker_internal.h
@@ -95,6 +95,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