logicalworker.h File Reference

#include <signal.h>

Include dependency graph for logicalworker.h:

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

Go to the source code of this file.

Functions
void	ApplyWorkerMain (Datum main_arg)
void	ParallelApplyWorkerMain (Datum main_arg)
bool	IsLogicalWorker (void)
bool	IsLogicalParallelApplyWorker (void)
void	HandleParallelApplyMessageInterrupt (void)
void	HandleParallelApplyMessages (void)
void	LogicalRepWorkersWakeupAtCommit (Oid subid)
void	AtEOXact_LogicalRepWorkers (bool isCommit)

Variables
PGDLLIMPORT volatile sig_atomic_t	ParallelApplyMessagePending

Function Documentation

ApplyWorkerMain()

void ApplyWorkerMain

(

Datum

main_arg

)

Definition at line 4473 of file worker.c.

{
    int         worker_slot = DatumGetInt32(main_arg);
    char        originname[NAMEDATALEN];
    XLogRecPtr  origin_startpos = InvalidXLogRecPtr;
    char       *myslotname = NULL;
    WalRcvStreamOptions options;
    int         server_version;
 
    /* Attach to slot */
    logicalrep_worker_attach(worker_slot);
 
    /* Setup signal handling */
    pqsignal(SIGHUP, SignalHandlerForConfigReload);
    pqsignal(SIGTERM, die);
    BackgroundWorkerUnblockSignals();
 
    /*
     * We don't currently need any ResourceOwner in a walreceiver process, but
     * if we did, we could call CreateAuxProcessResourceOwner here.
     */
 
    /* Initialise stats to a sanish value */
    MyLogicalRepWorker->last_send_time = MyLogicalRepWorker->last_recv_time =
        MyLogicalRepWorker->reply_time = GetCurrentTimestamp();
 
    /* Load the libpq-specific functions */
    load_file("libpqwalreceiver", false);
 
    InitializeApplyWorker();
 
    /* Connect to the origin and start the replication. */
    elog(DEBUG1, "connecting to publisher using connection string \"%s\"",
         MySubscription->conninfo);
 
    if (am_tablesync_worker())
    {
        start_table_sync(&origin_startpos, &myslotname);
 
        ReplicationOriginNameForLogicalRep(MySubscription->oid,
                                           MyLogicalRepWorker->relid,
                                           originname,
                                           sizeof(originname));
        set_apply_error_context_origin(originname);
    }
    else
    {
        /* This is the leader apply worker */
        RepOriginId originid;
        TimeLineID  startpointTLI;
        char       *err;
 
        myslotname = MySubscription->slotname;
 
        /*
         * This shouldn't happen if the subscription is enabled, but guard
         * against DDL bugs or manual catalog changes.  (libpqwalreceiver will
         * crash if slot is NULL.)
         */
        if (!myslotname)
            ereport(ERROR,
                    (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                     errmsg("subscription has no replication slot set")));
 
        /* Setup replication origin tracking. */
        StartTransactionCommand();
        ReplicationOriginNameForLogicalRep(MySubscription->oid, InvalidOid,
                                           originname, sizeof(originname));
        originid = replorigin_by_name(originname, true);
        if (!OidIsValid(originid))
            originid = replorigin_create(originname);
        replorigin_session_setup(originid, 0);
        replorigin_session_origin = originid;
        origin_startpos = replorigin_session_get_progress(false);
        CommitTransactionCommand();
 
        LogRepWorkerWalRcvConn = walrcv_connect(MySubscription->conninfo, true,
                                                MySubscription->name, &err);
        if (LogRepWorkerWalRcvConn == NULL)
            ereport(ERROR,
                    (errcode(ERRCODE_CONNECTION_FAILURE),
                     errmsg("could not connect to the publisher: %s", err)));
 
        /*
         * We don't really use the output identify_system for anything but it
         * does some initializations on the upstream so let's still call it.
         */
        (void) walrcv_identify_system(LogRepWorkerWalRcvConn, &startpointTLI);
 
        set_apply_error_context_origin(originname);
    }
 
    /*
     * Setup callback for syscache so that we know when something changes in
     * the subscription relation state.
     */
    CacheRegisterSyscacheCallback(SUBSCRIPTIONRELMAP,
                                  invalidate_syncing_table_states,
                                  (Datum) 0);
 
    /* Build logical replication streaming options. */
    options.logical = true;
    options.startpoint = origin_startpos;
    options.slotname = myslotname;
 
    server_version = walrcv_server_version(LogRepWorkerWalRcvConn);
    options.proto.logical.proto_version =
        server_version >= 160000 ? LOGICALREP_PROTO_STREAM_PARALLEL_VERSION_NUM :
        server_version >= 150000 ? LOGICALREP_PROTO_TWOPHASE_VERSION_NUM :
        server_version >= 140000 ? LOGICALREP_PROTO_STREAM_VERSION_NUM :
        LOGICALREP_PROTO_VERSION_NUM;
 
    options.proto.logical.publication_names = MySubscription->publications;
    options.proto.logical.binary = MySubscription->binary;
 
    /*
     * Assign the appropriate option value for streaming option according to
     * the 'streaming' mode and the publisher's ability to support that mode.
     */
    if (server_version >= 160000 &&
        MySubscription->stream == LOGICALREP_STREAM_PARALLEL)
    {
        options.proto.logical.streaming_str = "parallel";
        MyLogicalRepWorker->parallel_apply = true;
    }
    else if (server_version >= 140000 &&
             MySubscription->stream != LOGICALREP_STREAM_OFF)
    {
        options.proto.logical.streaming_str = "on";
        MyLogicalRepWorker->parallel_apply = false;
    }
    else
    {
        options.proto.logical.streaming_str = NULL;
        MyLogicalRepWorker->parallel_apply = false;
    }
 
    options.proto.logical.twophase = false;
    options.proto.logical.origin = pstrdup(MySubscription->origin);
 
    if (!am_tablesync_worker())
    {
        /*
         * Even when the two_phase mode is requested by the user, it remains
         * as the tri-state PENDING until all tablesyncs have reached READY
         * state. Only then, can it become ENABLED.
         *
         * Note: If the subscription has no tables then leave the state as
         * PENDING, which allows ALTER SUBSCRIPTION ... REFRESH PUBLICATION to
         * work.
         */
        if (MySubscription->twophasestate == LOGICALREP_TWOPHASE_STATE_PENDING &&
            AllTablesyncsReady())
        {
            /* Start streaming with two_phase enabled */
            options.proto.logical.twophase = true;
            walrcv_startstreaming(LogRepWorkerWalRcvConn, &options);
 
            StartTransactionCommand();
            UpdateTwoPhaseState(MySubscription->oid, LOGICALREP_TWOPHASE_STATE_ENABLED);
            MySubscription->twophasestate = LOGICALREP_TWOPHASE_STATE_ENABLED;
            CommitTransactionCommand();
        }
        else
        {
            walrcv_startstreaming(LogRepWorkerWalRcvConn, &options);
        }
 
        ereport(DEBUG1,
                (errmsg_internal("logical replication apply worker for subscription \"%s\" two_phase is %s",
                        MySubscription->name,
                        MySubscription->twophasestate == LOGICALREP_TWOPHASE_STATE_DISABLED ? "DISABLED" :
                        MySubscription->twophasestate == LOGICALREP_TWOPHASE_STATE_PENDING ? "PENDING" :
                        MySubscription->twophasestate == LOGICALREP_TWOPHASE_STATE_ENABLED ? "ENABLED" :
                        "?")));
    }
    else
    {
        /* Start normal logical streaming replication. */
        walrcv_startstreaming(LogRepWorkerWalRcvConn, &options);
    }
 
    /* Run the main loop. */
    start_apply(origin_startpos);
 
    proc_exit(0);
}

