#include "postgres.h"
#include "catalog/pg_subscription_rel.h"
#include "pgstat.h"
#include "replication/logicallauncher.h"
#include "replication/worker_internal.h"
#include "storage/ipc.h"
#include "utils/lsyscache.h"
#include "utils/memutils.h"

Include dependency graph for syncutils.c:

Enumerations
enum	SyncingRelationsState { SYNC_RELATIONS_STATE_NEEDS_REBUILD , SYNC_RELATIONS_STATE_REBUILD_STARTED , SYNC_RELATIONS_STATE_VALID }

Functions
pg_noreturn void	FinishSyncWorker (void)

void	InvalidateSyncingRelStates (Datum arg, int cacheid, uint32 hashvalue)

void	launch_sync_worker (LogicalRepWorkerType wtype, int nsyncworkers, Oid relid, TimestampTz *last_start_time)

void	ProcessSyncingRelations (XLogRecPtr current_lsn)

void	FetchRelationStates (bool has_pending_subtables, bool has_pending_subsequences, bool *started_tx)

Variables
static SyncingRelationsState	relation_states_validity = SYNC_RELATIONS_STATE_NEEDS_REBUILD

Enumeration Type Documentation

◆ SyncingRelationsState

enum SyncingRelationsState

Enumerator
SYNC_RELATIONS_STATE_NEEDS_REBUILD
SYNC_RELATIONS_STATE_REBUILD_STARTED
SYNC_RELATIONS_STATE_VALID

Definition at line 37 of file syncutils.c.

{
    SYNC_RELATIONS_STATE_NEEDS_REBUILD,
    SYNC_RELATIONS_STATE_REBUILD_STARTED,
    SYNC_RELATIONS_STATE_VALID,
} SyncingRelationsState;

Function Documentation

◆ FetchRelationStates()

void FetchRelationStates	(	bool *	has_pending_subtables,
		bool *	has_pending_subsequences,
		bool *	started_tx
	)

Definition at line 202 of file syncutils.c.

{
    /*
     * has_subtables and has_subsequences_non_ready are declared as static,
     * since the same value can be used until the system table is invalidated.
     */
    static bool has_subtables = false;
    static bool has_subsequences_non_ready = false;
 
    *started_tx = false;
 
    if (relation_states_validity != SYNC_RELATIONS_STATE_VALID)
    {
        MemoryContext oldctx;
        List       *rstates;
        SubscriptionRelState *rstate;
 
        relation_states_validity = SYNC_RELATIONS_STATE_REBUILD_STARTED;
        has_subsequences_non_ready = false;
 
        /* Clean the old lists. */
        list_free_deep(table_states_not_ready);
        table_states_not_ready = NIL;
 
        if (!IsTransactionState())
        {
            StartTransactionCommand();
            *started_tx = true;
        }
 
        /* Fetch tables and sequences that are in non-READY state. */
        rstates = GetSubscriptionRelations(MySubscription->oid, true, true,
                                           true);
 
        /* Allocate the tracking info in a permanent memory context. */
        oldctx = MemoryContextSwitchTo(CacheMemoryContext);
        foreach_ptr(SubscriptionRelState, subrel, rstates)
        {
            if (get_rel_relkind(subrel->relid) == RELKIND_SEQUENCE)
                has_subsequences_non_ready = true;
            else
            {
                rstate = palloc(sizeof(SubscriptionRelState));
                memcpy(rstate, subrel, sizeof(SubscriptionRelState));
                table_states_not_ready = lappend(table_states_not_ready,
                                                 rstate);
            }
        }
        MemoryContextSwitchTo(oldctx);
 
        /*
         * Does the subscription have tables?
         *
         * If there were not-READY tables 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_subtables = (table_states_not_ready != NIL) ||
            HasSubscriptionTables(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 (relation_states_validity == SYNC_RELATIONS_STATE_REBUILD_STARTED)
            relation_states_validity = SYNC_RELATIONS_STATE_VALID;
    }
 
    if (has_pending_subtables)
        *has_pending_subtables = has_subtables;
 
    if (has_pending_subsequences)
        *has_pending_subsequences = has_subsequences_non_ready;
}

References CacheMemoryContext, foreach_ptr, get_rel_relkind(), GetSubscriptionRelations(), HasSubscriptionTables(), IsTransactionState(), lappend(), list_free_deep(), MemoryContextSwitchTo(), MySubscription, NIL, Subscription::oid, palloc(), relation_states_validity, StartTransactionCommand(), SYNC_RELATIONS_STATE_REBUILD_STARTED, SYNC_RELATIONS_STATE_VALID, and table_states_not_ready.

Referenced by AllTablesyncsReady(), HasSubscriptionTablesCached(), ProcessSequencesForSync(), and ProcessSyncingTablesForApply().

◆ FinishSyncWorker()

pg_noreturn void FinishSyncWorker ( void )

Definition at line 50 of file syncutils.c.

{
    Assert(am_sequencesync_worker() || am_tablesync_worker());
 
    /*
     * 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());
 
    if (am_sequencesync_worker())
    {
        ereport(LOG,
                errmsg("logical replication sequence synchronization worker for subscription \"%s\" has finished",
                       MySubscription->name));
 
        /*
         * Reset last_seqsync_start_time, so that next time a sequencesync
         * worker is needed it can be started promptly.
         */
        logicalrep_reset_seqsync_start_time();
    }
    else
    {
        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(WORKERTYPE_APPLY, MyLogicalRepWorker->subid,
                                 InvalidOid);
    }
 
    /* Stop gracefully */
    proc_exit(0);
}

