From 0cf3079200abdbe428ce55b7997e2345c8a40471 Mon Sep 17 00:00:00 2001 From: wangw Date: Wed, 1 Jun 2022 19:59:27 +0800 Subject: [PATCH v8 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 num_sync ( + max_apply_bgworkers_per_subscription (integer) + + max_apply_bgworkers_per_subscription configuration parameter + + + + + 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 + streaming to apply. + + + The apply background workers are taken from the pool defined by + max_logical_replication_workers. + + + The default value is 3. This parameter can only be set in the + postgresql.conf file or on the server command + line. + + + + diff --git a/src/backend/replication/logical/applybgwroker.c b/src/backend/replication/logical/applybgwroker.c index 4184343f6c..82861b0bfd 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/worker_internal.h" @@ -558,9 +559,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 e810875ddd..67dd40598c 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 1dc1a491e9..3da852fed4 100644 --- a/src/include/replication/worker_internal.h +++ b/src/include/replication/worker_internal.h @@ -158,6 +158,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