References AllTablesyncsReady(), am_tablesync_worker(), BackgroundWorkerUnblockSignals(), Subscription::binary, CacheRegisterSyscacheCallback(), CommitTransactionCommand(), Subscription::conninfo, DatumGetInt32(), DEBUG1, die, elog(), ereport, err(), errcode(), errmsg(), errmsg_internal(), ERROR, GetCurrentTimestamp(), InitializeApplyWorker(), invalidate_syncing_table_states(), InvalidOid, InvalidXLogRecPtr, LogicalRepWorker::last_recv_time, LogicalRepWorker::last_send_time, load_file(), LOGICALREP_PROTO_STREAM_PARALLEL_VERSION_NUM, LOGICALREP_PROTO_STREAM_VERSION_NUM, LOGICALREP_PROTO_TWOPHASE_VERSION_NUM, LOGICALREP_PROTO_VERSION_NUM, LOGICALREP_STREAM_OFF, LOGICALREP_STREAM_PARALLEL, LOGICALREP_TWOPHASE_STATE_DISABLED, LOGICALREP_TWOPHASE_STATE_ENABLED, LOGICALREP_TWOPHASE_STATE_PENDING, logicalrep_worker_attach(), LogRepWorkerWalRcvConn, MyLogicalRepWorker, MySubscription, Subscription::name, NAMEDATALEN, Subscription::oid, OidIsValid, options, Subscription::origin, LogicalRepWorker::parallel_apply, pqsignal(), proc_exit(), pstrdup(), Subscription::publications, LogicalRepWorker::relid, ReplicationOriginNameForLogicalRep(), replorigin_by_name(), replorigin_create(), replorigin_session_get_progress(), replorigin_session_origin, replorigin_session_setup(), LogicalRepWorker::reply_time, server_version, set_apply_error_context_origin(), SIGHUP, SignalHandlerForConfigReload(), Subscription::slotname, start_apply(), start_table_sync(), StartTransactionCommand(), Subscription::stream, SUBSCRIPTIONRELMAP, Subscription::twophasestate, UpdateTwoPhaseState(), walrcv_connect, walrcv_identify_system, walrcv_server_version, and walrcv_startstreaming.

