logicalworker.h File Reference

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)
bool	IsLogicalWorker (void)

Function Documentation

ApplyWorkerMain()

void ApplyWorkerMain

(

Datum

main_arg

)

Definition at line 3555 of file worker.c.

{
    int         worker_slot = DatumGetInt32(main_arg);
    MemoryContext oldctx;
    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);
 
    /* Run as replica session replication role. */
    SetConfigOption("session_replication_role", "replica",
                    PGC_SUSET, PGC_S_OVERRIDE);
 
    /* Connect to our database. */
    BackgroundWorkerInitializeConnectionByOid(MyLogicalRepWorker->dbid,
                                              MyLogicalRepWorker->userid,
                                              0);
 
    /*
     * Set always-secure search path, so malicious users can't redirect user
     * code (e.g. pg_index.indexprs).
     */
    SetConfigOption("search_path", "", PGC_SUSET, PGC_S_OVERRIDE);
 
    /* Load the subscription into persistent memory context. */
    ApplyContext = AllocSetContextCreate(TopMemoryContext,
                                         "ApplyContext",
                                         ALLOCSET_DEFAULT_SIZES);
    StartTransactionCommand();
    oldctx = MemoryContextSwitchTo(ApplyContext);
 
    MySubscription = GetSubscription(MyLogicalRepWorker->subid, true);
    if (!MySubscription)
    {
        ereport(LOG,
                (errmsg("logical replication apply worker for subscription %u will not "
                        "start because the subscription was removed during startup",
                        MyLogicalRepWorker->subid)));
        proc_exit(0);
    }
 
    MySubscriptionValid = true;
    MemoryContextSwitchTo(oldctx);
 
    if (!MySubscription->enabled)
    {
        ereport(LOG,
                (errmsg("logical replication apply worker for subscription \"%s\" will not "
                        "start because the subscription was disabled during startup",
                        MySubscription->name)));
 
        proc_exit(0);
    }
 
    /* Setup synchronous commit according to the user's wishes */
    SetConfigOption("synchronous_commit", MySubscription->synccommit,
                    PGC_BACKEND, PGC_S_OVERRIDE);
 
    /* Keep us informed about subscription changes. */
    CacheRegisterSyscacheCallback(SUBSCRIPTIONOID,
                                  subscription_change_cb,
                                  (Datum) 0);
 
    if (am_tablesync_worker())
        ereport(LOG,
                (errmsg("logical replication table synchronization worker for subscription \"%s\", table \"%s\" has started",
                        MySubscription->name, get_rel_name(MyLogicalRepWorker->relid))));
    else
        ereport(LOG,
                (errmsg("logical replication apply worker for subscription \"%s\" has started",
                        MySubscription->name)));
 
    CommitTransactionCommand();
 
    /* 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);
 
        /*
         * Allocate the origin name in long-lived context for error context
         * message.
         */
        ReplicationOriginNameForTablesync(MySubscription->oid,
                                          MyLogicalRepWorker->relid,
                                          originname,
                                          sizeof(originname));
        apply_error_callback_arg.origin_name = MemoryContextStrdup(ApplyContext,
                                                                   originname);
    }
    else
    {
        /* This is main 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();
        snprintf(originname, sizeof(originname), "pg_%u", MySubscription->oid);
        originid = replorigin_by_name(originname, true);
        if (!OidIsValid(originid))
            originid = replorigin_create(originname);
        replorigin_session_setup(originid);
        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);
 
        /*
         * Allocate the origin name in long-lived context for error context
         * message.
         */
        apply_error_callback_arg.origin_name = MemoryContextStrdup(ApplyContext,
                                                                   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 >= 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;
    options.proto.logical.streaming = MySubscription->stream;
    options.proto.logical.twophase = false;
 
    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("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 ALLOCSET_DEFAULT_SIZES, AllocSetContextCreate, AllTablesyncsReady(), am_tablesync_worker(), apply_error_callback_arg, ApplyContext, BackgroundWorkerInitializeConnectionByOid(), BackgroundWorkerUnblockSignals(), Subscription::binary, CacheRegisterSyscacheCallback(), CommitTransactionCommand(), Subscription::conninfo, DatumGetInt32, LogicalRepWorker::dbid, DEBUG1, die, elog, Subscription::enabled, ereport, errcode(), errmsg(), ERROR, get_rel_name(), GetCurrentTimestamp(), GetSubscription(), invalidate_syncing_table_states(), InvalidXLogRecPtr, LogicalRepWorker::last_recv_time, LogicalRepWorker::last_send_time, load_file(), LOG, LOGICALREP_PROTO_STREAM_VERSION_NUM, LOGICALREP_PROTO_TWOPHASE_VERSION_NUM, LOGICALREP_PROTO_VERSION_NUM, LOGICALREP_TWOPHASE_STATE_DISABLED, LOGICALREP_TWOPHASE_STATE_ENABLED, LOGICALREP_TWOPHASE_STATE_PENDING, logicalrep_worker_attach(), LogRepWorkerWalRcvConn, MemoryContextStrdup(), MemoryContextSwitchTo(), MyLogicalRepWorker, MySubscription, MySubscriptionValid, Subscription::name, NAMEDATALEN, Subscription::oid, OidIsValid, options, ApplyErrorCallbackArg::origin_name, PGC_BACKEND, PGC_S_OVERRIDE, PGC_SUSET, pqsignal(), proc_exit(), Subscription::publications, LogicalRepWorker::relid, ReplicationOriginNameForTablesync(), replorigin_by_name(), replorigin_create(), replorigin_session_get_progress(), replorigin_session_origin, replorigin_session_setup(), LogicalRepWorker::reply_time, server_version, SetConfigOption(), SIGHUP, SignalHandlerForConfigReload(), Subscription::slotname, snprintf, start_apply(), start_table_sync(), StartTransactionCommand(), Subscription::stream, LogicalRepWorker::subid, subscription_change_cb(), SUBSCRIPTIONOID, SUBSCRIPTIONRELMAP, Subscription::synccommit, TopMemoryContext, Subscription::twophasestate, UpdateTwoPhaseState(), LogicalRepWorker::userid, walrcv_connect, walrcv_identify_system, walrcv_server_version, and walrcv_startstreaming.

IsLogicalWorker()

bool IsLogicalWorker

(

void

)

Definition at line 3832 of file worker.c.

{
    return MyLogicalRepWorker != NULL;
}

References MyLogicalRepWorker.

Referenced by ProcessInterrupts().