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)
bool	XactManipulatesLogicalReplicationWorkers (void)
void	AtEOXact_ApplyLauncher (bool isCommit)
void	AtEOSubXact_ApplyLauncher (bool isCommit, int nestDepth)
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 972 of file launcher.c.

References ALLOCSET_DEFAULT_SIZES, AllocSetContextCreate, Assert, BackgroundWorkerInitializeConnection(), BackgroundWorkerUnblockSignals(), before_shmem_exit(), CHECK_FOR_INTERRUPTS, Subscription::dbid, DEBUG1, DEFAULT_NAPTIME_PER_CYCLE, die, Subscription::enabled, ereport, errmsg(), get_subscription_list(), GetCurrentTimestamp(), got_SIGHUP, InvalidOid, LogicalRepCtxStruct::launcher_pid, lfirst, logicalrep_launcher_onexit(), logicalrep_launcher_sighup(), 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, 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("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, logicalrep_launcher_sighup);
    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 (got_SIGHUP)
        {
            got_SIGHUP = false;
            ProcessConfigFile(PGC_SIGHUP);
        }
    }

    /* Not reachable */
}

ApplyLauncherRegister()

void ApplyLauncherRegister

(

void

)

Definition at line 789 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 818 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 770 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 955 of file launcher.c.

References on_commit_launcher_wakeup.

Referenced by AlterSubscription(), and CreateSubscription().

{
    if (!on_commit_launcher_wakeup)
        on_commit_launcher_wakeup = true;
}

AtEOSubXact_ApplyLauncher()

void AtEOSubXact_ApplyLauncher	(	bool	isCommit,
		int	nestDepth
	)

Definition at line 900 of file launcher.c.

References Assert, list_concat(), list_free_deep(), StopWorkersData::nestDepth, StopWorkersData::parent, pfree(), and StopWorkersData::workers.

Referenced by AbortSubTransaction(), and CommitSubTransaction().

{
    StopWorkersData *parent;

    /* Exit immediately if there's no work to do at this level. */
    if (on_commit_stop_workers == NULL ||
        on_commit_stop_workers->nestDepth < nestDepth)
        return;

    Assert(on_commit_stop_workers->nestDepth == nestDepth);

    parent = on_commit_stop_workers->parent;

    if (isCommit)
    {
        /*
         * If the upper stack element is not an immediate parent
         * subtransaction, just decrement the notional nesting depth without
         * doing any real work.  Else, we need to merge the current workers
         * list into the parent.
         */
        if (!parent || parent->nestDepth < nestDepth - 1)
        {
            on_commit_stop_workers->nestDepth--;
            return;
        }

        parent->workers =
            list_concat(parent->workers, on_commit_stop_workers->workers);
    }
    else
    {
        /*
         * Abandon everything that was done at this nesting level.  Explicitly
         * free memory to avoid a transaction-lifespan leak.
         */
        list_free_deep(on_commit_stop_workers->workers);
    }

    /*
     * We have taken care of the current subtransaction workers list for both
     * abort or commit. So we are ready to pop the stack.
     */
    pfree(on_commit_stop_workers);
    on_commit_stop_workers = parent;
}

AtEOXact_ApplyLauncher()

void AtEOXact_ApplyLauncher

(

bool

isCommit

)

Definition at line 858 of file launcher.c.

References ApplyLauncherWakeup(), Assert, lfirst, logicalrep_worker_stop(), StopWorkersData::nestDepth, on_commit_launcher_wakeup, StopWorkersData::parent, LogicalRepWorkerId::relid, LogicalRepWorkerId::subid, LogicalRepCtxStruct::workers, and StopWorkersData::workers.

Referenced by AbortTransaction(), and CommitTransaction().

{

    Assert(on_commit_stop_workers == NULL ||
           (on_commit_stop_workers->nestDepth == 1 &&
            on_commit_stop_workers->parent == NULL));

    if (isCommit)
    {
        ListCell   *lc;

        if (on_commit_stop_workers != NULL)
        {
            List       *workers = on_commit_stop_workers->workers;

            foreach(lc, workers)
            {
                LogicalRepWorkerId *wid = lfirst(lc);

                logicalrep_worker_stop(wid->subid, wid->relid);
            }
        }

        if (on_commit_launcher_wakeup)
            ApplyLauncherWakeup();
    }

    /*
     * No need to pfree on_commit_stop_workers.  It was allocated in
     * transaction memory context, which is going to be cleaned soon.
     */
    on_commit_stop_workers = NULL;
    on_commit_launcher_wakeup = false;
}

IsLogicalLauncher()

bool IsLogicalLauncher

(

void

)

Definition at line 1088 of file launcher.c.

References LogicalRepCtxStruct::launcher_pid, and MyProcPid.

Referenced by ProcessInterrupts().

{
    return LogicalRepCtx->launcher_pid == MyProcPid;
}

XactManipulatesLogicalReplicationWorkers()

bool XactManipulatesLogicalReplicationWorkers

(

void

)

Definition at line 849 of file launcher.c.

Referenced by PrepareTransaction().

{
    return (on_commit_stop_workers != NULL);
}

Variable Documentation

max_logical_replication_workers

int max_logical_replication_workers

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

Referenced by logicalrep_worker_launch(), and process_syncing_tables_for_apply().