AtEOXact_LogicalRepWorkers()

void AtEOXact_LogicalRepWorkers

(

bool

isCommit

)

Definition at line 4964 of file worker.c.

{
    if (isCommit && on_commit_wakeup_workers_subids != NIL)
    {
        ListCell   *lc;
 
        LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
        foreach(lc, on_commit_wakeup_workers_subids)
        {
            Oid         subid = lfirst_oid(lc);
            List       *workers;
            ListCell   *lc2;
 
            workers = logicalrep_workers_find(subid, true);
            foreach(lc2, workers)
            {
                LogicalRepWorker *worker = (LogicalRepWorker *) lfirst(lc2);
 
                logicalrep_worker_wakeup_ptr(worker);
            }
        }
        LWLockRelease(LogicalRepWorkerLock);
    }
 
    /* The List storage will be reclaimed automatically in xact cleanup. */
    on_commit_wakeup_workers_subids = NIL;
}

References lfirst, lfirst_oid, logicalrep_worker_wakeup_ptr(), logicalrep_workers_find(), LW_SHARED, LWLockAcquire(), LWLockRelease(), NIL, and on_commit_wakeup_workers_subids.

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

HandleParallelApplyMessageInterrupt()

void HandleParallelApplyMessageInterrupt

(

void

)

Definition at line 986 of file applyparallelworker.c.

{
    InterruptPending = true;
    ParallelApplyMessagePending = true;
    SetLatch(MyLatch);
}

References InterruptPending, MyLatch, ParallelApplyMessagePending, and SetLatch().

Referenced by procsignal_sigusr1_handler().

HandleParallelApplyMessages()

void HandleParallelApplyMessages

(

void

)

Definition at line 1060 of file applyparallelworker.c.

{
    ListCell   *lc;
    MemoryContext oldcontext;
 
    static MemoryContext hpam_context = NULL;
 
    /*
     * This is invoked from ProcessInterrupts(), and since some of the
     * functions it calls contain CHECK_FOR_INTERRUPTS(), there is a potential
     * for recursive calls if more signals are received while this runs. It's
     * unclear that recursive entry would be safe, and it doesn't seem useful
     * even if it is safe, so let's block interrupts until done.
     */
    HOLD_INTERRUPTS();
 
    /*
     * Moreover, CurrentMemoryContext might be pointing almost anywhere. We
     * don't want to risk leaking data into long-lived contexts, so let's do
     * our work here in a private context that we can reset on each use.
     */
    if (!hpam_context)          /* first time through? */
        hpam_context = AllocSetContextCreate(TopMemoryContext,
                                             "HandleParallelApplyMessages",
                                             ALLOCSET_DEFAULT_SIZES);
    else
        MemoryContextReset(hpam_context);
 
    oldcontext = MemoryContextSwitchTo(hpam_context);
 
    ParallelApplyMessagePending = false;
 
    foreach(lc, ParallelApplyWorkerPool)
    {
        shm_mq_result res;
        Size        nbytes;
        void       *data;
        ParallelApplyWorkerInfo *winfo = (ParallelApplyWorkerInfo *) lfirst(lc);
 
        /*
         * The leader will detach from the error queue and set it to NULL
         * before preparing to stop all parallel apply workers, so we don't
         * need to handle error messages anymore. See
         * logicalrep_worker_detach.
         */
        if (!winfo->error_mq_handle)
            continue;
 
        res = shm_mq_receive(winfo->error_mq_handle, &nbytes, &data, true);
 
        if (res == SHM_MQ_WOULD_BLOCK)
            continue;
        else if (res == SHM_MQ_SUCCESS)
        {
            StringInfoData msg;
 
            initStringInfo(&msg);
            appendBinaryStringInfo(&msg, data, nbytes);
            HandleParallelApplyMessage(&msg);
            pfree(msg.data);
        }
        else
            ereport(ERROR,
                    (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                     errmsg("lost connection to the logical replication parallel apply worker")));
    }
 
    MemoryContextSwitchTo(oldcontext);
 
    /* Might as well clear the context on our way out */
    MemoryContextReset(hpam_context);
 
    RESUME_INTERRUPTS();
}

