From 20fd18b40748ebab540f2bce16afca781b1a7be8 Mon Sep 17 00:00:00 2001 From: Vignesh C Date: Mon, 12 Aug 2024 14:43:11 +0530 Subject: [PATCH v20240817 3/5] Reorganize tablesync Code and Introduce syncutils Reorganized the tablesync code by creating a new syncutils file. This refactoring will facilitate the development of sequence synchronization worker code. This commit separates code reorganization from functional changes, making it clearer to reviewers that only existing code has been moved. The changes in this patch can be merged with subsequent patches during the commit process. --- src/backend/replication/logical/Makefile | 1 + src/backend/replication/logical/meson.build | 1 + src/backend/replication/logical/syncutils.c | 478 ++++++++++++++++++++ src/backend/replication/logical/tablesync.c | 457 +------------------ src/include/replication/worker_internal.h | 3 + 5 files changed, 485 insertions(+), 455 deletions(-) create mode 100644 src/backend/replication/logical/syncutils.c diff --git a/src/backend/replication/logical/Makefile b/src/backend/replication/logical/Makefile index ba03eeff1c..3964a30109 100644 --- a/src/backend/replication/logical/Makefile +++ b/src/backend/replication/logical/Makefile @@ -27,6 +27,7 @@ OBJS = \ reorderbuffer.o \ slotsync.o \ snapbuild.o \ + syncutils.o \ tablesync.o \ worker.o diff --git a/src/backend/replication/logical/meson.build b/src/backend/replication/logical/meson.build index 3dec36a6de..27a0e30ab7 100644 --- a/src/backend/replication/logical/meson.build +++ b/src/backend/replication/logical/meson.build @@ -13,6 +13,7 @@ backend_sources += files( 'reorderbuffer.c', 'slotsync.c', 'snapbuild.c', + 'syncutils.c', 'tablesync.c', 'worker.c', ) diff --git a/src/backend/replication/logical/syncutils.c b/src/backend/replication/logical/syncutils.c new file mode 100644 index 0000000000..b841ab7941 --- /dev/null +++ b/src/backend/replication/logical/syncutils.c @@ -0,0 +1,478 @@ +/*------------------------------------------------------------------------- + * syncutils.c + * PostgreSQL logical replication: common synchronization code + * + * Copyright (c) 2024, PostgreSQL Global Development Group + * + * IDENTIFICATION + * src/backend/replication/logical/syncutils.c + * + * NOTES + * This file contains code common to table synchronization workers, and + * the sequence synchronization worker. + *------------------------------------------------------------------------- + */ + +#include "postgres.h" + +#include "catalog/pg_subscription_rel.h" +#include "pgstat.h" +#include "replication/logicallauncher.h" +#include "replication/origin.h" +#include "replication/worker_internal.h" +#include "storage/ipc.h" +#include "utils/lsyscache.h" +#include "utils/memutils.h" + +typedef enum +{ + SYNC_TABLE_STATE_NEEDS_REBUILD, + SYNC_TABLE_STATE_REBUILD_STARTED, + SYNC_TABLE_STATE_VALID, +} SyncingTablesState; + +static SyncingTablesState table_states_validity = SYNC_TABLE_STATE_NEEDS_REBUILD; +static List *table_states_not_ready = NIL; +static bool FetchTableStates(bool *started_tx); + +/* + * Exit routine for synchronization worker. + */ +void +pg_attribute_noreturn() +finish_sync_worker(void) +{ + /* + * Commit any outstanding transaction. This is the usual case, unless + * there was nothing to do for the table. + */ + if (IsTransactionState()) + { + CommitTransactionCommand(); + pgstat_report_stat(true); + } + + /* And flush all writes. */ + XLogFlush(GetXLogWriteRecPtr()); + + StartTransactionCommand(); + ereport(LOG, + (errmsg("logical replication table synchronization worker for subscription \"%s\", table \"%s\" has finished", + MySubscription->name, + get_rel_name(MyLogicalRepWorker->relid)))); + CommitTransactionCommand(); + + /* Find the leader apply worker and signal it. */ + logicalrep_worker_wakeup(MyLogicalRepWorker->subid, InvalidOid); + + /* Stop gracefully */ + proc_exit(0); +} + +/* + * Callback from syscache invalidation. + */ +void +invalidate_syncing_table_states(Datum arg, int cacheid, uint32 hashvalue) +{ + table_states_validity = SYNC_TABLE_STATE_NEEDS_REBUILD; +} + +/* + * Handle table synchronization cooperation from the apply worker. + * + * Walk over all subscription tables that are individually tracked by the + * apply process (currently, all that have state other than + * SUBREL_STATE_READY) and manage synchronization for them. + * + * If there are tables that need synchronizing and are not being synchronized + * yet, start sync workers for them (if there are free slots for sync + * workers). To prevent starting the sync worker for the same relation at a + * high frequency after a failure, we store its last start time with each sync + * state info. We start the sync worker for the same relation after waiting + * at least wal_retrieve_retry_interval. + * + * For tables that are being synchronized already, check if sync workers + * either need action from the apply worker or have finished. This is the + * SYNCWAIT to CATCHUP transition. + * + * If the synchronization position is reached (SYNCDONE), then the table can + * be marked as READY and is no longer tracked. + */ +static void +process_syncing_tables_for_apply(XLogRecPtr current_lsn) +{ + struct tablesync_start_time_mapping + { + Oid relid; + TimestampTz last_start_time; + }; + static HTAB *last_start_times = NULL; + ListCell *lc; + bool started_tx = false; + bool should_exit = false; + + Assert(!IsTransactionState()); + + /* We need up-to-date sync state info for subscription tables here. */ + FetchTableStates(&started_tx); + + /* + * Prepare a hash table for tracking last start times of workers, to avoid + * immediate restarts. We don't need it if there are no tables that need + * syncing. + */ + if (table_states_not_ready != NIL && !last_start_times) + { + HASHCTL ctl; + + ctl.keysize = sizeof(Oid); + ctl.entrysize = sizeof(struct tablesync_start_time_mapping); + last_start_times = hash_create("Logical replication table sync worker start times", + 256, &ctl, HASH_ELEM | HASH_BLOBS); + } + + /* + * Clean up the hash table when we're done with all tables (just to + * release the bit of memory). + */ + else if (table_states_not_ready == NIL && last_start_times) + { + hash_destroy(last_start_times); + last_start_times = NULL; + } + + /* + * Process all tables that are being synchronized. + */ + foreach(lc, table_states_not_ready) + { + SubscriptionRelState *rstate = (SubscriptionRelState *) lfirst(lc); + + if (rstate->state == SUBREL_STATE_SYNCDONE) + { + /* + * Apply has caught up to the position where the table sync has + * finished. Mark the table as ready so that the apply will just + * continue to replicate it normally. + */ + if (current_lsn >= rstate->lsn) + { + char originname[NAMEDATALEN]; + + rstate->state = SUBREL_STATE_READY; + rstate->lsn = current_lsn; + if (!started_tx) + { + StartTransactionCommand(); + started_tx = true; + } + + /* + * Remove the tablesync origin tracking if exists. + * + * There is a chance that the user is concurrently performing + * refresh for the subscription where we remove the table + * state and its origin or the tablesync worker would have + * already removed this origin. We can't rely on tablesync + * worker to remove the origin tracking as if there is any + * error while dropping we won't restart it to drop the + * origin. So passing missing_ok = true. + */ + ReplicationOriginNameForLogicalRep(MyLogicalRepWorker->subid, + rstate->relid, + originname, + sizeof(originname)); + replorigin_drop_by_name(originname, true, false); + + /* + * Update the state to READY only after the origin cleanup. + */ + UpdateSubscriptionRelState(MyLogicalRepWorker->subid, + rstate->relid, rstate->state, + rstate->lsn); + } + } + else + { + LogicalRepWorker *syncworker; + + /* + * Look for a sync worker for this relation. + */ + LWLockAcquire(LogicalRepWorkerLock, LW_SHARED); + + syncworker = logicalrep_worker_find(MyLogicalRepWorker->subid, + rstate->relid, false); + + if (syncworker) + { + /* Found one, update our copy of its state */ + SpinLockAcquire(&syncworker->relmutex); + rstate->state = syncworker->relstate; + rstate->lsn = syncworker->relstate_lsn; + if (rstate->state == SUBREL_STATE_SYNCWAIT) + { + /* + * Sync worker is waiting for apply. Tell sync worker it + * can catchup now. + */ + syncworker->relstate = SUBREL_STATE_CATCHUP; + syncworker->relstate_lsn = + Max(syncworker->relstate_lsn, current_lsn); + } + SpinLockRelease(&syncworker->relmutex); + + /* If we told worker to catch up, wait for it. */ + if (rstate->state == SUBREL_STATE_SYNCWAIT) + { + /* Signal the sync worker, as it may be waiting for us. */ + if (syncworker->proc) + logicalrep_worker_wakeup_ptr(syncworker); + + /* Now safe to release the LWLock */ + LWLockRelease(LogicalRepWorkerLock); + + if (started_tx) + { + /* + * We must commit the existing transaction to release + * the existing locks before entering a busy loop. + * This is required to avoid any undetected deadlocks + * due to any existing lock as deadlock detector won't + * be able to detect the waits on the latch. + */ + CommitTransactionCommand(); + pgstat_report_stat(false); + } + + /* + * Enter busy loop and wait for synchronization worker to + * reach expected state (or die trying). + */ + StartTransactionCommand(); + started_tx = true; + + wait_for_relation_state_change(rstate->relid, + SUBREL_STATE_SYNCDONE); + } + else + LWLockRelease(LogicalRepWorkerLock); + } + else + { + /* + * If there is no sync worker for this table yet, count + * running sync workers for this subscription, while we have + * the lock. + */ + int nsyncworkers = + logicalrep_sync_worker_count(MyLogicalRepWorker->subid); + + /* Now safe to release the LWLock */ + LWLockRelease(LogicalRepWorkerLock); + + /* + * If there are free sync worker slot(s), start a new sync + * worker for the table. + */ + if (nsyncworkers < max_sync_workers_per_subscription) + { + TimestampTz now = GetCurrentTimestamp(); + struct tablesync_start_time_mapping *hentry; + bool found; + + hentry = hash_search(last_start_times, &rstate->relid, + HASH_ENTER, &found); + + if (!found || + TimestampDifferenceExceeds(hentry->last_start_time, now, + wal_retrieve_retry_interval)) + { + logicalrep_worker_launch(WORKERTYPE_TABLESYNC, + MyLogicalRepWorker->dbid, + MySubscription->oid, + MySubscription->name, + MyLogicalRepWorker->userid, + rstate->relid, + DSM_HANDLE_INVALID); + hentry->last_start_time = now; + } + } + } + } + } + + if (started_tx) + { + /* + * Even when the two_phase mode is requested by the user, it remains + * as 'pending' until all tablesyncs have reached READY state. + * + * When this happens, we restart the apply worker and (if the + * conditions are still ok) then the two_phase tri-state will become + * 'enabled' at that time. + * + * Note: If the subscription has no tables then leave the state as + * PENDING, which allows ALTER SUBSCRIPTION ... REFRESH PUBLICATION to + * work. + */ + if (MySubscription->twophasestate == LOGICALREP_TWOPHASE_STATE_PENDING) + { + CommandCounterIncrement(); /* make updates visible */ + if (AllTablesyncsReady()) + { + ereport(LOG, + (errmsg("logical replication apply worker for subscription \"%s\" will restart so that two_phase can be enabled", + MySubscription->name))); + should_exit = true; + } + } + + CommitTransactionCommand(); + pgstat_report_stat(true); + } + + if (should_exit) + { + /* + * Reset the last-start time for this worker so that the launcher will + * restart it without waiting for wal_retrieve_retry_interval. + */ + ApplyLauncherForgetWorkerStartTime(MySubscription->oid); + + proc_exit(0); + } +} + +/* + * Process possible state change(s) of tables that are being synchronized. + */ +void +process_syncing_tables(XLogRecPtr current_lsn) +{ + switch (MyLogicalRepWorker->type) + { + case WORKERTYPE_PARALLEL_APPLY: + + /* + * Skip for parallel apply workers because they only operate on + * tables that are in a READY state. See pa_can_start() and + * should_apply_changes_for_rel(). + */ + break; + + case WORKERTYPE_TABLESYNC: + process_syncing_tables_for_sync(current_lsn); + break; + + case WORKERTYPE_APPLY: + process_syncing_tables_for_apply(current_lsn); + break; + + case WORKERTYPE_UNKNOWN: + /* Should never happen. */ + elog(ERROR, "Unknown worker type"); + } +} + +/* + * Common code to fetch the up-to-date sync state info into the static lists. + * + * Returns true if subscription has 1 or more tables, else false. + * + * Note: If this function started the transaction (indicated by the parameter) + * then it is the caller's responsibility to commit it. + */ +static bool +FetchTableStates(bool *started_tx) +{ + static bool has_subrels = false; + + *started_tx = false; + + if (table_states_validity != SYNC_TABLE_STATE_VALID) + { + MemoryContext oldctx; + List *rstates; + ListCell *lc; + SubscriptionRelState *rstate; + + table_states_validity = SYNC_TABLE_STATE_REBUILD_STARTED; + + /* Clean the old lists. */ + list_free_deep(table_states_not_ready); + table_states_not_ready = NIL; + + if (!IsTransactionState()) + { + StartTransactionCommand(); + *started_tx = true; + } + + /* Fetch all non-ready tables. */ + rstates = GetSubscriptionRelations(MySubscription->oid, true); + + /* Allocate the tracking info in a permanent memory context. */ + oldctx = MemoryContextSwitchTo(CacheMemoryContext); + foreach(lc, rstates) + { + rstate = palloc(sizeof(SubscriptionRelState)); + memcpy(rstate, lfirst(lc), sizeof(SubscriptionRelState)); + table_states_not_ready = lappend(table_states_not_ready, rstate); + } + MemoryContextSwitchTo(oldctx); + + /* + * Does the subscription have tables? + * + * If there were not-READY relations found then we know it does. But + * if table_states_not_ready was empty we still need to check again to + * see if there are 0 tables. + */ + has_subrels = (table_states_not_ready != NIL) || + HasSubscriptionRelations(MySubscription->oid); + + /* + * If the subscription relation cache has been invalidated since we + * entered this routine, we still use and return the relations we just + * finished constructing, to avoid infinite loops, but we leave the + * table states marked as stale so that we'll rebuild it again on next + * access. Otherwise, we mark the table states as valid. + */ + if (table_states_validity == SYNC_TABLE_STATE_REBUILD_STARTED) + table_states_validity = SYNC_TABLE_STATE_VALID; + } + + return has_subrels; +} + +/* + * If the subscription has no tables then return false. + * + * Otherwise, are all tablesyncs READY? + * + * Note: This function is not suitable to be called from outside of apply or + * tablesync workers because MySubscription needs to be already initialized. + */ +bool +AllTablesyncsReady(void) +{ + bool started_tx = false; + bool has_subrels = false; + + /* We need up-to-date sync state info for subscription tables here. */ + has_subrels = FetchTableStates(&started_tx); + + if (started_tx) + { + CommitTransactionCommand(); + pgstat_report_stat(true); + } + + /* + * Return false when there are no tables in subscription or not all tables + * are in ready state; true otherwise. + */ + return has_subrels && (table_states_not_ready == NIL); +} diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c index e03e761392..17c4d2e76a 100644 --- a/src/backend/replication/logical/tablesync.c +++ b/src/backend/replication/logical/tablesync.c @@ -117,59 +117,13 @@ #include "utils/array.h" #include "utils/builtins.h" #include "utils/lsyscache.h" -#include "utils/memutils.h" #include "utils/rls.h" #include "utils/snapmgr.h" #include "utils/syscache.h" #include "utils/usercontext.h" -typedef enum -{ - SYNC_TABLE_STATE_NEEDS_REBUILD, - SYNC_TABLE_STATE_REBUILD_STARTED, - SYNC_TABLE_STATE_VALID, -} SyncingTablesState; - -static SyncingTablesState table_states_validity = SYNC_TABLE_STATE_NEEDS_REBUILD; -static List *table_states_not_ready = NIL; -static bool FetchTableStates(bool *started_tx); - static StringInfo copybuf = NULL; -/* - * Exit routine for synchronization worker. - */ -static void -pg_attribute_noreturn() -finish_sync_worker(void) -{ - /* - * Commit any outstanding transaction. This is the usual case, unless - * there was nothing to do for the table. - */ - if (IsTransactionState()) - { - CommitTransactionCommand(); - pgstat_report_stat(true); - } - - /* And flush all writes. */ - XLogFlush(GetXLogWriteRecPtr()); - - StartTransactionCommand(); - ereport(LOG, - (errmsg("logical replication table synchronization worker for subscription \"%s\", table \"%s\" has finished", - MySubscription->name, - get_rel_name(MyLogicalRepWorker->relid)))); - CommitTransactionCommand(); - - /* Find the leader apply worker and signal it. */ - logicalrep_worker_wakeup(MyLogicalRepWorker->subid, InvalidOid); - - /* Stop gracefully */ - proc_exit(0); -} - /* * Wait until the relation sync state is set in the catalog to the expected * one; return true when it happens. @@ -180,7 +134,7 @@ finish_sync_worker(void) * Currently, this is used in the apply worker when transitioning from * CATCHUP state to SYNCDONE. */ -static bool +bool wait_for_relation_state_change(Oid relid, char expected_state) { char state; @@ -274,15 +228,6 @@ wait_for_worker_state_change(char expected_state) return false; } -/* - * Callback from syscache invalidation. - */ -void -invalidate_syncing_table_states(Datum arg, int cacheid, uint32 hashvalue) -{ - table_states_validity = SYNC_TABLE_STATE_NEEDS_REBUILD; -} - /* * Handle table synchronization cooperation from the synchronization * worker. @@ -291,7 +236,7 @@ invalidate_syncing_table_states(Datum arg, int cacheid, uint32 hashvalue) * predetermined synchronization point in the WAL stream, mark the table as * SYNCDONE and finish. */ -static void +void process_syncing_tables_for_sync(XLogRecPtr current_lsn) { SpinLockAcquire(&MyLogicalRepWorker->relmutex); @@ -393,303 +338,6 @@ process_syncing_tables_for_sync(XLogRecPtr current_lsn) SpinLockRelease(&MyLogicalRepWorker->relmutex); } -/* - * Handle table synchronization cooperation from the apply worker. - * - * Walk over all subscription tables that are individually tracked by the - * apply process (currently, all that have state other than - * SUBREL_STATE_READY) and manage synchronization for them. - * - * If there are tables that need synchronizing and are not being synchronized - * yet, start sync workers for them (if there are free slots for sync - * workers). To prevent starting the sync worker for the same relation at a - * high frequency after a failure, we store its last start time with each sync - * state info. We start the sync worker for the same relation after waiting - * at least wal_retrieve_retry_interval. - * - * For tables that are being synchronized already, check if sync workers - * either need action from the apply worker or have finished. This is the - * SYNCWAIT to CATCHUP transition. - * - * If the synchronization position is reached (SYNCDONE), then the table can - * be marked as READY and is no longer tracked. - */ -static void -process_syncing_tables_for_apply(XLogRecPtr current_lsn) -{ - struct tablesync_start_time_mapping - { - Oid relid; - TimestampTz last_start_time; - }; - static HTAB *last_start_times = NULL; - ListCell *lc; - bool started_tx = false; - bool should_exit = false; - - Assert(!IsTransactionState()); - - /* We need up-to-date sync state info for subscription tables here. */ - FetchTableStates(&started_tx); - - /* - * Prepare a hash table for tracking last start times of workers, to avoid - * immediate restarts. We don't need it if there are no tables that need - * syncing. - */ - if (table_states_not_ready != NIL && !last_start_times) - { - HASHCTL ctl; - - ctl.keysize = sizeof(Oid); - ctl.entrysize = sizeof(struct tablesync_start_time_mapping); - last_start_times = hash_create("Logical replication table sync worker start times", - 256, &ctl, HASH_ELEM | HASH_BLOBS); - } - - /* - * Clean up the hash table when we're done with all tables (just to - * release the bit of memory). - */ - else if (table_states_not_ready == NIL && last_start_times) - { - hash_destroy(last_start_times); - last_start_times = NULL; - } - - /* - * Process all tables that are being synchronized. - */ - foreach(lc, table_states_not_ready) - { - SubscriptionRelState *rstate = (SubscriptionRelState *) lfirst(lc); - - if (rstate->state == SUBREL_STATE_SYNCDONE) - { - /* - * Apply has caught up to the position where the table sync has - * finished. Mark the table as ready so that the apply will just - * continue to replicate it normally. - */ - if (current_lsn >= rstate->lsn) - { - char originname[NAMEDATALEN]; - - rstate->state = SUBREL_STATE_READY; - rstate->lsn = current_lsn; - if (!started_tx) - { - StartTransactionCommand(); - started_tx = true; - } - - /* - * Remove the tablesync origin tracking if exists. - * - * There is a chance that the user is concurrently performing - * refresh for the subscription where we remove the table - * state and its origin or the tablesync worker would have - * already removed this origin. We can't rely on tablesync - * worker to remove the origin tracking as if there is any - * error while dropping we won't restart it to drop the - * origin. So passing missing_ok = true. - */ - ReplicationOriginNameForLogicalRep(MyLogicalRepWorker->subid, - rstate->relid, - originname, - sizeof(originname)); - replorigin_drop_by_name(originname, true, false); - - /* - * Update the state to READY only after the origin cleanup. - */ - UpdateSubscriptionRelState(MyLogicalRepWorker->subid, - rstate->relid, rstate->state, - rstate->lsn); - } - } - else - { - LogicalRepWorker *syncworker; - - /* - * Look for a sync worker for this relation. - */ - LWLockAcquire(LogicalRepWorkerLock, LW_SHARED); - - syncworker = logicalrep_worker_find(MyLogicalRepWorker->subid, - rstate->relid, false); - - if (syncworker) - { - /* Found one, update our copy of its state */ - SpinLockAcquire(&syncworker->relmutex); - rstate->state = syncworker->relstate; - rstate->lsn = syncworker->relstate_lsn; - if (rstate->state == SUBREL_STATE_SYNCWAIT) - { - /* - * Sync worker is waiting for apply. Tell sync worker it - * can catchup now. - */ - syncworker->relstate = SUBREL_STATE_CATCHUP; - syncworker->relstate_lsn = - Max(syncworker->relstate_lsn, current_lsn); - } - SpinLockRelease(&syncworker->relmutex); - - /* If we told worker to catch up, wait for it. */ - if (rstate->state == SUBREL_STATE_SYNCWAIT) - { - /* Signal the sync worker, as it may be waiting for us. */ - if (syncworker->proc) - logicalrep_worker_wakeup_ptr(syncworker); - - /* Now safe to release the LWLock */ - LWLockRelease(LogicalRepWorkerLock); - - if (started_tx) - { - /* - * We must commit the existing transaction to release - * the existing locks before entering a busy loop. - * This is required to avoid any undetected deadlocks - * due to any existing lock as deadlock detector won't - * be able to detect the waits on the latch. - */ - CommitTransactionCommand(); - pgstat_report_stat(false); - } - - /* - * Enter busy loop and wait for synchronization worker to - * reach expected state (or die trying). - */ - StartTransactionCommand(); - started_tx = true; - - wait_for_relation_state_change(rstate->relid, - SUBREL_STATE_SYNCDONE); - } - else - LWLockRelease(LogicalRepWorkerLock); - } - else - { - /* - * If there is no sync worker for this table yet, count - * running sync workers for this subscription, while we have - * the lock. - */ - int nsyncworkers = - logicalrep_sync_worker_count(MyLogicalRepWorker->subid); - - /* Now safe to release the LWLock */ - LWLockRelease(LogicalRepWorkerLock); - - /* - * If there are free sync worker slot(s), start a new sync - * worker for the table. - */ - if (nsyncworkers < max_sync_workers_per_subscription) - { - TimestampTz now = GetCurrentTimestamp(); - struct tablesync_start_time_mapping *hentry; - bool found; - - hentry = hash_search(last_start_times, &rstate->relid, - HASH_ENTER, &found); - - if (!found || - TimestampDifferenceExceeds(hentry->last_start_time, now, - wal_retrieve_retry_interval)) - { - logicalrep_worker_launch(WORKERTYPE_TABLESYNC, - MyLogicalRepWorker->dbid, - MySubscription->oid, - MySubscription->name, - MyLogicalRepWorker->userid, - rstate->relid, - DSM_HANDLE_INVALID); - hentry->last_start_time = now; - } - } - } - } - } - - if (started_tx) - { - /* - * Even when the two_phase mode is requested by the user, it remains - * as 'pending' until all tablesyncs have reached READY state. - * - * When this happens, we restart the apply worker and (if the - * conditions are still ok) then the two_phase tri-state will become - * 'enabled' at that time. - * - * Note: If the subscription has no tables then leave the state as - * PENDING, which allows ALTER SUBSCRIPTION ... REFRESH PUBLICATION to - * work. - */ - if (MySubscription->twophasestate == LOGICALREP_TWOPHASE_STATE_PENDING) - { - CommandCounterIncrement(); /* make updates visible */ - if (AllTablesyncsReady()) - { - ereport(LOG, - (errmsg("logical replication apply worker for subscription \"%s\" will restart so that two_phase can be enabled", - MySubscription->name))); - should_exit = true; - } - } - - CommitTransactionCommand(); - pgstat_report_stat(true); - } - - if (should_exit) - { - /* - * Reset the last-start time for this worker so that the launcher will - * restart it without waiting for wal_retrieve_retry_interval. - */ - ApplyLauncherForgetWorkerStartTime(MySubscription->oid); - - proc_exit(0); - } -} - -/* - * Process possible state change(s) of tables that are being synchronized. - */ -void -process_syncing_tables(XLogRecPtr current_lsn) -{ - switch (MyLogicalRepWorker->type) - { - case WORKERTYPE_PARALLEL_APPLY: - - /* - * Skip for parallel apply workers because they only operate on - * tables that are in a READY state. See pa_can_start() and - * should_apply_changes_for_rel(). - */ - break; - - case WORKERTYPE_TABLESYNC: - process_syncing_tables_for_sync(current_lsn); - break; - - case WORKERTYPE_APPLY: - process_syncing_tables_for_apply(current_lsn); - break; - - case WORKERTYPE_UNKNOWN: - /* Should never happen. */ - elog(ERROR, "Unknown worker type"); - } -} /* * Create list of columns for COPY based on logical relation mapping. @@ -1561,77 +1209,6 @@ copy_table_done: return slotname; } -/* - * Common code to fetch the up-to-date sync state info into the static lists. - * - * Returns true if subscription has 1 or more tables, else false. - * - * Note: If this function started the transaction (indicated by the parameter) - * then it is the caller's responsibility to commit it. - */ -static bool -FetchTableStates(bool *started_tx) -{ - static bool has_subrels = false; - - *started_tx = false; - - if (table_states_validity != SYNC_TABLE_STATE_VALID) - { - MemoryContext oldctx; - List *rstates; - ListCell *lc; - SubscriptionRelState *rstate; - - table_states_validity = SYNC_TABLE_STATE_REBUILD_STARTED; - - /* Clean the old lists. */ - list_free_deep(table_states_not_ready); - table_states_not_ready = NIL; - - if (!IsTransactionState()) - { - StartTransactionCommand(); - *started_tx = true; - } - - /* Fetch all non-ready tables. */ - rstates = GetSubscriptionRelations(MySubscription->oid, true); - - /* Allocate the tracking info in a permanent memory context. */ - oldctx = MemoryContextSwitchTo(CacheMemoryContext); - foreach(lc, rstates) - { - rstate = palloc(sizeof(SubscriptionRelState)); - memcpy(rstate, lfirst(lc), sizeof(SubscriptionRelState)); - table_states_not_ready = lappend(table_states_not_ready, rstate); - } - MemoryContextSwitchTo(oldctx); - - /* - * Does the subscription have tables? - * - * If there were not-READY relations found then we know it does. But - * if table_states_not_ready was empty we still need to check again to - * see if there are 0 tables. - */ - has_subrels = (table_states_not_ready != NIL) || - HasSubscriptionRelations(MySubscription->oid); - - /* - * If the subscription relation cache has been invalidated since we - * entered this routine, we still use and return the relations we just - * finished constructing, to avoid infinite loops, but we leave the - * table states marked as stale so that we'll rebuild it again on next - * access. Otherwise, we mark the table states as valid. - */ - if (table_states_validity == SYNC_TABLE_STATE_REBUILD_STARTED) - table_states_validity = SYNC_TABLE_STATE_VALID; - } - - return has_subrels; -} - /* * Execute the initial sync with error handling. Disable the subscription, * if it's required. @@ -1720,36 +1297,6 @@ TablesyncWorkerMain(Datum main_arg) finish_sync_worker(); } -/* - * If the subscription has no tables then return false. - * - * Otherwise, are all tablesyncs READY? - * - * Note: This function is not suitable to be called from outside of apply or - * tablesync workers because MySubscription needs to be already initialized. - */ -bool -AllTablesyncsReady(void) -{ - bool started_tx = false; - bool has_subrels = false; - - /* We need up-to-date sync state info for subscription tables here. */ - has_subrels = FetchTableStates(&started_tx); - - if (started_tx) - { - CommitTransactionCommand(); - pgstat_report_stat(true); - } - - /* - * Return false when there are no tables in subscription or not all tables - * are in ready state; true otherwise. - */ - return has_subrels && (table_states_not_ready == NIL); -} - /* * Update the two_phase state of the specified subscription in pg_subscription. */ diff --git a/src/include/replication/worker_internal.h b/src/include/replication/worker_internal.h index 9646261d7e..6ff5643132 100644 --- a/src/include/replication/worker_internal.h +++ b/src/include/replication/worker_internal.h @@ -250,6 +250,7 @@ extern void logicalrep_worker_stop(Oid subid, Oid relid); extern void logicalrep_pa_worker_stop(ParallelApplyWorkerInfo *winfo); extern void logicalrep_worker_wakeup(Oid subid, Oid relid); extern void logicalrep_worker_wakeup_ptr(LogicalRepWorker *worker); +extern void pg_attribute_noreturn() finish_sync_worker(void); extern int logicalrep_sync_worker_count(Oid subid); @@ -259,7 +260,9 @@ extern void ReplicationOriginNameForLogicalRep(Oid suboid, Oid relid, extern bool AllTablesyncsReady(void); extern void UpdateTwoPhaseState(Oid suboid, char new_state); +extern bool wait_for_relation_state_change(Oid relid, char expected_state); extern void process_syncing_tables(XLogRecPtr current_lsn); +extern void process_syncing_tables_for_sync(XLogRecPtr current_lsn); extern void invalidate_syncing_table_states(Datum arg, int cacheid, uint32 hashvalue); -- 2.34.1