logicallauncher.h File Reference

This graph shows which files directly or indirectly include this file:

Go to the source code of this file.

Functions
void	ApplyLauncherRegister (void)
void	ApplyLauncherMain (Datum main_arg)
Size	ApplyLauncherShmemSize (void)
void	ApplyLauncherShmemInit (void)
void	ApplyLauncherForgetWorkerStartTime (Oid subid)
void	ApplyLauncherWakeupAtCommit (void)
void	ApplyLauncherWakeup (void)
void	AtEOXact_ApplyLauncher (bool isCommit)
void	CreateConflictDetectionSlot (void)
bool	IsLogicalLauncher (void)
pid_t	GetLeaderApplyWorkerPid (pid_t pid)

Variables
PGDLLIMPORT int	max_logical_replication_workers
PGDLLIMPORT int	max_sync_workers_per_subscription
PGDLLIMPORT int	max_parallel_apply_workers_per_subscription

Function Documentation

ApplyLauncherForgetWorkerStartTime()

void ApplyLauncherForgetWorkerStartTime

(

Oid

subid

)

Definition at line 1101 of file launcher.c.

{
    logicalrep_launcher_attach_dshmem();
 
    (void) dshash_delete_key(last_start_times, &subid);
}

References dshash_delete_key(), last_start_times, and logicalrep_launcher_attach_dshmem().

Referenced by apply_worker_exit(), DisableSubscriptionAndExit(), DropSubscription(), InitializeLogRepWorker(), maybe_reread_subscription(), and process_syncing_tables_for_apply().

ApplyLauncherMain()

void ApplyLauncherMain

(

Datum

main_arg

)

Definition at line 1151 of file launcher.c.