References ALLOCSET_DEFAULT_SIZES, AllocSetContextCreate, appendBinaryStringInfo(), StringInfoData::data, data, ereport, errcode(), errmsg(), ERROR, ParallelApplyWorkerInfo::error_mq_handle, HandleParallelApplyMessage(), HOLD_INTERRUPTS, initStringInfo(), lfirst, MemoryContextReset(), MemoryContextSwitchTo(), ParallelApplyMessagePending, ParallelApplyWorkerPool, pfree(), res, RESUME_INTERRUPTS, shm_mq_receive(), SHM_MQ_SUCCESS, SHM_MQ_WOULD_BLOCK, and TopMemoryContext.

Referenced by ProcessInterrupts().

IsLogicalParallelApplyWorker()

bool IsLogicalParallelApplyWorker

(

void

)

Definition at line 4714 of file worker.c.

{
    return IsLogicalWorker() && am_parallel_apply_worker();
}

References am_parallel_apply_worker(), and IsLogicalWorker().

Referenced by mq_putmessage().

IsLogicalWorker()

bool IsLogicalWorker

(

void

)

Definition at line 4705 of file worker.c.

{
    return MyLogicalRepWorker != NULL;
}

References MyLogicalRepWorker.

Referenced by IsLogicalParallelApplyWorker(), and ProcessInterrupts().

LogicalRepWorkersWakeupAtCommit()

void LogicalRepWorkersWakeupAtCommit

(

Oid

subid

)

Definition at line 4950 of file worker.c.

{
    MemoryContext oldcxt;
 
    oldcxt = MemoryContextSwitchTo(TopTransactionContext);
    on_commit_wakeup_workers_subids =
        list_append_unique_oid(on_commit_wakeup_workers_subids, subid);
    MemoryContextSwitchTo(oldcxt);
}

References list_append_unique_oid(), MemoryContextSwitchTo(), on_commit_wakeup_workers_subids, and TopTransactionContext.

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

ParallelApplyWorkerMain()

void ParallelApplyWorkerMain

(

Datum

main_arg

)

Definition at line 863 of file applyparallelworker.c.

