v62-0006-Stop-extra-worker-if-GUC-was-changed.patch

From fac14b4ffbdbf1d759a84260c1248f7a13cf42f2 Mon Sep 17 00:00:00 2001
From: sherlockcpp <sherlockcpp@foxmail.com>
Date: Sat, 17 Dec 2022 20:43:21 +0800
Subject: [PATCH v62 6/8] Stop extra worker if GUC was changed

If the max_parallel_apply_workers_per_subscription is changed to a
lower value, try to stop free workers in the pool to keep the number of
workers lower than half of the max_parallel_apply_workers_per_subscription 

---
 .../replication/logical/applyparallelworker.c      | 73 ++++++++++++++++++----
 src/backend/replication/logical/worker.c           |  7 +++
 src/include/replication/worker_internal.h          |  1 +
 3 files changed, 69 insertions(+), 12 deletions(-)

diff --git a/src/backend/replication/logical/applyparallelworker.c b/src/backend/replication/logical/applyparallelworker.c
index 593d850..623e551 100644
--- a/src/backend/replication/logical/applyparallelworker.c
+++ b/src/backend/replication/logical/applyparallelworker.c
@@ -541,6 +541,25 @@ pa_find_worker(TransactionId xid)
 }
 
 /*
+ * Stop the given parallel apply worker and free the corresponding info.
+ */
+static void
+pa_stop_worker(ParallelApplyWorkerInfo *winfo)
+{
+	int			slot_no;
+	uint16		generation;
+
+	SpinLockAcquire(&winfo->shared->mutex);
+	generation = winfo->shared->logicalrep_worker_generation;
+	slot_no = winfo->shared->logicalrep_worker_slot_no;
+	SpinLockRelease(&winfo->shared->mutex);
+
+	logicalrep_pa_worker_stop(slot_no, generation);
+
+	pa_free_worker_info(winfo);
+}
+
+/*
  * Makes the worker available for reuse.
  *
  * This removes the parallel apply worker entry from the hash table so that it
@@ -585,18 +604,7 @@ pa_free_worker(ParallelApplyWorkerInfo *winfo)
 		list_length(ParallelApplyWorkerPool) >
 		(max_parallel_apply_workers_per_subscription / 2))
 	{
-		int			slot_no;
-		uint16		generation;
-
-		SpinLockAcquire(&winfo->shared->mutex);
-		generation = winfo->shared->logicalrep_worker_generation;
-		slot_no = winfo->shared->logicalrep_worker_slot_no;
-		SpinLockRelease(&winfo->shared->mutex);
-
-		logicalrep_pa_worker_stop(slot_no, generation);
-
-		pa_free_worker_info(winfo);
-
+		pa_stop_worker(winfo);
 		return true;
 	}
 
@@ -607,6 +615,47 @@ pa_free_worker(ParallelApplyWorkerInfo *winfo)
 }
 
 /*
+ * Try to stop parallel apply workers that are not in use to keep the number of
+ * workers lower than half of the max_parallel_apply_workers_per_subscription.
+ */
+void
+pa_stop_idle_workers(void)
+{
+	List *active_workers;
+	ListCell *lc;
+	int max_applyworkers = max_parallel_apply_workers_per_subscription / 2;
+
+	if (list_length(ParallelApplyWorkerPool) <= max_applyworkers)
+		return;
+
+	active_workers = list_copy(ParallelApplyWorkerPool);
+
+	foreach(lc, active_workers)
+	{
+		ParallelApplyWorkerInfo *winfo = (ParallelApplyWorkerInfo *) lfirst(lc);
+
+		/*
+		 * Try to free and stop the worker that is in use. Normally, we free
+		 * the worker after ensuring that the transaction is committed by the
+		 * parallel worker but for rollbacks, we don't wait for the transaction
+		 * to finish so can't free the worker information immediately.
+		 */
+		if (winfo->in_use)
+			pa_free_worker(winfo);
+
+		/* Directly stop the worker if not in use. */
+		else
+			pa_stop_worker(winfo);
+
+		/* Recheck the number of workers. */
+		if (list_length(ParallelApplyWorkerPool) <= max_applyworkers)
+			break;
+	}
+
+	list_free(active_workers);
+}
+
+/*
  * Free the parallel apply worker information and unlink the files with
  * serialized changes if any.
  */
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index df67851..d3eb507 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -3610,6 +3610,13 @@ LogicalRepApplyLoop(XLogRecPtr last_received)
 		{
 			ConfigReloadPending = false;
 			ProcessConfigFile(PGC_SIGHUP);
+
+			/*
+			 * Try to stop free workers in the pool in case the
+			 * max_parallel_apply_workers_per_subscription is changed to a
+			 * lower value.
+			 */
+			pa_stop_idle_workers();
 		}
 
 		if (rc & WL_TIMEOUT)
diff --git a/src/include/replication/worker_internal.h b/src/include/replication/worker_internal.h
index 849b36c..70cfe4a 100644
--- a/src/include/replication/worker_internal.h
+++ b/src/include/replication/worker_internal.h
@@ -265,6 +265,7 @@ extern void set_apply_error_context_origin(char *originname);
 extern void pa_allocate_worker(TransactionId xid);
 extern ParallelApplyWorkerInfo *pa_find_worker(TransactionId xid);
 extern bool pa_free_worker(ParallelApplyWorkerInfo *winfo);
+extern void pa_stop_idle_workers(void);
 extern void pa_detach_all_error_mq(void);
 
 extern bool pa_send_data(ParallelApplyWorkerInfo *winfo, Size nbytes,
-- 
2.7.2.windows.1