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	AtEOXact_ApplyLauncher (bool isCommit)
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 1075 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 1122 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);
 
    /* Enter main loop */
    for (;;)
    {
        int         rc;
        List       *sublist;
        ListCell   *lc;
        MemoryContext subctx;
        MemoryContext oldctx;
        long        wait_time = DEFAULT_NAPTIME_PER_CYCLE;
 
        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. */
        sublist = get_subscription_list();
        foreach(lc, sublist)
        {
            Subscription *sub = (Subscription *) lfirst(lc);
            LogicalRepWorker *w;
            TimestampTz last_start;
            TimestampTz now;
            long        elapsed;
 
            if (!sub->enabled)
                continue;
 
            LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
            w = logicalrep_worker_find(sub->oid, InvalidOid, false);
            LWLockRelease(LogicalRepWorkerLock);
 
            if (w != NULL)
                continue;       /* worker is running already */
 
            /*
             * 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);
                logicalrep_worker_launch(WORKERTYPE_APPLY,
                                         sub->dbid, sub->oid, sub->name,
                                         sub->owner, InvalidOid,
                                         DSM_HANDLE_INVALID);
            }
            else
            {
                wait_time = Min(wait_time,
                                wal_retrieve_retry_interval - elapsed);
            }
        }
 
        /* 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 ALLOCSET_DEFAULT_SIZES, AllocSetContextCreate, ApplyLauncherGetWorkerStartTime(), ApplyLauncherSetWorkerStartTime(), Assert, BackgroundWorkerInitializeConnection(), BackgroundWorkerUnblockSignals(), before_shmem_exit(), CHECK_FOR_INTERRUPTS, ConfigReloadPending, Subscription::dbid, DEBUG1, DEFAULT_NAPTIME_PER_CYCLE, die, DSM_HANDLE_INVALID, Subscription::enabled, ereport, errmsg_internal(), get_subscription_list(), GetCurrentTimestamp(), InvalidOid, LogicalRepCtxStruct::launcher_pid, lfirst, logicalrep_launcher_onexit(), logicalrep_worker_find(), logicalrep_worker_launch(), LogicalRepCtx, LW_SHARED, LWLockAcquire(), LWLockRelease(), MemoryContextDelete(), MemoryContextSwitchTo(), Min, MyLatch, MyProcPid, Subscription::name, now(), Subscription::oid, Subscription::owner, PGC_SIGHUP, pqsignal(), ProcessConfigFile(), ResetLatch(), SIGHUP, SignalHandlerForConfigReload(), TimestampDifferenceMilliseconds(), TopMemoryContext, WaitLatch(), wal_retrieve_retry_interval, WL_EXIT_ON_PM_DEATH, WL_LATCH_SET, WL_TIMEOUT, and WORKERTYPE_APPLY.

ApplyLauncherRegister()

void ApplyLauncherRegister

(

void

)

Definition at line 918 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 954 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 899 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, mul_size(), and size.

Referenced by ApplyLauncherShmemInit(), and CalculateShmemSize().

ApplyLauncherWakeupAtCommit()

void ApplyLauncherWakeupAtCommit

(

void

)

Definition at line 1105 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 1086 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().

GetLeaderApplyWorkerPid()

pid_t GetLeaderApplyWorkerPid

(

pid_t

pid

)

Definition at line 1253 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 1243 of file launcher.c.

{
    return LogicalRepCtx->launcher_pid == MyProcPid;
}

References LogicalRepCtxStruct::launcher_pid, LogicalRepCtx, and MyProcPid.

Referenced by ProcessInterrupts().

Variable Documentation

max_logical_replication_workers

PGDLLIMPORT int max_logical_replication_workers

extern

Definition at line 50 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 52 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 51 of file launcher.c.

Referenced by logicalrep_worker_launch(), and process_syncing_tables_for_apply().