{
    ParallelApplyWorkerShared *shared;
    dsm_handle  handle;
    dsm_segment *seg;
    shm_toc    *toc;
    shm_mq     *mq;
    shm_mq_handle *mqh;
    shm_mq_handle *error_mqh;
    RepOriginId originid;
    int         worker_slot = DatumGetInt32(main_arg);
    char        originname[NAMEDATALEN];
 
    /* Setup signal handling. */
    pqsignal(SIGHUP, SignalHandlerForConfigReload);
    pqsignal(SIGINT, SignalHandlerForShutdownRequest);
    pqsignal(SIGTERM, die);
    BackgroundWorkerUnblockSignals();
 
    /*
     * Attach to the dynamic shared memory segment for the parallel apply, and
     * find its table of contents.
     *
     * Like parallel query, we don't need resource owner by this time. See
     * ParallelWorkerMain.
     */
    memcpy(&handle, MyBgworkerEntry->bgw_extra, sizeof(dsm_handle));
    seg = dsm_attach(handle);
    if (!seg)
        ereport(ERROR,
                (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                 errmsg("unable to map dynamic shared memory segment")));
 
    toc = shm_toc_attach(PG_LOGICAL_APPLY_SHM_MAGIC, dsm_segment_address(seg));
    if (!toc)
        ereport(ERROR,
                (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                 errmsg("bad magic number in dynamic shared memory segment")));
 
    before_shmem_exit(pa_shutdown, PointerGetDatum(seg));
 
    /* Look up the shared information. */
    shared = shm_toc_lookup(toc, PARALLEL_APPLY_KEY_SHARED, false);
    MyParallelShared = shared;
 
    /*
     * Attach to the message queue.
     */
    mq = shm_toc_lookup(toc, PARALLEL_APPLY_KEY_MQ, false);
    shm_mq_set_receiver(mq, MyProc);
    mqh = shm_mq_attach(mq, seg, NULL);
 
    /*
     * Primary initialization is complete. Now, we can attach to our slot.
     * This is to ensure that the leader apply worker does not write data to
     * the uninitialized memory queue.
     */
    logicalrep_worker_attach(worker_slot);
 
    SpinLockAcquire(&MyParallelShared->mutex);
    MyParallelShared->logicalrep_worker_generation = MyLogicalRepWorker->generation;
    MyParallelShared->logicalrep_worker_slot_no = worker_slot;
    SpinLockRelease(&MyParallelShared->mutex);
 
    /*
     * Attach to the error queue.
     */
    mq = shm_toc_lookup(toc, PARALLEL_APPLY_KEY_ERROR_QUEUE, false);
    shm_mq_set_sender(mq, MyProc);
    error_mqh = shm_mq_attach(mq, seg, NULL);
 
    pq_redirect_to_shm_mq(seg, error_mqh);
    pq_set_parallel_leader(MyLogicalRepWorker->leader_pid,
                           InvalidBackendId);
 
    MyLogicalRepWorker->last_send_time = MyLogicalRepWorker->last_recv_time =
        MyLogicalRepWorker->reply_time = 0;
 
    InitializeApplyWorker();
 
    /* Setup replication origin tracking. */
    StartTransactionCommand();
    ReplicationOriginNameForLogicalRep(MySubscription->oid, InvalidOid,
                                       originname, sizeof(originname));
    originid = replorigin_by_name(originname, false);
 
    /*
     * The parallel apply worker doesn't need to monopolize this replication
     * origin which was already acquired by its leader process.
     */
    replorigin_session_setup(originid, MyLogicalRepWorker->leader_pid);
    replorigin_session_origin = originid;
    CommitTransactionCommand();
 
    /*
     * Setup callback for syscache so that we know when something changes in
     * the subscription relation state.
     */
    CacheRegisterSyscacheCallback(SUBSCRIPTIONRELMAP,
                                  invalidate_syncing_table_states,
                                  (Datum) 0);
 
    set_apply_error_context_origin(originname);
 
    LogicalParallelApplyLoop(mqh);
 
    /*
     * The parallel apply worker must not get here because the parallel apply
     * worker will only stop when it receives a SIGTERM or SIGINT from the
     * leader, or when there is an error. None of these cases will allow the
     * code to reach here.
     */
    Assert(false);
}

References Assert(), BackgroundWorkerUnblockSignals(), before_shmem_exit(), BackgroundWorker::bgw_extra, CacheRegisterSyscacheCallback(), CommitTransactionCommand(), DatumGetInt32(), die, dsm_attach(), dsm_segment_address(), ereport, errcode(), errmsg(), ERROR, LogicalRepWorker::generation, InitializeApplyWorker(), invalidate_syncing_table_states(), InvalidBackendId, InvalidOid, LogicalRepWorker::last_recv_time, LogicalRepWorker::last_send_time, LogicalRepWorker::leader_pid, LogicalParallelApplyLoop(), logicalrep_worker_attach(), ParallelApplyWorkerShared::logicalrep_worker_generation, ParallelApplyWorkerShared::logicalrep_worker_slot_no, ParallelApplyWorkerShared::mutex, MyBgworkerEntry, MyLogicalRepWorker, MyParallelShared, MyProc, MySubscription, NAMEDATALEN, Subscription::oid, pa_shutdown(), PARALLEL_APPLY_KEY_ERROR_QUEUE, PARALLEL_APPLY_KEY_MQ, PARALLEL_APPLY_KEY_SHARED, PG_LOGICAL_APPLY_SHM_MAGIC, PointerGetDatum(), pq_redirect_to_shm_mq(), pq_set_parallel_leader(), pqsignal(), ReplicationOriginNameForLogicalRep(), replorigin_by_name(), replorigin_session_origin, replorigin_session_setup(), LogicalRepWorker::reply_time, set_apply_error_context_origin(), shm_mq_attach(), shm_mq_set_receiver(), shm_mq_set_sender(), shm_toc_attach(), shm_toc_lookup(), SIGHUP, SignalHandlerForConfigReload(), SignalHandlerForShutdownRequest(), SpinLockAcquire, SpinLockRelease, StartTransactionCommand(), and SUBSCRIPTIONRELMAP.

Variable Documentation

ParallelApplyMessagePending

PGDLLIMPORT volatile sig_atomic_t ParallelApplyMessagePending

extern

Definition at line 245 of file applyparallelworker.c.

Referenced by HandleParallelApplyMessageInterrupt(), HandleParallelApplyMessages(), and ProcessInterrupts().