public inbox for [email protected]  
help / color / mirror / Atom feed
From: Peter Smith <[email protected]>
To: PostgreSQL Hackers <[email protected]>
Subject: Re: logicalrep_worker_launch -- counting/checking the worker limits
Date: Tue, 15 Aug 2023 12:38:28 +1000
Message-ID: <CAHut+Psxo7bHj_3n3EYVnZekr7MUBvqkFQQ_6d=woOoNGqOPWQ@mail.gmail.com> (raw)
In-Reply-To: <CAHut+PuDYZ-hHmMnKB+O2KoEZspqHOeWeXqQmYf=B=DQdEhErQ@mail.gmail.com>
References: <CAHut+Pu-7DiCwMdbBf5vzUL1t8H=_NpyFR4RvpZbpMGXV=+QLQ@mail.gmail.com>
	<CAHut+PuDYZ-hHmMnKB+O2KoEZspqHOeWeXqQmYf=B=DQdEhErQ@mail.gmail.com>

A rebase was needed due to a recent push [1].

PSA v3.

------
[1] https://github.com/postgres/postgres/commit/2a8b40e3681921943a2989fd4ec6cdbf8766566c

Kind Regards,
Peter Smith.
Fujitsu Australia


Attachments:

  [application/octet-stream] v3-0001-logicalrep_worker_launch-limit-checks.patch (3.3K, ../CAHut+Psxo7bHj_3n3EYVnZekr7MUBvqkFQQ_6d=woOoNGqOPWQ@mail.gmail.com/2-v3-0001-logicalrep_worker_launch-limit-checks.patch)
  download | inline diff:
From d9a8efd395daf14f7ac945c0106a289902958653 Mon Sep 17 00:00:00 2001
From: Peter Smith <[email protected]>
Date: Tue, 15 Aug 2023 12:35:30 +1000
Subject: [PATCH v3] logicalrep_worker_launch limit checks

---
 src/backend/replication/logical/launcher.c | 35 +++++++++++++++++++++---------
 1 file changed, 25 insertions(+), 10 deletions(-)

diff --git a/src/backend/replication/logical/launcher.c b/src/backend/replication/logical/launcher.c
index 7cc0a16..81f7af4 100644
--- a/src/backend/replication/logical/launcher.c
+++ b/src/backend/replication/logical/launcher.c
@@ -313,11 +313,10 @@ logicalrep_worker_launch(LogicalRepWorkerType wtype,
 	int			i;
 	int			slot = 0;
 	LogicalRepWorker *worker = NULL;
-	int			nsyncworkers;
-	int			nparallelapplyworkers;
 	TimestampTz now;
 	bool		is_tablesync_worker = (wtype == WORKERTYPE_TABLESYNC);
 	bool		is_parallel_apply_worker = (wtype == WORKERTYPE_PARALLEL_APPLY);
+	bool		reached_limit_for_type = false;
 
 	/*----------
 	 * Sanity checks:
@@ -359,7 +358,19 @@ retry:
 		}
 	}
 
-	nsyncworkers = logicalrep_sync_worker_count(subid);
+	/* Check if we are at the configured limit for the worker type */
+	if (is_tablesync_worker)
+	{
+		int n = logicalrep_sync_worker_count(subid);
+
+		reached_limit_for_type = (n >= max_sync_workers_per_subscription);
+	}
+	else if (is_parallel_apply_worker)
+	{
+		int n = logicalrep_pa_worker_count(subid);
+
+		reached_limit_for_type = (n >= max_parallel_apply_workers_per_subscription);
+	}
 
 	now = GetCurrentTimestamp();
 
@@ -367,8 +378,12 @@ retry:
 	 * If we didn't find a free slot, try to do garbage collection.  The
 	 * reason we do this is because if some worker failed to start up and its
 	 * parent has crashed while waiting, the in_use state was never cleared.
+	 *
+	 * If the worker type has reached its limit we also want to trigger garbage
+	 * collection. This is in case one of those workers was previously in error
+	 * and maybe now can be re-used.
 	 */
-	if (worker == NULL || nsyncworkers >= max_sync_workers_per_subscription)
+	if (worker == NULL || reached_limit_for_type)
 	{
 		bool		did_cleanup = false;
 
@@ -402,20 +417,17 @@ retry:
 	 * sync worker limit per subscription. So, just return silently as we
 	 * might get here because of an otherwise harmless race condition.
 	 */
-	if (is_tablesync_worker && nsyncworkers >= max_sync_workers_per_subscription)
+	if (is_tablesync_worker && reached_limit_for_type)
 	{
 		LWLockRelease(LogicalRepWorkerLock);
 		return false;
 	}
 
-	nparallelapplyworkers = logicalrep_pa_worker_count(subid);
-
 	/*
 	 * Return false if the number of parallel apply workers reached the limit
 	 * per subscription.
 	 */
-	if (is_parallel_apply_worker &&
-		nparallelapplyworkers >= max_parallel_apply_workers_per_subscription)
+	if (is_parallel_apply_worker && reached_limit_for_type)
 	{
 		LWLockRelease(LogicalRepWorkerLock);
 		return false;
@@ -852,7 +864,10 @@ logicalrep_sync_worker_count(Oid subid)
 
 	Assert(LWLockHeldByMe(LogicalRepWorkerLock));
 
-	/* Search for attached worker for a given subscription id. */
+	/*
+	 * Scan all attached tablesync workers, only counting those which
+	 * have the given subscription id.
+	 */
 	for (i = 0; i < max_logical_replication_workers; i++)
 	{
 		LogicalRepWorker *w = &LogicalRepCtx->workers[i];
-- 
1.8.3.1



view thread (5+ messages)

reply

Reply instructions:

You may reply publicly to this message via plain-text email
using any one of the following methods:

* Reply to all the recipients using the --to and --cc options:
  reply via email

  To: [email protected]
  Cc: [email protected], [email protected]
  Subject: Re: logicalrep_worker_launch -- counting/checking the worker limits
  In-Reply-To: <CAHut+Psxo7bHj_3n3EYVnZekr7MUBvqkFQQ_6d=woOoNGqOPWQ@mail.gmail.com>

* Save the following mbox file, import it into your mail client,
  and reply-to-all from there: mbox

This inbox is served by agora; see mirroring instructions
for how to clone and mirror all data and code used for this inbox