From 4305b9fa223955fdbff4a5439a72db313ade9b93 Mon Sep 17 00:00:00 2001 From: Shveta Malik Date: Tue, 1 Aug 2023 14:29:14 +0530 Subject: [PATCH v10 3/3] max_slot_sync_workers GUC based implementation. This patch adds a new GUC max_slot_sync_workers, default value and max value is kept at 2 and 50 respectively for this PoC patch. Now replication launcher divides the work equally between these slot-sync workers. Let us say there are multiple slots on primary belonging to 10 DBs and say new GUC is set at default value of 2, then each worker will manage 5 dbs and will keep on synching the slots for them. If a new DB is found by replciation launcher, it will assign this new db to the worker handling the minimum number of dbs currently (or first worker in case of equal count) New SlotSyncWorker array is added to LogicalRepCtxStruct. The shared memory for these worker slots are allocated based on max_slot_sync_workers value. Each worker slot will have its own dbids array. Since the upper limit of this dbids array is not known, so it needs to be handled using dsm. For this PoC patch, the size for dbids is kept fixed (i.e. 50 per worker) --- src/backend/commands/subscriptioncmds.c | 4 +- src/backend/replication/logical/launcher.c | 406 ++++++++++++++++-- src/backend/replication/logical/slotsync.c | 124 ++++-- src/backend/replication/logical/tablesync.c | 12 +- src/backend/replication/logical/worker.c | 3 +- src/backend/storage/lmgr/lwlocknames.txt | 1 + src/backend/utils/misc/guc_tables.c | 13 + src/backend/utils/misc/postgresql.conf.sample | 1 + src/include/postmaster/bgworker_internals.h | 1 + src/include/replication/logicallauncher.h | 2 + src/include/replication/worker_internal.h | 35 +- 11 files changed, 514 insertions(+), 88 deletions(-) diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c index 42e9b1056c..d4e798baeb 100644 --- a/src/backend/commands/subscriptioncmds.c +++ b/src/backend/commands/subscriptioncmds.c @@ -993,7 +993,7 @@ AlterSubscription_refresh(Subscription *sub, bool copy_data, RemoveSubscriptionRel(sub->oid, relid); - logicalrep_worker_stop(MyDatabaseId, sub->oid, relid); + logicalrep_worker_stop(sub->oid, relid); /* * For READY state, we would have already dropped the @@ -1591,7 +1591,7 @@ DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel) { LogicalRepWorker *w = (LogicalRepWorker *) lfirst(lc); - logicalrep_worker_stop(w->dbid, w->subid, w->relid); + logicalrep_worker_stop(w->subid, w->relid); } list_free(subworkers); diff --git a/src/backend/replication/logical/launcher.c b/src/backend/replication/logical/launcher.c index 640f7647cc..9a390930a1 100644 --- a/src/backend/replication/logical/launcher.c +++ b/src/backend/replication/logical/launcher.c @@ -58,6 +58,16 @@ int max_logical_replication_workers = 4; int max_sync_workers_per_subscription = 2; int max_parallel_apply_workers_per_subscription = 2; +int max_slot_sync_workers = 2; + +/* + * allocation size for dbids array for each SlotSyncWorker in shared memory. + * This fixed size is only for PoC patch. This needs to be otherwise done + * using dsm. + */ +#define ALLOC_DB_PER_WORKER 50 + +SlotSyncWorker *MySlotSyncWorker = NULL; LogicalRepWorker *MyLogicalRepWorker = NULL; @@ -71,6 +81,7 @@ typedef struct LogicalRepCtxStruct dshash_table_handle last_start_dsh; /* Background workers. */ + SlotSyncWorker *ss_workers; /* slot sync workers */ LogicalRepWorker workers[FLEXIBLE_ARRAY_MEMBER]; } LogicalRepCtxStruct; @@ -108,7 +119,6 @@ static void logicalrep_launcher_attach_dshmem(void); static void ApplyLauncherSetWorkerStartTime(Oid subid, TimestampTz start_time); static TimestampTz ApplyLauncherGetWorkerStartTime(Oid subid); - /* * Load the list of subscriptions. * @@ -247,7 +257,7 @@ WaitForReplicationWorkerAttach(LogicalRepWorker *worker, * We are only interested in the leader apply worker or table sync worker. */ LogicalRepWorker * -logicalrep_worker_find(Oid dbid, Oid subid, Oid relid, bool only_running) +logicalrep_worker_find(Oid subid, Oid relid, bool only_running) { int i; LogicalRepWorker *res = NULL; @@ -263,8 +273,8 @@ logicalrep_worker_find(Oid dbid, Oid subid, Oid relid, bool only_running) if (isParallelApplyWorker(w)) continue; - if (w->in_use && w->dbid == dbid && w->subid == subid && - w->relid == relid && (!only_running || w->proc)) + if (w->in_use && w->subid == subid && w->relid == relid && + (!only_running || w->proc)) { res = w; break; @@ -321,13 +331,9 @@ logicalrep_worker_launch(Oid dbid, Oid subid, const char *subname, Oid userid, /* Sanity check - tablesync worker cannot be a subworker */ Assert(!(is_parallel_apply_worker && OidIsValid(relid))); - if (OidIsValid(subid)) - ereport(DEBUG1, - (errmsg_internal("starting logical replication worker for subscription \"%s\"", - subname))); - else - ereport(DEBUG1, - (errmsg_internal("starting replication slot synchronization worker"))); + ereport(DEBUG1, + (errmsg_internal("starting logical replication worker for subscription \"%s\"", + subname))); /* Report this after the initial starting message for consistency. */ if (max_replication_slots == 0) @@ -364,9 +370,7 @@ retry: * 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 (worker == NULL || - (OidIsValid(relid) && - nsyncworkers >= max_sync_workers_per_subscription)) + if (worker == NULL || nsyncworkers >= max_sync_workers_per_subscription) { bool did_cleanup = false; @@ -465,17 +469,12 @@ retry: bgw.bgw_start_time = BgWorkerStart_ConsistentState; snprintf(bgw.bgw_library_name, MAXPGPATH, "postgres"); - if (!OidIsValid(subid)) - snprintf(bgw.bgw_function_name, BGW_MAXLEN, "ReplSlotSyncMain"); - else if (is_parallel_apply_worker) + if (is_parallel_apply_worker) snprintf(bgw.bgw_function_name, BGW_MAXLEN, "ParallelApplyWorkerMain"); else snprintf(bgw.bgw_function_name, BGW_MAXLEN, "ApplyWorkerMain"); - if (!OidIsValid(subid)) - snprintf(bgw.bgw_name, BGW_MAXLEN, - "replication slot synchronization worker"); - else if (OidIsValid(relid)) + if (OidIsValid(relid)) snprintf(bgw.bgw_name, BGW_MAXLEN, "logical replication worker for subscription %u sync %u", subid, relid); else if (is_parallel_apply_worker) @@ -603,13 +602,13 @@ logicalrep_worker_stop_internal(LogicalRepWorker *worker, int signo) * Stop the logical replication worker for subid/relid, if any. */ void -logicalrep_worker_stop(Oid dbid, Oid subid, Oid relid) +logicalrep_worker_stop(Oid subid, Oid relid) { LogicalRepWorker *worker; LWLockAcquire(LogicalRepWorkerLock, LW_SHARED); - worker = logicalrep_worker_find(dbid, subid, relid, false); + worker = logicalrep_worker_find(subid, relid, false); if (worker) { @@ -670,13 +669,13 @@ logicalrep_pa_worker_stop(ParallelApplyWorkerInfo *winfo) * Wake up (using latch) any logical replication worker for specified sub/rel. */ void -logicalrep_worker_wakeup(Oid dbid, Oid subid, Oid relid) +logicalrep_worker_wakeup(Oid subid, Oid relid) { LogicalRepWorker *worker; LWLockAcquire(LogicalRepWorkerLock, LW_SHARED); - worker = logicalrep_worker_find(dbid, subid, relid, true); + worker = logicalrep_worker_find(subid, relid, true); if (worker) logicalrep_worker_wakeup_ptr(worker); @@ -943,6 +942,8 @@ void ApplyLauncherShmemInit(void) { bool found; + bool foundSlotSync; + Size ss_size; LogicalRepCtx = (LogicalRepCtxStruct *) ShmemInitStruct("Logical Replication Launcher Data", @@ -967,6 +968,32 @@ ApplyLauncherShmemInit(void) SpinLockInit(&worker->relmutex); } } + + /* Allocate shared-memory for slot-sync workers now */ + ss_size = mul_size(max_slot_sync_workers, sizeof(SlotSyncWorker)); + ss_size = add_size(ss_size, mul_size(max_slot_sync_workers * + ALLOC_DB_PER_WORKER, sizeof(Oid))); + + LogicalRepCtx->ss_workers = (SlotSyncWorker *) + ShmemInitStruct("Slot Sync Workers", ss_size, &foundSlotSync); + + if (!foundSlotSync) + { + int slot; + + for (slot = 0; slot < max_slot_sync_workers; slot++) + { + int idx; + SlotSyncWorker *worker = &LogicalRepCtx->ss_workers[slot]; + + memset(worker, 0, sizeof(SlotSyncWorker)); + + for (idx = 0; idx < ALLOC_DB_PER_WORKER; idx++) + { + worker->dbids[idx] = InvalidOid; + } + } + } } /* @@ -1104,6 +1131,313 @@ ApplyLauncherWakeup(void) kill(LogicalRepCtx->launcher_pid, SIGUSR1); } +/* + * Clean up slot-sync worker info. + */ +static void +sloysync_worker_cleanup(SlotSyncWorker *worker) +{ + uint i; + + Assert(LWLockHeldByMeInMode(SlotSyncWorkerLock, LW_EXCLUSIVE)); + + worker->in_use = false; + worker->proc = NULL; + worker->dbid = InvalidOid; + + for (i = 0; i < worker->dbcount; i++) + { + worker->dbids[i] = InvalidOid; + } + + worker->dbcount = 0; + worker->userid = InvalidOid; +} + +/* + * Cleanup function. + * + * Called on slot-sync worker exit. + */ +static void +slotsync_worker_onexit(int code, Datum arg) +{ + + LWLockAcquire(SlotSyncWorkerLock, LW_EXCLUSIVE); + + sloysync_worker_cleanup(MySlotSyncWorker); + + LWLockRelease(SlotSyncWorkerLock); +} + +/* + * Attach Slot-sync worker to worker-slot assigned by launcher. + */ +void +slotsync_worker_attach(int slot) +{ + /* Block concurrent access. */ + LWLockAcquire(SlotSyncWorkerLock, LW_EXCLUSIVE); + + Assert(slot >= 0 && slot < max_slot_sync_workers); + MySlotSyncWorker = &LogicalRepCtx->ss_workers[slot]; + + if (!MySlotSyncWorker->in_use) + { + LWLockRelease(SlotSyncWorkerLock); + ereport(ERROR, + (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("logical replication worker slot %d is empty, cannot attach", + slot))); + } + + if (MySlotSyncWorker->proc) + { + LWLockRelease(SlotSyncWorkerLock); + ereport(ERROR, + (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("logical replication worker slot %d is already used by " + "another worker, cannot attach", slot))); + } + + MySlotSyncWorker->proc = MyProc; + before_shmem_exit(slotsync_worker_onexit, (Datum) 0); + + LWLockRelease(SlotSyncWorkerLock); +} + +/* + * Wait for a background worker to start up and attach to the shmem context. + * + * This is only needed for cleaning up the shared memory in case the worker + * fails to attach. + * + * Returns whether the attach was successful. + */ +static bool +WaitForSlotSyncWorkerAttach(SlotSyncWorker *worker, + uint16 generation, + BackgroundWorkerHandle *handle) +{ + BgwHandleStatus status; + int rc; + + for (;;) + { + pid_t pid; + + CHECK_FOR_INTERRUPTS(); + + LWLockAcquire(SlotSyncWorkerLock, LW_SHARED); + + /* Worker either died or has started. Return false if died. */ + if (!worker->in_use || worker->proc) + { + LWLockRelease(SlotSyncWorkerLock); + return worker->in_use; + } + + LWLockRelease(SlotSyncWorkerLock); + + /* Check if worker has died before attaching, and clean up after it. */ + status = GetBackgroundWorkerPid(handle, &pid); + + if (status == BGWH_STOPPED) + { + LWLockAcquire(SlotSyncWorkerLock, LW_EXCLUSIVE); + /* Ensure that this was indeed the worker we waited for. */ + if (generation == worker->generation) + sloysync_worker_cleanup(worker); + LWLockRelease(SlotSyncWorkerLock); + return false; + } + + /* + * We need timeout because we generally don't get notified via latch + * about the worker attach. But we don't expect to have to wait long. + */ + rc = WaitLatch(MyLatch, + WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH, + 10L, WAIT_EVENT_BGWORKER_STARTUP); + + if (rc & WL_LATCH_SET) + { + ResetLatch(MyLatch); + CHECK_FOR_INTERRUPTS(); + } + } +} + +/* + * Walks the slot-sync workers array and searches for one that matches given + * dbid. Since one worker can manage multiple dbs, so it walks the db array in + * each worker to find the match. + */ +static SlotSyncWorker * +slotsync_worker_find(Oid dbid) +{ + int i; + SlotSyncWorker *res = NULL; + + Assert(LWLockHeldByMe(SlotSyncWorkerLock)); + + /* Search for attached worker for a given dbid */ + for (i = 0; i < max_slot_sync_workers; i++) + { + SlotSyncWorker *w = &LogicalRepCtx->ss_workers[i]; + int cnt; + + if (!w->in_use) + continue; + + for (cnt = 0; cnt < w->dbcount; cnt++) + { + Oid wdbid = w->dbids[cnt]; + if (wdbid == dbid) + { + res = w; + break; + } + } + + /* if worker is found, break the outer loop */ + if (res) + break; + } + + return res; +} + +/* + * Start new slot-sync background worker, if possible. + * + * Returns true on success, false on failure. + */ +static bool +slot_sync_worker_launch(Oid dbid, Oid userid) +{ + BackgroundWorker bgw; + BackgroundWorkerHandle *bgw_handle; + uint16 generation; + int i; + int worker_slot = 0; + SlotSyncWorker *worker = NULL; + int mindbcnt = 0; + + Assert(OidIsValid(dbid)); + + /* + * We need to do the modification of the shared memory under lock so that + * we have consistent view. + */ + LWLockAcquire(SlotSyncWorkerLock, LW_EXCLUSIVE); + + /* Find unused worker slot. */ + for (i = 0; i < max_slot_sync_workers; i++) + { + SlotSyncWorker *w = &LogicalRepCtx->ss_workers[i]; + + if (!w->in_use) + { + worker = w; + worker_slot = i; + break; + } + } + + /* If all the workers are currently in use. Find the one with + * minimum number of dbs and use that. */ + if (!worker) + { + for (i = 0; i < max_slot_sync_workers; i++) + { + SlotSyncWorker *w = &LogicalRepCtx->ss_workers[i]; + + if (i == 0) + { + mindbcnt = w->dbcount; + worker = w; + worker_slot = i; + } + else if (w->dbcount < mindbcnt) + { + mindbcnt = w->dbcount; + worker = w; + worker_slot = i; + } + } + } + + /* If worker is being reused, just update dbids array and count */ + if (worker->in_use) + { + worker->dbids[worker->dbcount++] = dbid; + LWLockRelease(SlotSyncWorkerLock); + + ereport(LOG, + (errmsg("adding database %d to replication slot synchronization" + " worker %d", + dbid, worker_slot))); + + return true; + } + + /* Else prepare the new worker. */ + worker->launch_time = GetCurrentTimestamp(); + worker->in_use = true; + worker->generation++; + + /* 'proc' will be assigned in ReplSlotSyncMain when we attach + * that worker to a particular worker-array slot */ + worker->proc = NULL; + + worker->dbid = dbid; /* TODO: do we really need this? analyse more here */ + worker->dbids[worker->dbcount++] = dbid; + worker->userid = userid; + + /* Before releasing lock, remember generation for future identification. */ + generation = worker->generation; + + LWLockRelease(SlotSyncWorkerLock); + + /* Register the new dynamic worker. */ + memset(&bgw, 0, sizeof(bgw)); + bgw.bgw_flags = BGWORKER_SHMEM_ACCESS | + BGWORKER_BACKEND_DATABASE_CONNECTION; + bgw.bgw_start_time = BgWorkerStart_ConsistentState; + snprintf(bgw.bgw_library_name, MAXPGPATH, "postgres"); + + snprintf(bgw.bgw_function_name, BGW_MAXLEN, "ReplSlotSyncMain"); + + snprintf(bgw.bgw_name, BGW_MAXLEN, + "slot synchronization worker %d", worker_slot); + + snprintf(bgw.bgw_type, BGW_MAXLEN, "slot synchronization worker"); + + bgw.bgw_restart_time = BGW_NEVER_RESTART; + bgw.bgw_notify_pid = MyProcPid; + bgw.bgw_main_arg = Int32GetDatum(worker_slot); + + if (!RegisterDynamicBackgroundWorker(&bgw, &bgw_handle)) + { + /* Failed to start worker, so clean up the worker slot. */ + LWLockAcquire(SlotSyncWorkerLock, LW_EXCLUSIVE); + Assert(generation == worker->generation); + sloysync_worker_cleanup(worker); + LWLockRelease(SlotSyncWorkerLock); + + ereport(WARNING, + (errcode(ERRCODE_CONFIGURATION_LIMIT_EXCEEDED), + errmsg("out of background worker slots"), + errhint("You might need to increase %s.", "max_worker_processes"))); + return false; + } + + /* Now wait until it attaches. */ + return WaitForSlotSyncWorkerAttach(worker, generation, bgw_handle); +} + + static void ApplyLauncherStartSlotSync(long *wait_time) { @@ -1114,6 +1448,9 @@ ApplyLauncherStartSlotSync(long *wait_time) MemoryContext tmpctx; MemoryContext oldctx; + if (max_slot_sync_workers == 0) + return; + if (strcmp(synchronize_slot_names, "") == 0) return; @@ -1125,7 +1462,7 @@ ApplyLauncherStartSlotSync(long *wait_time) /* Use temporary context for the slot list and worker info. */ tmpctx = AllocSetContextCreate(TopMemoryContext, - "Logical Replication Launcher slot sync ctx", + "Logical Replication Launcher Slot Sync ctx", ALLOCSET_DEFAULT_SIZES); oldctx = MemoryContextSwitchTo(tmpctx); @@ -1134,7 +1471,7 @@ ApplyLauncherStartSlotSync(long *wait_time) foreach(lc, slots) { WalRecvReplicationSlotData *slot_data = lfirst(lc); - LogicalRepWorker *w; + SlotSyncWorker *w; TimestampTz last_sync; TimestampTz now; long elapsed; @@ -1142,10 +1479,9 @@ ApplyLauncherStartSlotSync(long *wait_time) if (!OidIsValid(slot_data->persistent_data.database)) continue; - LWLockAcquire(LogicalRepWorkerLock, LW_SHARED); - w = logicalrep_worker_find(slot_data->persistent_data.database, InvalidOid, - InvalidOid, false); - LWLockRelease(LogicalRepWorkerLock); + LWLockAcquire(SlotSyncWorkerLock, LW_SHARED); + w = slotsync_worker_find(slot_data->persistent_data.database); + LWLockRelease(SlotSyncWorkerLock); if (w != NULL) continue; /* worker is running already */ @@ -1165,10 +1501,8 @@ ApplyLauncherStartSlotSync(long *wait_time) (elapsed = TimestampDifferenceMilliseconds(last_sync, now)) >= wal_retrieve_retry_interval) { slot_data->last_sync_time = now; - logicalrep_worker_launch(slot_data->persistent_data.database, - InvalidOid, NULL, - BOOTSTRAP_SUPERUSERID, InvalidOid, - DSM_HANDLE_INVALID); + slot_sync_worker_launch(slot_data->persistent_data.database, + BOOTSTRAP_SUPERUSERID); } else { @@ -1213,7 +1547,7 @@ ApplyLauncherStartSubs(long *wait_time) continue; LWLockAcquire(LogicalRepWorkerLock, LW_SHARED); - w = logicalrep_worker_find(sub->dbid, sub->oid, InvalidOid, false); + w = logicalrep_worker_find(sub->oid, InvalidOid, false); LWLockRelease(LogicalRepWorkerLock); if (w != NULL) diff --git a/src/backend/replication/logical/slotsync.c b/src/backend/replication/logical/slotsync.c index 77457001e7..51745c9b08 100644 --- a/src/backend/replication/logical/slotsync.c +++ b/src/backend/replication/logical/slotsync.c @@ -14,6 +14,7 @@ #include "commands/dbcommands.h" #include "pgstat.h" #include "postmaster/bgworker.h" +#include "replication/logical.h" #include "replication/logicallauncher.h" #include "replication/logicalworker.h" #include "replication/walreceiver.h" @@ -25,6 +26,16 @@ #include "utils/pg_lsn.h" #include "utils/varlena.h" +typedef struct RemoteSlot +{ + char *name; + char *plugin; + char *database; + XLogRecPtr restart_lsn; + XLogRecPtr confirmed_lsn; + TransactionId catalog_xmin; +} RemoteSlot; + /* * Wait for remote slot to pass localy reserved position. */ @@ -86,17 +97,29 @@ wait_for_primary_slot_catchup(WalReceiverConn *wrconn, char *slot_name, } } +/* + * Advance local slot to remote_slot's positions + */ +static void +local_slot_advance(RemoteSlot *remote_slot) +{ + LogicalConfirmReceivedLocation(remote_slot->confirmed_lsn); + LogicalIncreaseXminForSlot(remote_slot->confirmed_lsn, + remote_slot->catalog_xmin); + LogicalIncreaseRestartDecodingForSlot(remote_slot->confirmed_lsn, + remote_slot->restart_lsn); + ReplicationSlotMarkDirty(); +} + /* * Synchronize single slot to given position. * * This optionally creates new slot if there is no existing one. */ static void -synchronize_one_slot(WalReceiverConn *wrconn, char *slot_name, char *database, - char *plugin_name, XLogRecPtr target_lsn) +synchronize_one_slot(WalReceiverConn *wrconn, RemoteSlot *remote_slot) { bool found = false; - XLogRecPtr endlsn; /* Search for the named slot and mark it active if we find it. */ LWLockAcquire(ReplicationSlotControlLock, LW_SHARED); @@ -107,7 +130,7 @@ synchronize_one_slot(WalReceiverConn *wrconn, char *slot_name, char *database, if (!s->in_use) continue; - if (strcmp(NameStr(s->data.name), slot_name) == 0) + if (strcmp(NameStr(s->data.name), remote_slot->name) == 0) { found = true; break; @@ -120,18 +143,22 @@ synchronize_one_slot(WalReceiverConn *wrconn, char *slot_name, char *database, /* Already existing slot, acquire */ if (found) { - ReplicationSlotAcquire(slot_name, true); + ReplicationSlotAcquire(remote_slot->name, true); - if (target_lsn < MyReplicationSlot->data.confirmed_flush) + if (remote_slot->confirmed_lsn < MyReplicationSlot->data.confirmed_flush) { elog(DEBUG1, "not synchronizing slot %s; synchronization would move it backward", - slot_name); + remote_slot->name); ReplicationSlotRelease(); CommitTransactionCommand(); return; } + + /* advance current lsns of slot to remote slot's current position */ + local_slot_advance(remote_slot); + ReplicationSlotSave(); } /* Otherwise create the slot first. */ else @@ -139,12 +166,12 @@ synchronize_one_slot(WalReceiverConn *wrconn, char *slot_name, char *database, TransactionId xmin_horizon = InvalidTransactionId; ReplicationSlot *slot; - ReplicationSlotCreate(slot_name, true, RS_EPHEMERAL, false); + ReplicationSlotCreate(remote_slot->name, true, RS_EPHEMERAL, false); slot = MyReplicationSlot; SpinLockAcquire(&slot->mutex); - slot->data.database = get_database_oid(database, false); - namestrcpy(&slot->data.plugin, plugin_name); + slot->data.database = get_database_oid(remote_slot->database, false); + namestrcpy(&slot->data.plugin, remote_slot->plugin); SpinLockRelease(&slot->mutex); ReplicationSlotReserveWal(); @@ -156,25 +183,23 @@ synchronize_one_slot(WalReceiverConn *wrconn, char *slot_name, char *database, ReplicationSlotsComputeRequiredXmin(true); LWLockRelease(ProcArrayLock); - if (target_lsn < MyReplicationSlot->data.restart_lsn) + if (remote_slot->confirmed_lsn < MyReplicationSlot->data.restart_lsn) { ereport(LOG, errmsg("waiting for remote slot \"%s\" LSN (%X/%X) to pass local slot LSN (%X/%X)", - slot_name, - LSN_FORMAT_ARGS(target_lsn), LSN_FORMAT_ARGS(MyReplicationSlot->data.restart_lsn))); + remote_slot->name, + LSN_FORMAT_ARGS(remote_slot->confirmed_lsn), LSN_FORMAT_ARGS(MyReplicationSlot->data.restart_lsn))); - wait_for_primary_slot_catchup(wrconn, slot_name, + wait_for_primary_slot_catchup(wrconn, remote_slot->name, MyReplicationSlot->data.restart_lsn); } + + /* advance current lsns of slot to remote slot's current position */ + local_slot_advance(remote_slot); ReplicationSlotPersist(); } - endlsn = pg_logical_replication_slot_advance(target_lsn); - - elog(DEBUG3, "synchronized slot %s to lsn (%X/%X)", - slot_name, LSN_FORMAT_ARGS(endlsn)); - ReplicationSlotRelease(); CommitTransactionCommand(); } @@ -185,10 +210,11 @@ synchronize_slots(void) WalRcvExecResult *res; WalReceiverConn *wrconn = NULL; TupleTableSlot *slot; - Oid slotRow[3] = {TEXTOID, TEXTOID, LSNOID}; + Oid slotRow[6] = {TEXTOID, TEXTOID, LSNOID, LSNOID, XIDOID, TEXTOID}; StringInfoData s; char *database; char *err; + int i; MemoryContext oldctx = CurrentMemoryContext; if (!WalRcv) @@ -210,10 +236,27 @@ synchronize_slots(void) resetStringInfo(&s); appendStringInfo(&s, - "SELECT slot_name, plugin, confirmed_flush_lsn" + "SELECT slot_name, plugin, confirmed_flush_lsn," + " restart_lsn, catalog_xmin, database" " FROM pg_catalog.pg_replication_slots" - " WHERE database = %s", - quote_literal_cstr(database)); + " WHERE database IN "); + + Assert (MySlotSyncWorker->dbcount); + + appendStringInfoChar(&s, '('); + for (i = 0; i < MySlotSyncWorker->dbcount; i++) + { + char *dbname; + if (i != 0) + appendStringInfoChar(&s, ','); + + dbname = get_database_name(MySlotSyncWorker->dbids[i]); + appendStringInfo(&s, "%s", + quote_literal_cstr(dbname)); + pfree(dbname); + } + appendStringInfoChar(&s, ')'); + if (strcmp(synchronize_slot_names, "") != 0 && strcmp(synchronize_slot_names, "*") != 0) { char *rawname; @@ -234,7 +277,7 @@ synchronize_slots(void) appendStringInfoChar(&s, ')'); } - res = walrcv_exec(wrconn, s.data, 3, slotRow); + res = walrcv_exec(wrconn, s.data, 6, slotRow); pfree(s.data); if (res->status != WALRCV_OK_TUPLES) @@ -249,22 +292,29 @@ synchronize_slots(void) slot = MakeSingleTupleTableSlot(res->tupledesc, &TTSOpsMinimalTuple); while (tuplestore_gettupleslot(res->tuplestore, true, false, slot)) { - char *slot_name; - char *plugin_name; - XLogRecPtr confirmed_flush_lsn; bool isnull; + RemoteSlot *remote_slot = palloc0(sizeof(RemoteSlot)); + + remote_slot->name = TextDatumGetCString(slot_getattr(slot, 1, &isnull)); + Assert(!isnull); + + remote_slot->plugin = TextDatumGetCString(slot_getattr(slot, 2, &isnull)); + Assert(!isnull); + + remote_slot->confirmed_lsn = DatumGetLSN(slot_getattr(slot, 3, &isnull)); + Assert(!isnull); - slot_name = TextDatumGetCString(slot_getattr(slot, 1, &isnull)); + remote_slot->restart_lsn = DatumGetLSN(slot_getattr(slot, 4, &isnull)); Assert(!isnull); - plugin_name = TextDatumGetCString(slot_getattr(slot, 2, &isnull)); + remote_slot->catalog_xmin = DatumGetTransactionId(slot_getattr(slot, 5, &isnull)); Assert(!isnull); - confirmed_flush_lsn = DatumGetLSN(slot_getattr(slot, 3, &isnull)); + remote_slot->database = TextDatumGetCString(slot_getattr(slot, 6, &isnull)); Assert(!isnull); - synchronize_one_slot(wrconn, slot_name, database, plugin_name, - confirmed_flush_lsn); + synchronize_one_slot(wrconn, remote_slot); + pfree(remote_slot); ExecClearTuple(slot); } @@ -284,7 +334,7 @@ ReplSlotSyncMain(Datum main_arg) int worker_slot = DatumGetInt32(main_arg); /* Attach to slot */ - logicalrep_worker_attach(worker_slot); + slotsync_worker_attach(worker_slot); /* Establish signal handlers. */ BackgroundWorkerUnblockSignals(); @@ -293,14 +343,14 @@ ReplSlotSyncMain(Datum main_arg) load_file("libpqwalreceiver", false); /* Connect to our database. */ - BackgroundWorkerInitializeConnectionByOid(MyLogicalRepWorker->dbid, - MyLogicalRepWorker->userid, + BackgroundWorkerInitializeConnectionByOid(MySlotSyncWorker->dbid, + MySlotSyncWorker->userid, 0); StartTransactionCommand(); ereport(LOG, - (errmsg("replication slot synchronization worker for database \"%s\" has started", - get_database_name(MyLogicalRepWorker->dbid)))); + (errmsg("replication slot synchronization worker %d for database \"%s\" has started", + worker_slot, get_database_name(MySlotSyncWorker->dbid)))); CommitTransactionCommand(); /* Main wait loop. */ diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c index 5a98d2b699..9a501a4606 100644 --- a/src/backend/replication/logical/tablesync.c +++ b/src/backend/replication/logical/tablesync.c @@ -157,8 +157,7 @@ finish_sync_worker(void) CommitTransactionCommand(); /* Find the leader apply worker and signal it. */ - logicalrep_worker_wakeup(MyLogicalRepWorker->dbid, - MyLogicalRepWorker->subid, InvalidOid); + logicalrep_worker_wakeup(MyLogicalRepWorker->subid, InvalidOid); /* Stop gracefully */ proc_exit(0); @@ -198,8 +197,7 @@ wait_for_relation_state_change(Oid relid, char expected_state) /* Check if the sync worker is still running and bail if not. */ LWLockAcquire(LogicalRepWorkerLock, LW_SHARED); - worker = logicalrep_worker_find(MyLogicalRepWorker->dbid, - MyLogicalRepWorker->subid, relid, + worker = logicalrep_worker_find(MyLogicalRepWorker->subid, relid, false); LWLockRelease(LogicalRepWorkerLock); if (!worker) @@ -246,8 +244,7 @@ wait_for_worker_state_change(char expected_state) * waiting. */ LWLockAcquire(LogicalRepWorkerLock, LW_SHARED); - worker = logicalrep_worker_find(MyLogicalRepWorker->dbid, - MyLogicalRepWorker->subid, + worker = logicalrep_worker_find(MyLogicalRepWorker->subid, InvalidOid, false); if (worker && worker->proc) logicalrep_worker_wakeup_ptr(worker); @@ -513,8 +510,7 @@ process_syncing_tables_for_apply(XLogRecPtr current_lsn) */ LWLockAcquire(LogicalRepWorkerLock, LW_SHARED); - syncworker = logicalrep_worker_find(MyLogicalRepWorker->dbid, - MyLogicalRepWorker->subid, + syncworker = logicalrep_worker_find(MyLogicalRepWorker->subid, rstate->relid, false); if (syncworker) diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c index bb9a49e08c..832b1cf764 100644 --- a/src/backend/replication/logical/worker.c +++ b/src/backend/replication/logical/worker.c @@ -1590,8 +1590,7 @@ apply_handle_stream_start(StringInfo s) * Signal the leader apply worker, as it may be waiting for * us. */ - logicalrep_worker_wakeup(MyLogicalRepWorker->dbid, - MyLogicalRepWorker->subid, InvalidOid); + logicalrep_worker_wakeup(MyLogicalRepWorker->subid, InvalidOid); } parallel_stream_nchanges = 0; diff --git a/src/backend/storage/lmgr/lwlocknames.txt b/src/backend/storage/lmgr/lwlocknames.txt index b34b6afecd..33bc68ad07 100644 --- a/src/backend/storage/lmgr/lwlocknames.txt +++ b/src/backend/storage/lmgr/lwlocknames.txt @@ -53,3 +53,4 @@ XactTruncationLock 44 # 45 was XactTruncationLock until removal of BackendRandomLock WrapLimitsVacuumLock 46 NotifyQueueTailLock 47 +SlotSyncWorkerLock 48 diff --git a/src/backend/utils/misc/guc_tables.c b/src/backend/utils/misc/guc_tables.c index 131e32273c..9cfc359950 100644 --- a/src/backend/utils/misc/guc_tables.c +++ b/src/backend/utils/misc/guc_tables.c @@ -3510,6 +3510,19 @@ struct config_int ConfigureNamesInt[] = NULL, NULL, NULL }, + { + {"max_slot_sync_workers", + PGC_SIGHUP, + REPLICATION_STANDBY, + gettext_noop("Maximum number of slots synchronization workers " + "on a standby."), + NULL, + }, + &max_slot_sync_workers, + 2, 0, MAX_SLOT_SYNC_WORKER_LIMIT, + NULL, NULL, NULL + }, + /* End-of-list marker */ { {NULL, 0, 0, NULL, NULL}, NULL, 0, 0, 0, NULL, NULL, NULL diff --git a/src/backend/utils/misc/postgresql.conf.sample b/src/backend/utils/misc/postgresql.conf.sample index 63daf586f3..4e0ae87b54 100644 --- a/src/backend/utils/misc/postgresql.conf.sample +++ b/src/backend/utils/misc/postgresql.conf.sample @@ -359,6 +359,7 @@ #recovery_min_apply_delay = 0 # minimum delay for applying changes during recovery #synchronize_slot_names = '' # replication slot names to synchronize from # primary to streaming replication standby server +#max_slot_sync_workers = 2 # max number of slot synchronization workers # - Subscribers - diff --git a/src/include/postmaster/bgworker_internals.h b/src/include/postmaster/bgworker_internals.h index 4ad63fd9bd..19c5421a55 100644 --- a/src/include/postmaster/bgworker_internals.h +++ b/src/include/postmaster/bgworker_internals.h @@ -22,6 +22,7 @@ * Maximum possible value of parallel workers. */ #define MAX_PARALLEL_WORKER_LIMIT 1024 +#define MAX_SLOT_SYNC_WORKER_LIMIT 50 /* * List of background workers, private to postmaster. diff --git a/src/include/replication/logicallauncher.h b/src/include/replication/logicallauncher.h index 80fdbf9657..690f3deebd 100644 --- a/src/include/replication/logicallauncher.h +++ b/src/include/replication/logicallauncher.h @@ -15,6 +15,8 @@ extern PGDLLIMPORT int max_logical_replication_workers; extern PGDLLIMPORT int max_sync_workers_per_subscription; extern PGDLLIMPORT int max_parallel_apply_workers_per_subscription; +extern PGDLLIMPORT int max_slot_sync_workers; + 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 d42cff3f2c..7ea2ea8fa2 100644 --- a/src/include/replication/worker_internal.h +++ b/src/include/replication/worker_internal.h @@ -84,6 +84,33 @@ typedef struct LogicalRepWorker TimestampTz reply_time; } LogicalRepWorker; +typedef struct SlotSyncWorker +{ + /* Time at which this worker was launched. */ + TimestampTz launch_time; + + /* Indicates if this slot is used or free. */ + bool in_use; + + /* Increased every time the slot is taken by new worker. */ + uint16 generation; + + /* Pointer to proc array. NULL if not running. */ + PGPROC *proc; + + /* User to use for connection (will be same as owner of subscription). */ + Oid userid; + + /* Database id to connect to. */ + Oid dbid; + + /* Count of Database ids it manages */ + uint32 dbcount; + + /* Database ids it manages */ + Oid dbids[FLEXIBLE_ARRAY_MEMBER]; +} SlotSyncWorker; + /* * State of the transaction in parallel apply worker. * @@ -222,21 +249,23 @@ extern PGDLLIMPORT struct WalReceiverConn *LogRepWorkerWalRcvConn; /* Worker and subscription objects. */ extern PGDLLIMPORT Subscription *MySubscription; extern PGDLLIMPORT LogicalRepWorker *MyLogicalRepWorker; +extern PGDLLIMPORT SlotSyncWorker *MySlotSyncWorker; extern PGDLLIMPORT bool in_remote_transaction; extern PGDLLIMPORT bool InitializingApplyWorker; extern void logicalrep_worker_attach(int slot); -extern LogicalRepWorker *logicalrep_worker_find(Oid dbid, Oid subid, Oid relid, +extern void slotsync_worker_attach(int slot); +extern LogicalRepWorker *logicalrep_worker_find(Oid subid, Oid relid, bool only_running); extern List *logicalrep_workers_find(Oid subid, bool only_running); extern bool logicalrep_worker_launch(Oid dbid, Oid subid, const char *subname, Oid userid, Oid relid, dsm_handle subworker_dsm); -extern void logicalrep_worker_stop(Oid dbid, Oid subid, Oid relid); +extern void logicalrep_worker_stop(Oid subid, Oid relid); extern void logicalrep_pa_worker_stop(ParallelApplyWorkerInfo *winfo); -extern void logicalrep_worker_wakeup(Oid dbid, Oid subid, Oid relid); +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); -- 2.34.1