{
    ereport(DEBUG1,
            (errmsg_internal("logical replication launcher started")));
 
    before_shmem_exit(logicalrep_launcher_onexit, (Datum) 0);
 
    Assert(LogicalRepCtx->launcher_pid == 0);
    LogicalRepCtx->launcher_pid = MyProcPid;
 
    /* Establish signal handlers. */
    pqsignal(SIGHUP, SignalHandlerForConfigReload);
    pqsignal(SIGTERM, die);
    BackgroundWorkerUnblockSignals();
 
    /*
     * Establish connection to nailed catalogs (we only ever access
     * pg_subscription).
     */
    BackgroundWorkerInitializeConnection(NULL, NULL, 0);
 
    /*
     * Acquire the conflict detection slot at startup to ensure it can be
     * dropped if no longer needed after a restart.
     */
    acquire_conflict_slot_if_exists();
 
    /* Enter main loop */
    for (;;)
    {
        int         rc;
        List       *sublist;
        ListCell   *lc;
        MemoryContext subctx;
        MemoryContext oldctx;
        long        wait_time = DEFAULT_NAPTIME_PER_CYCLE;
        bool        can_update_xmin = true;
        bool        retain_dead_tuples = false;
        TransactionId xmin = InvalidTransactionId;
 
        CHECK_FOR_INTERRUPTS();
 
        /* Use temporary context to avoid leaking memory across cycles. */
        subctx = AllocSetContextCreate(TopMemoryContext,
                                       "Logical Replication Launcher sublist",
                                       ALLOCSET_DEFAULT_SIZES);
        oldctx = MemoryContextSwitchTo(subctx);
 
        /*
         * Start any missing workers for enabled subscriptions.
         *
         * Also, during the iteration through all subscriptions, we compute
         * the minimum XID required to protect deleted tuples for conflict
         * detection if one of the subscription enables retain_dead_tuples
         * option.
         */
        sublist = get_subscription_list();
        foreach(lc, sublist)
        {
            Subscription *sub = (Subscription *) lfirst(lc);
            LogicalRepWorker *w;
            TimestampTz last_start;
            TimestampTz now;
            long        elapsed;
 
            if (sub->retaindeadtuples)
            {
                retain_dead_tuples = true;
 
                /*
                 * Create a replication slot to retain information necessary
                 * for conflict detection such as dead tuples, commit
                 * timestamps, and origins.
                 *
                 * The slot is created before starting the apply worker to
                 * prevent it from unnecessarily maintaining its
                 * oldest_nonremovable_xid.
                 *
                 * The slot is created even for a disabled subscription to
                 * ensure that conflict-related information is available when
                 * applying remote changes that occurred before the
                 * subscription was enabled.
                 */
                CreateConflictDetectionSlot();
 
                if (sub->retentionactive)
                {
                    /*
                     * Can't advance xmin of the slot unless all the
                     * subscriptions actively retaining dead tuples are
                     * enabled. This is required to ensure that we don't
                     * advance the xmin of CONFLICT_DETECTION_SLOT if one of
                     * the subscriptions is not enabled. Otherwise, we won't
                     * be able to detect conflicts reliably for such a
                     * subscription even though it has set the
                     * retain_dead_tuples option.
                     */
                    can_update_xmin &= sub->enabled;
 
                    /*
                     * Initialize the slot once the subscription activiates
                     * retention.
                     */
                    if (!TransactionIdIsValid(MyReplicationSlot->data.xmin))
                        init_conflict_slot_xmin();
                }
            }
 
            if (!sub->enabled)
                continue;
 
            LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
            w = logicalrep_worker_find(sub->oid, InvalidOid, false);
 
            if (w != NULL)
            {
                /*
                 * Compute the minimum xmin required to protect dead tuples
                 * required for conflict detection among all running apply
                 * workers. This computation is performed while holding
                 * LogicalRepWorkerLock to prevent accessing invalid worker
                 * data, in scenarios where a worker might exit and reset its
                 * state concurrently.
                 */
                if (sub->retaindeadtuples &&
                    sub->retentionactive &&
                    can_update_xmin)
                    compute_min_nonremovable_xid(w, &xmin);
 
                LWLockRelease(LogicalRepWorkerLock);
 
                /* worker is running already */
                continue;
            }
 
            LWLockRelease(LogicalRepWorkerLock);
 
            /*
             * Can't advance xmin of the slot unless all the workers
             * corresponding to subscriptions actively retaining dead tuples
             * are running, disabling the further computation of the minimum
             * nonremovable xid.
             */
            if (sub->retaindeadtuples && sub->retentionactive)
                can_update_xmin = false;
 
            /*
             * If the worker is eligible to start now, launch it.  Otherwise,
             * adjust wait_time so that we'll wake up as soon as it can be
             * started.
             *
             * Each subscription's apply worker can only be restarted once per
             * wal_retrieve_retry_interval, so that errors do not cause us to
             * repeatedly restart the worker as fast as possible.  In cases
             * where a restart is expected (e.g., subscription parameter
             * changes), another process should remove the last-start entry
             * for the subscription so that the worker can be restarted
             * without waiting for wal_retrieve_retry_interval to elapse.
             */
            last_start = ApplyLauncherGetWorkerStartTime(sub->oid);
            now = GetCurrentTimestamp();
            if (last_start == 0 ||
                (elapsed = TimestampDifferenceMilliseconds(last_start, now)) >= wal_retrieve_retry_interval)
            {
                ApplyLauncherSetWorkerStartTime(sub->oid, now);
                if (!logicalrep_worker_launch(WORKERTYPE_APPLY,
                                              sub->dbid, sub->oid, sub->name,
                                              sub->owner, InvalidOid,
                                              DSM_HANDLE_INVALID,
                                              sub->retaindeadtuples &&
                                              sub->retentionactive))
                {
                    /*
                     * We get here either if we failed to launch a worker
                     * (perhaps for resource-exhaustion reasons) or if we
                     * launched one but it immediately quit.  Either way, it
                     * seems appropriate to try again after
                     * wal_retrieve_retry_interval.
                     */
                    wait_time = Min(wait_time,
                                    wal_retrieve_retry_interval);
                }
            }
            else
            {
                wait_time = Min(wait_time,
                                wal_retrieve_retry_interval - elapsed);
            }
        }
 
        /*
         * Drop the CONFLICT_DETECTION_SLOT slot if there is no subscription
         * that requires us to retain dead tuples. Otherwise, if required,
         * advance the slot's xmin to protect dead tuples required for the
         * conflict detection.
         *
         * Additionally, if all apply workers for subscriptions with
         * retain_dead_tuples enabled have requested to stop retention, the
         * slot's xmin will be set to InvalidTransactionId allowing the
         * removal of dead tuples.
         */
        if (MyReplicationSlot)
        {
            if (!retain_dead_tuples)
                ReplicationSlotDropAcquired();
            else if (can_update_xmin)
                update_conflict_slot_xmin(xmin);
        }
 
        /* Switch back to original memory context. */
        MemoryContextSwitchTo(oldctx);
        /* Clean the temporary memory. */
        MemoryContextDelete(subctx);
 
        /* Wait for more work. */
        rc = WaitLatch(MyLatch,
                       WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
                       wait_time,
                       WAIT_EVENT_LOGICAL_LAUNCHER_MAIN);
 
        if (rc & WL_LATCH_SET)
        {
            ResetLatch(MyLatch);
            CHECK_FOR_INTERRUPTS();
        }
 
        if (ConfigReloadPending)
        {
            ConfigReloadPending = false;
            ProcessConfigFile(PGC_SIGHUP);
        }
    }
 
    /* Not reachable */
}