References am_sequencesync_worker(), am_tablesync_worker(), Assert(), CommitTransactionCommand(), ereport, errmsg(), get_rel_name(), GetXLogWriteRecPtr(), InvalidOid, IsTransactionState(), LOG, logicalrep_reset_seqsync_start_time(), logicalrep_worker_wakeup(), MyLogicalRepWorker, MySubscription, Subscription::name, pgstat_report_stat(), proc_exit(), LogicalRepWorker::relid, StartTransactionCommand(), LogicalRepWorker::subid, WORKERTYPE_APPLY, and XLogFlush().

Referenced by LogicalRepSyncTableStart(), ProcessSyncingTablesForSync(), SequenceSyncWorkerMain(), and TableSyncWorkerMain().

◆ InvalidateSyncingRelStates()

void InvalidateSyncingRelStates	(	Datum	arg,
		int	cacheid,
		uint32	hashvalue
	)

Definition at line 101 of file syncutils.c.

{
    relation_states_validity = SYNC_RELATIONS_STATE_NEEDS_REBUILD;
}

References relation_states_validity, and SYNC_RELATIONS_STATE_NEEDS_REBUILD.

Referenced by ParallelApplyWorkerMain(), and SetupApplyOrSyncWorker().

◆ launch_sync_worker()

void launch_sync_worker	(	LogicalRepWorkerType	wtype,
		int	nsyncworkers,
		Oid	relid,
		TimestampTz *	last_start_time
	)

Definition at line 117 of file syncutils.c.

{
    TimestampTz now;
 
    Assert((wtype == WORKERTYPE_TABLESYNC && OidIsValid(relid)) ||
           (wtype == WORKERTYPE_SEQUENCESYNC && !OidIsValid(relid)));
 
    /* If there is a free sync worker slot, start a new sync worker */
    if (nsyncworkers >= max_sync_workers_per_subscription)
        return;
 
    now = GetCurrentTimestamp();
 
    if (!(*last_start_time) ||
        TimestampDifferenceExceeds(*last_start_time, now,
                                   wal_retrieve_retry_interval))
    {
        /*
         * Set the last_start_time even if we fail to start the worker, so
         * that we won't retry until wal_retrieve_retry_interval has elapsed.
         */
        *last_start_time = now;
        (void) logicalrep_worker_launch(wtype,
                                        MyLogicalRepWorker->dbid,
                                        MySubscription->oid,
                                        MySubscription->name,
                                        MyLogicalRepWorker->userid,
                                        relid, DSM_HANDLE_INVALID, false);
    }
}

References Assert(), LogicalRepWorker::dbid, DSM_HANDLE_INVALID, GetCurrentTimestamp(), logicalrep_worker_launch(), max_sync_workers_per_subscription, MyLogicalRepWorker, MySubscription, Subscription::name, now(), Subscription::oid, OidIsValid, TimestampDifferenceExceeds(), LogicalRepWorker::userid, wal_retrieve_retry_interval, WORKERTYPE_SEQUENCESYNC, and WORKERTYPE_TABLESYNC.

Referenced by ProcessSequencesForSync(), and ProcessSyncingTablesForApply().

◆ ProcessSyncingRelations()

void ProcessSyncingRelations ( XLogRecPtr current_lsn )

Definition at line 155 of file syncutils.c.

{
    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:
            ProcessSyncingTablesForSync(current_lsn);
            break;
 
        case WORKERTYPE_APPLY:
            ProcessSyncingTablesForApply(current_lsn);
            ProcessSequencesForSync();
            break;
 
        case WORKERTYPE_SEQUENCESYNC:
            /* Should never happen. */
            elog(ERROR, "sequence synchronization worker is not expected to process relations");
            break;
 
        case WORKERTYPE_UNKNOWN:
            /* Should never happen. */
            elog(ERROR, "Unknown worker type");
    }
}