public inbox for [email protected]
help / color / mirror / Atom feedFrom: 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