References acquire_conflict_slot_if_exists(), ALLOCSET_DEFAULT_SIZES, AllocSetContextCreate, ApplyLauncherGetWorkerStartTime(), ApplyLauncherSetWorkerStartTime(), Assert(), BackgroundWorkerInitializeConnection(), BackgroundWorkerUnblockSignals(), before_shmem_exit(), CHECK_FOR_INTERRUPTS, compute_min_nonremovable_xid(), ConfigReloadPending, CreateConflictDetectionSlot(), ReplicationSlot::data, Subscription::dbid, DEBUG1, DEFAULT_NAPTIME_PER_CYCLE, die, DSM_HANDLE_INVALID, Subscription::enabled, ereport, errmsg_internal(), get_subscription_list(), GetCurrentTimestamp(), init_conflict_slot_xmin(), InvalidOid, InvalidTransactionId, LogicalRepCtxStruct::launcher_pid, lfirst, logicalrep_launcher_onexit(), logicalrep_worker_find(), logicalrep_worker_launch(), LogicalRepCtx, LW_SHARED, LWLockAcquire(), LWLockRelease(), MemoryContextDelete(), MemoryContextSwitchTo(), Min, MyLatch, MyProcPid, MyReplicationSlot, Subscription::name, now(), Subscription::oid, Subscription::owner, PGC_SIGHUP, pqsignal, ProcessConfigFile(), ReplicationSlotDropAcquired(), ResetLatch(), Subscription::retaindeadtuples, Subscription::retentionactive, SIGHUP, SignalHandlerForConfigReload(), TimestampDifferenceMilliseconds(), TopMemoryContext, TransactionIdIsValid, update_conflict_slot_xmin(), WaitLatch(), wal_retrieve_retry_interval, WL_EXIT_ON_PM_DEATH, WL_LATCH_SET, WL_TIMEOUT, WORKERTYPE_APPLY, and ReplicationSlotPersistentData::xmin.

ApplyLauncherRegister()

void ApplyLauncherRegister

(

void

)

Definition at line 944 of file launcher.c.

{
    BackgroundWorker bgw;
 
    /*
     * The logical replication launcher is disabled during binary upgrades, to
     * prevent logical replication workers from running on the source cluster.
     * That could cause replication origins to move forward after having been
     * copied to the target cluster, potentially creating conflicts with the
     * copied data files.
     */
    if (max_logical_replication_workers == 0 || IsBinaryUpgrade)
        return;
 
    memset(&bgw, 0, sizeof(bgw));
    bgw.bgw_flags = BGWORKER_SHMEM_ACCESS |
        BGWORKER_BACKEND_DATABASE_CONNECTION;
    bgw.bgw_start_time = BgWorkerStart_RecoveryFinished;
    snprintf(bgw.bgw_library_name, MAXPGPATH, "postgres");
    snprintf(bgw.bgw_function_name, BGW_MAXLEN, "ApplyLauncherMain");
    snprintf(bgw.bgw_name, BGW_MAXLEN,
             "logical replication launcher");
    snprintf(bgw.bgw_type, BGW_MAXLEN,
             "logical replication launcher");
    bgw.bgw_restart_time = 5;
    bgw.bgw_notify_pid = 0;
    bgw.bgw_main_arg = (Datum) 0;
 
    RegisterBackgroundWorker(&bgw);
}

