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	ApplyLauncherWakeupAtCommit (void)
void	AtEOXact_ApplyLauncher (bool isCommit)
bool	IsLogicalLauncher (void)

Variables
int	max_logical_replication_workers
int	max_sync_workers_per_subscription

Function Documentation

ApplyLauncherMain()

void ApplyLauncherMain

(

Datum

main_arg

)

Definition at line 797 of file launcher.c.

References ALLOCSET_DEFAULT_SIZES, AllocSetContextCreate, Assert, BackgroundWorkerInitializeConnection(), BackgroundWorkerUnblockSignals(), before_shmem_exit(), CHECK_FOR_INTERRUPTS, ConfigReloadPending, Subscription::dbid, DEBUG1, DEFAULT_NAPTIME_PER_CYCLE, die, Subscription::enabled, ereport, errmsg_internal(), get_subscription_list(), GetCurrentTimestamp(), InvalidOid, LogicalRepCtxStruct::launcher_pid, lfirst, logicalrep_launcher_onexit(), logicalrep_worker_find(), logicalrep_worker_launch(), LW_SHARED, LWLockAcquire(), LWLockRelease(), MemoryContextDelete(), MemoryContextSwitchTo(), MyLatch, MyProcPid, Subscription::name, now(), Subscription::oid, Subscription::owner, PGC_SIGHUP, pqsignal(), ProcessConfigFile(), ResetLatch(), SIGHUP, SignalHandlerForConfigReload(), TimestampDifferenceExceeds(), TopMemoryContext, WAIT_EVENT_LOGICAL_LAUNCHER_MAIN, WaitLatch(), wal_retrieve_retry_interval, WL_EXIT_ON_PM_DEATH, WL_LATCH_SET, and WL_TIMEOUT.

{
    TimestampTz last_start_time = 0;

    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;
        TimestampTz now;
        long        wait_time = DEFAULT_NAPTIME_PER_CYCLE;

        CHECK_FOR_INTERRUPTS();

        now = GetCurrentTimestamp();

        /* Limit the start retry to once a wal_retrieve_retry_interval */
        if (TimestampDifferenceExceeds(last_start_time, now,
                                       wal_retrieve_retry_interval))
        {
            /* Use temporary context for the database list and worker info. */
            subctx = AllocSetContextCreate(TopMemoryContext,
                                           "Logical Replication Launcher sublist",
                                           ALLOCSET_DEFAULT_SIZES);
            oldctx = MemoryContextSwitchTo(subctx);

            /* search for subscriptions to start or stop. */
            sublist = get_subscription_list();

            /* Start the missing workers for enabled subscriptions. */
            foreach(lc, sublist)
            {
                Subscription *sub = (Subscription *) lfirst(lc);
                LogicalRepWorker *w;

                if (!sub->enabled)
                    continue;

                LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
                w = logicalrep_worker_find(sub->oid, InvalidOid, false);
                LWLockRelease(LogicalRepWorkerLock);

                if (w == NULL)
                {
                    last_start_time = now;
                    wait_time = wal_retrieve_retry_interval;

                    logicalrep_worker_launch(sub->dbid, sub->oid, sub->name,
                                             sub->owner, InvalidOid);
                }
            }

            /* Switch back to original memory context. */
            MemoryContextSwitchTo(oldctx);
            /* Clean the temporary memory. */
            MemoryContextDelete(subctx);
        }
        else
        {
            /*
             * The wait in previous cycle was interrupted in less than
             * wal_retrieve_retry_interval since last worker was started, this
             * usually means crash of the worker, so we should retry in
             * wal_retrieve_retry_interval again.
             */
            wait_time = wal_retrieve_retry_interval;
        }

        /* 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 */
}

ApplyLauncherRegister()

void ApplyLauncherRegister

(

void

)

Definition at line 702 of file launcher.c.

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, max_logical_replication_workers, RegisterBackgroundWorker(), and snprintf.

Referenced by PostmasterMain().

{
    BackgroundWorker bgw;

    if (max_logical_replication_workers == 0)
        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, BGW_MAXLEN, "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);
}

ApplyLauncherShmemInit()

void ApplyLauncherShmemInit

(

void

)

Definition at line 731 of file launcher.c.

References ApplyLauncherShmemSize(), max_logical_replication_workers, LogicalRepWorker::relmutex, ShmemInitStruct(), SpinLockInit, and LogicalRepCtxStruct::workers.

Referenced by CreateSharedMemoryAndSemaphores().

{
    bool        found;

    LogicalRepCtx = (LogicalRepCtxStruct *)
        ShmemInitStruct("Logical Replication Launcher Data",
                        ApplyLauncherShmemSize(),
                        &found);

    if (!found)
    {
        int         slot;

        memset(LogicalRepCtx, 0, ApplyLauncherShmemSize());

        /* 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);
        }
    }
}

ApplyLauncherShmemSize()

Size ApplyLauncherShmemSize

(

void

)

Definition at line 683 of file launcher.c.

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

Referenced by ApplyLauncherShmemInit(), and CreateSharedMemoryAndSemaphores().

{
    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;
}

ApplyLauncherWakeupAtCommit()

void ApplyLauncherWakeupAtCommit

(

void

)

Definition at line 780 of file launcher.c.

References on_commit_launcher_wakeup.

Referenced by AlterSubscription(), and CreateSubscription().

{
    if (!on_commit_launcher_wakeup)
        on_commit_launcher_wakeup = true;
}

AtEOXact_ApplyLauncher()

void AtEOXact_ApplyLauncher

(

bool

isCommit

)

Definition at line 761 of file launcher.c.

References ApplyLauncherWakeup(), and on_commit_launcher_wakeup.

Referenced by AbortTransaction(), and CommitTransaction().

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

    on_commit_launcher_wakeup = false;
}

IsLogicalLauncher()

bool IsLogicalLauncher

(

void

)

Definition at line 913 of file launcher.c.

References LogicalRepCtxStruct::launcher_pid, and MyProcPid.

Referenced by ProcessInterrupts().

{
    return LogicalRepCtx->launcher_pid == MyProcPid;
}

Variable Documentation

max_logical_replication_workers

int max_logical_replication_workers

Definition at line 54 of file launcher.c.

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

max_sync_workers_per_subscription

int max_sync_workers_per_subscription

Definition at line 55 of file launcher.c.

Referenced by logicalrep_worker_launch(), and process_syncing_tables_for_apply().