References BackgroundWorker::bgw_flags, BackgroundWorker::bgw_function_name, BackgroundWorker::bgw_library_name, BackgroundWorker::bgw_main_arg, BGW_MAXLEN, BackgroundWorker::bgw_name, BackgroundWorker::bgw_notify_pid, BackgroundWorker::bgw_restart_time, BackgroundWorker::bgw_start_time, BackgroundWorker::bgw_type, BGWORKER_BACKEND_DATABASE_CONNECTION, BGWORKER_SHMEM_ACCESS, BgWorkerStart_RecoveryFinished, IsBinaryUpgrade, max_logical_replication_workers, MAXPGPATH, RegisterBackgroundWorker(), and snprintf.

Referenced by PostmasterMain().

ApplyLauncherShmemInit()

void ApplyLauncherShmemInit

(

void

)

Definition at line 980 of file launcher.c.

{
    bool        found;
 
    LogicalRepCtx = (LogicalRepCtxStruct *)
        ShmemInitStruct("Logical Replication Launcher Data",
                        ApplyLauncherShmemSize(),
                        &found);
 
    if (!found)
    {
        int         slot;
 
        memset(LogicalRepCtx, 0, ApplyLauncherShmemSize());
 
        LogicalRepCtx->last_start_dsa = DSA_HANDLE_INVALID;
        LogicalRepCtx->last_start_dsh = DSHASH_HANDLE_INVALID;
 
        /* Initialize memory and spin locks for each worker slot. */
        for (slot = 0; slot < max_logical_replication_workers; slot++)
        {
            LogicalRepWorker *worker = &LogicalRepCtx->workers[slot];
 
            memset(worker, 0, sizeof(LogicalRepWorker));
            SpinLockInit(&worker->relmutex);
        }
    }
}

References ApplyLauncherShmemSize(), DSA_HANDLE_INVALID, DSHASH_HANDLE_INVALID, LogicalRepCtxStruct::last_start_dsa, LogicalRepCtxStruct::last_start_dsh, LogicalRepCtx, max_logical_replication_workers, LogicalRepWorker::relmutex, ShmemInitStruct(), SpinLockInit, and LogicalRepCtxStruct::workers.

Referenced by CreateOrAttachShmemStructs().

ApplyLauncherShmemSize()

Size ApplyLauncherShmemSize

(

void

)

Definition at line 925 of file launcher.c.

{
    Size        size;
 
    /*
     * Need the fixed struct and the array of LogicalRepWorker.
     */
    size = sizeof(LogicalRepCtxStruct);
    size = MAXALIGN(size);
    size = add_size(size, mul_size(max_logical_replication_workers,
                                   sizeof(LogicalRepWorker)));
    return size;
}

References add_size(), max_logical_replication_workers, MAXALIGN, and mul_size().

Referenced by ApplyLauncherShmemInit(), and CalculateShmemSize().

ApplyLauncherWakeup()

void ApplyLauncherWakeup

(

void

)

Definition at line 1141 of file launcher.c.

{
    if (LogicalRepCtx->launcher_pid != 0)
        kill(LogicalRepCtx->launcher_pid, SIGUSR1);
}

References kill, LogicalRepCtxStruct::launcher_pid, LogicalRepCtx, and SIGUSR1.

Referenced by AtEOXact_ApplyLauncher(), logicalrep_worker_onexit(), update_retention_status(), and wait_for_local_flush().

ApplyLauncherWakeupAtCommit()

void ApplyLauncherWakeupAtCommit

(

void

)

Definition at line 1131 of file launcher.c.

{
    if (!on_commit_launcher_wakeup)
        on_commit_launcher_wakeup = true;
}

References on_commit_launcher_wakeup.

Referenced by AlterSubscription(), AlterSubscriptionOwner_internal(), and CreateSubscription().

AtEOXact_ApplyLauncher()

void AtEOXact_ApplyLauncher

(

bool

isCommit

)

Definition at line 1112 of file launcher.c.

{
    if (isCommit)
    {
        if (on_commit_launcher_wakeup)
            ApplyLauncherWakeup();
    }
 
    on_commit_launcher_wakeup = false;
}

References ApplyLauncherWakeup(), and on_commit_launcher_wakeup.

Referenced by AbortTransaction(), CommitTransaction(), and PrepareTransaction().

CreateConflictDetectionSlot()

void CreateConflictDetectionSlot

(

void

)

Definition at line 1512 of file launcher.c.

{
    /* Exit early, if the replication slot is already created and acquired */
    if (MyReplicationSlot)
        return;
 
    ereport(LOG,
            errmsg("creating replication conflict detection slot"));
 
    ReplicationSlotCreate(CONFLICT_DETECTION_SLOT, false, RS_PERSISTENT, false,
                          false, false);
 
    init_conflict_slot_xmin();
}

References CONFLICT_DETECTION_SLOT, ereport, errmsg(), init_conflict_slot_xmin(), LOG, MyReplicationSlot, ReplicationSlotCreate(), and RS_PERSISTENT.

Referenced by ApplyLauncherMain(), and binary_upgrade_create_conflict_detection_slot().

GetLeaderApplyWorkerPid()

pid_t GetLeaderApplyWorkerPid

(

pid_t

pid

)

Definition at line 1541 of file launcher.c.

{
    int         leader_pid = InvalidPid;
    int         i;
 
    LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
 
    for (i = 0; i < max_logical_replication_workers; i++)
    {
        LogicalRepWorker *w = &LogicalRepCtx->workers[i];
 
        if (isParallelApplyWorker(w) && w->proc && pid == w->proc->pid)
        {
            leader_pid = w->leader_pid;
            break;
        }
    }
 
    LWLockRelease(LogicalRepWorkerLock);
 
    return leader_pid;
}

References i, InvalidPid, isParallelApplyWorker, LogicalRepWorker::leader_pid, LogicalRepCtx, LW_SHARED, LWLockAcquire(), LWLockRelease(), max_logical_replication_workers, PGPROC::pid, LogicalRepWorker::proc, and LogicalRepCtxStruct::workers.

Referenced by pg_stat_get_activity().

IsLogicalLauncher()

bool IsLogicalLauncher

(

void

)

Definition at line 1531 of file launcher.c.

{
    return LogicalRepCtx->launcher_pid == MyProcPid;
}

References LogicalRepCtxStruct::launcher_pid, LogicalRepCtx, and MyProcPid.

Referenced by ProcessInterrupts(), ReplicationSlotAcquire(), and ReplicationSlotCreate().

Variable Documentation

max_logical_replication_workers

PGDLLIMPORT int max_logical_replication_workers

extern

Definition at line 52 of file launcher.c.

Referenced by ApplyLauncherRegister(), ApplyLauncherShmemInit(), ApplyLauncherShmemSize(), GetLeaderApplyWorkerPid(), logicalrep_pa_worker_count(), logicalrep_pa_worker_stop(), logicalrep_sync_worker_count(), logicalrep_worker_attach(), logicalrep_worker_find(), logicalrep_worker_launch(), logicalrep_workers_find(), and pg_stat_get_subscription().

max_parallel_apply_workers_per_subscription

PGDLLIMPORT int max_parallel_apply_workers_per_subscription

extern

Definition at line 54 of file launcher.c.

Referenced by logicalrep_worker_launch(), and pa_free_worker().

max_sync_workers_per_subscription

PGDLLIMPORT int max_sync_workers_per_subscription

extern

Definition at line 53 of file launcher.c.

Referenced by logicalrep_worker_launch(), and process_syncing_tables_for_apply().