#include "access/xlogdefs.h"
#include "catalog/pg_subscription.h"
#include "datatype/timestamp.h"
#include "miscadmin.h"
#include "replication/logicalrelation.h"
#include "replication/walreceiver.h"
#include "storage/buffile.h"
#include "storage/fileset.h"
#include "storage/lock.h"
#include "storage/shm_mq.h"
#include "storage/shm_toc.h"
#include "storage/spin.h"

Include dependency graph for worker_internal.h:

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

Go to the source code of this file.

Data Structures
struct	LogicalRepWorker

struct	ParallelApplyWorkerShared

struct	ParallelApplyWorkerInfo

Macros
#define	isParallelApplyWorker(worker)

#define	isTablesyncWorker(worker)

Typedefs
typedef enum LogicalRepWorkerType	LogicalRepWorkerType

typedef struct LogicalRepWorker	LogicalRepWorker

typedef enum ParallelTransState	ParallelTransState

typedef enum PartialFileSetState	PartialFileSetState

typedef struct ParallelApplyWorkerShared	ParallelApplyWorkerShared

typedef struct ParallelApplyWorkerInfo	ParallelApplyWorkerInfo

Enumerations
enum	LogicalRepWorkerType { WORKERTYPE_UNKNOWN = 0 , WORKERTYPE_TABLESYNC , WORKERTYPE_APPLY , WORKERTYPE_PARALLEL_APPLY }

enum	ParallelTransState { PARALLEL_TRANS_UNKNOWN , PARALLEL_TRANS_STARTED , PARALLEL_TRANS_FINISHED }

enum	PartialFileSetState { FS_EMPTY , FS_SERIALIZE_IN_PROGRESS , FS_SERIALIZE_DONE , FS_READY }

Functions
void	logicalrep_worker_attach (int slot)

LogicalRepWorker *	logicalrep_worker_find (Oid subid, Oid relid, bool only_running)

List *	logicalrep_workers_find (Oid subid, bool only_running, bool acquire_lock)

bool	logicalrep_worker_launch (LogicalRepWorkerType wtype, Oid dbid, Oid subid, const char *subname, Oid userid, Oid relid, dsm_handle subworker_dsm)

void	logicalrep_worker_stop (Oid subid, Oid relid)

void	logicalrep_pa_worker_stop (ParallelApplyWorkerInfo *winfo)

void	logicalrep_worker_wakeup (Oid subid, Oid relid)

void	logicalrep_worker_wakeup_ptr (LogicalRepWorker *worker)

int	logicalrep_sync_worker_count (Oid subid)

void	ReplicationOriginNameForLogicalRep (Oid suboid, Oid relid, char *originname, Size szoriginname)

bool	AllTablesyncsReady (void)

void	UpdateTwoPhaseState (Oid suboid, char new_state)

void	process_syncing_tables (XLogRecPtr current_lsn)

void	invalidate_syncing_table_states (Datum arg, int cacheid, uint32 hashvalue)

void	stream_start_internal (TransactionId xid, bool first_segment)

void	stream_stop_internal (TransactionId xid)

void	apply_spooled_messages (FileSet *stream_fileset, TransactionId xid, XLogRecPtr lsn)

void	apply_dispatch (StringInfo s)

void	maybe_reread_subscription (void)

void	stream_cleanup_files (Oid subid, TransactionId xid)

void	set_stream_options (WalRcvStreamOptions options, char slotname, XLogRecPtr *origin_startpos)

void	start_apply (XLogRecPtr origin_startpos)

void	InitializeLogRepWorker (void)

void	SetupApplyOrSyncWorker (int worker_slot)

void	DisableSubscriptionAndExit (void)

void	store_flush_position (XLogRecPtr remote_lsn, XLogRecPtr local_lsn)

void	apply_error_callback (void *arg)

void	set_apply_error_context_origin (char *originname)

void	pa_allocate_worker (TransactionId xid)

ParallelApplyWorkerInfo *	pa_find_worker (TransactionId xid)

void	pa_detach_all_error_mq (void)

bool	pa_send_data (ParallelApplyWorkerInfo winfo, Size nbytes, const void data)

void	pa_switch_to_partial_serialize (ParallelApplyWorkerInfo *winfo, bool stream_locked)

void	pa_set_xact_state (ParallelApplyWorkerShared *wshared, ParallelTransState xact_state)

void	pa_set_stream_apply_worker (ParallelApplyWorkerInfo *winfo)

void	pa_start_subtrans (TransactionId current_xid, TransactionId top_xid)

void	pa_reset_subtrans (void)

void	pa_stream_abort (LogicalRepStreamAbortData *abort_data)

void	pa_set_fileset_state (ParallelApplyWorkerShared *wshared, PartialFileSetState fileset_state)

void	pa_lock_stream (TransactionId xid, LOCKMODE lockmode)

void	pa_unlock_stream (TransactionId xid, LOCKMODE lockmode)

void	pa_lock_transaction (TransactionId xid, LOCKMODE lockmode)

void	pa_unlock_transaction (TransactionId xid, LOCKMODE lockmode)

void	pa_decr_and_wait_stream_block (void)

void	pa_xact_finish (ParallelApplyWorkerInfo *winfo, XLogRecPtr remote_lsn)

static bool	am_tablesync_worker (void)

static bool	am_leader_apply_worker (void)

static bool	am_parallel_apply_worker (void)

Macro Definition Documentation

◆ isParallelApplyWorker

#define isParallelApplyWorker ( worker )

Value:

((worker)->in_use && \

(worker)->type == WORKERTYPE_PARALLEL_APPLY)

WORKERTYPE_PARALLEL_APPLY

@ WORKERTYPE_PARALLEL_APPLY

Definition: worker_internal.h:34

Definition at line 329 of file worker_internal.h.

◆ isTablesyncWorker

#define isTablesyncWorker ( worker )

Value:

((worker)->in_use && \

(worker)->type == WORKERTYPE_TABLESYNC)

WORKERTYPE_TABLESYNC

@ WORKERTYPE_TABLESYNC

Definition: worker_internal.h:32

Definition at line 331 of file worker_internal.h.

Typedef Documentation

◆ LogicalRepWorker

typedef struct LogicalRepWorker LogicalRepWorker

◆ LogicalRepWorkerType

typedef enum LogicalRepWorkerType LogicalRepWorkerType

◆ ParallelApplyWorkerInfo

typedef struct ParallelApplyWorkerInfo ParallelApplyWorkerInfo

◆ ParallelApplyWorkerShared

typedef struct ParallelApplyWorkerShared ParallelApplyWorkerShared

◆ ParallelTransState

typedef enum ParallelTransState ParallelTransState

◆ PartialFileSetState

typedef enum PartialFileSetState PartialFileSetState

Enumeration Type Documentation

◆ LogicalRepWorkerType

enum LogicalRepWorkerType

Enumerator
WORKERTYPE_UNKNOWN
WORKERTYPE_TABLESYNC
WORKERTYPE_APPLY
WORKERTYPE_PARALLEL_APPLY

Definition at line 29 of file worker_internal.h.

{
    WORKERTYPE_UNKNOWN = 0,
    WORKERTYPE_TABLESYNC,
    WORKERTYPE_APPLY,
    WORKERTYPE_PARALLEL_APPLY,
} LogicalRepWorkerType;

◆ ParallelTransState

enum ParallelTransState

Enumerator
PARALLEL_TRANS_UNKNOWN
PARALLEL_TRANS_STARTED
PARALLEL_TRANS_FINISHED

Definition at line 103 of file worker_internal.h.

{
    PARALLEL_TRANS_UNKNOWN,
    PARALLEL_TRANS_STARTED,
    PARALLEL_TRANS_FINISHED,
} ParallelTransState;

◆ PartialFileSetState

enum PartialFileSetState

Enumerator
FS_EMPTY
FS_SERIALIZE_IN_PROGRESS
FS_SERIALIZE_DONE
FS_READY

Definition at line 126 of file worker_internal.h.

{
    FS_EMPTY,
    FS_SERIALIZE_IN_PROGRESS,
    FS_SERIALIZE_DONE,
    FS_READY,
} PartialFileSetState;

Function Documentation

◆ AllTablesyncsReady()

bool AllTablesyncsReady ( void )

Definition at line 1738 of file tablesync.c.

{
    bool        started_tx = false;
    bool        has_subrels = false;
 
    /* We need up-to-date sync state info for subscription tables here. */
    has_subrels = FetchTableStates(&started_tx);
 
    if (started_tx)
    {
        CommitTransactionCommand();
        pgstat_report_stat(true);
    }
 
    /*
     * Return false when there are no tables in subscription or not all tables
     * are in ready state; true otherwise.
     */
    return has_subrels && (table_states_not_ready == NIL);
}

References CommitTransactionCommand(), FetchTableStates(), NIL, pgstat_report_stat(), and table_states_not_ready.

Referenced by pa_can_start(), process_syncing_tables_for_apply(), and run_apply_worker().

◆ am_leader_apply_worker()

static bool am_leader_apply_worker ( void )

inlinestatic

Definition at line 341 of file worker_internal.h.

{
    Assert(MyLogicalRepWorker->in_use);
    return (MyLogicalRepWorker->type == WORKERTYPE_APPLY);
}

References Assert(), LogicalRepWorker::in_use, MyLogicalRepWorker, LogicalRepWorker::type, and WORKERTYPE_APPLY.

Referenced by apply_worker_exit(), DisableSubscriptionAndExit(), InitializeLogRepWorker(), logicalrep_worker_detach(), maybe_reread_subscription(), pa_can_start(), pa_set_fileset_state(), pa_xact_finish(), and SetupApplyOrSyncWorker().

◆ am_parallel_apply_worker()

static bool am_parallel_apply_worker ( void )

inlinestatic

Definition at line 348 of file worker_internal.h.

{
    Assert(MyLogicalRepWorker->in_use);
    return isParallelApplyWorker(MyLogicalRepWorker);
}

References Assert(), LogicalRepWorker::in_use, isParallelApplyWorker, and MyLogicalRepWorker.

Referenced by apply_spooled_messages(), apply_worker_exit(), clear_subscription_skip_lsn(), get_transaction_apply_action(), IsLogicalParallelApplyWorker(), maybe_reread_subscription(), pa_decr_and_wait_stream_block(), pa_free_worker(), pa_get_fileset_state(), and store_flush_position().

◆ am_tablesync_worker()

static bool am_tablesync_worker ( void )

inlinestatic

Definition at line 335 of file worker_internal.h.

{
    return isTablesyncWorker(MyLogicalRepWorker);
}

References isTablesyncWorker, and MyLogicalRepWorker.

Referenced by apply_handle_begin_prepare(), apply_handle_origin(), apply_handle_stream_prepare(), DisableSubscriptionAndExit(), InitializeLogRepWorker(), SetupApplyOrSyncWorker(), start_apply(), and start_table_sync().

◆ apply_dispatch()

void apply_dispatch ( StringInfo s )

Definition at line 3375 of file worker.c.

{
    LogicalRepMsgType action = pq_getmsgbyte(s);
    LogicalRepMsgType saved_command;
 
    /*
     * Set the current command being applied. Since this function can be
     * called recursively when applying spooled changes, save the current
     * command.
     */
    saved_command = apply_error_callback_arg.command;
    apply_error_callback_arg.command = action;
 
    switch (action)
    {
        case LOGICAL_REP_MSG_BEGIN:
            apply_handle_begin(s);
            break;
 
        case LOGICAL_REP_MSG_COMMIT:
            apply_handle_commit(s);
            break;
 
        case LOGICAL_REP_MSG_INSERT:
            apply_handle_insert(s);
            break;
 
        case LOGICAL_REP_MSG_UPDATE:
            apply_handle_update(s);
            break;
 
        case LOGICAL_REP_MSG_DELETE:
            apply_handle_delete(s);
            break;
 
        case LOGICAL_REP_MSG_TRUNCATE:
            apply_handle_truncate(s);
            break;
 
        case LOGICAL_REP_MSG_RELATION:
            apply_handle_relation(s);
            break;
 
        case LOGICAL_REP_MSG_TYPE:
            apply_handle_type(s);
            break;
 
        case LOGICAL_REP_MSG_ORIGIN:
            apply_handle_origin(s);
            break;
 
        case LOGICAL_REP_MSG_MESSAGE:
 
            /*
             * Logical replication does not use generic logical messages yet.
             * Although, it could be used by other applications that use this
             * output plugin.
             */
            break;
 
        case LOGICAL_REP_MSG_STREAM_START:
            apply_handle_stream_start(s);
            break;
 
        case LOGICAL_REP_MSG_STREAM_STOP:
            apply_handle_stream_stop(s);
            break;
 
        case LOGICAL_REP_MSG_STREAM_ABORT:
            apply_handle_stream_abort(s);
            break;
 
        case LOGICAL_REP_MSG_STREAM_COMMIT:
            apply_handle_stream_commit(s);
            break;
 
        case LOGICAL_REP_MSG_BEGIN_PREPARE:
            apply_handle_begin_prepare(s);
            break;
 
        case LOGICAL_REP_MSG_PREPARE:
            apply_handle_prepare(s);
            break;
 
        case LOGICAL_REP_MSG_COMMIT_PREPARED:
            apply_handle_commit_prepared(s);
            break;
 
        case LOGICAL_REP_MSG_ROLLBACK_PREPARED:
            apply_handle_rollback_prepared(s);
            break;
 
        case LOGICAL_REP_MSG_STREAM_PREPARE:
            apply_handle_stream_prepare(s);
            break;
 
        default:
            ereport(ERROR,
                    (errcode(ERRCODE_PROTOCOL_VIOLATION),
                     errmsg("invalid logical replication message type \"??? (%d)\"", action)));
    }
 
    /* Reset the current command */
    apply_error_callback_arg.command = saved_command;
}

References generate_unaccent_rules::action, apply_error_callback_arg, apply_handle_begin(), apply_handle_begin_prepare(), apply_handle_commit(), apply_handle_commit_prepared(), apply_handle_delete(), apply_handle_insert(), apply_handle_origin(), apply_handle_prepare(), apply_handle_relation(), apply_handle_rollback_prepared(), apply_handle_stream_abort(), apply_handle_stream_commit(), apply_handle_stream_prepare(), apply_handle_stream_start(), apply_handle_stream_stop(), apply_handle_truncate(), apply_handle_type(), apply_handle_update(), ApplyErrorCallbackArg::command, ereport, errcode(), errmsg(), ERROR, LOGICAL_REP_MSG_BEGIN, LOGICAL_REP_MSG_BEGIN_PREPARE, LOGICAL_REP_MSG_COMMIT, LOGICAL_REP_MSG_COMMIT_PREPARED, LOGICAL_REP_MSG_DELETE, LOGICAL_REP_MSG_INSERT, LOGICAL_REP_MSG_MESSAGE, LOGICAL_REP_MSG_ORIGIN, LOGICAL_REP_MSG_PREPARE, LOGICAL_REP_MSG_RELATION, LOGICAL_REP_MSG_ROLLBACK_PREPARED, LOGICAL_REP_MSG_STREAM_ABORT, LOGICAL_REP_MSG_STREAM_COMMIT, LOGICAL_REP_MSG_STREAM_PREPARE, LOGICAL_REP_MSG_STREAM_START, LOGICAL_REP_MSG_STREAM_STOP, LOGICAL_REP_MSG_TRUNCATE, LOGICAL_REP_MSG_TYPE, LOGICAL_REP_MSG_UPDATE, and pq_getmsgbyte().

Referenced by apply_spooled_messages(), LogicalParallelApplyLoop(), and LogicalRepApplyLoop().

◆ apply_error_callback()

void apply_error_callback ( void * arg )

Definition at line 5014 of file worker.c.

{
    ApplyErrorCallbackArg *errarg = &apply_error_callback_arg;
 
    if (apply_error_callback_arg.command == 0)
        return;
 
    Assert(errarg->origin_name);
 
    if (errarg->rel == NULL)
    {
        if (!TransactionIdIsValid(errarg->remote_xid))
            errcontext("processing remote data for replication origin \"%s\" during message type \"%s\"",
                       errarg->origin_name,
                       logicalrep_message_type(errarg->command));
        else if (XLogRecPtrIsInvalid(errarg->finish_lsn))
            errcontext("processing remote data for replication origin \"%s\" during message type \"%s\" in transaction %u",
                       errarg->origin_name,
                       logicalrep_message_type(errarg->command),
                       errarg->remote_xid);
        else
            errcontext("processing remote data for replication origin \"%s\" during message type \"%s\" in transaction %u, finished at %X/%X",
                       errarg->origin_name,
                       logicalrep_message_type(errarg->command),
                       errarg->remote_xid,
                       LSN_FORMAT_ARGS(errarg->finish_lsn));
    }
    else
    {
        if (errarg->remote_attnum < 0)
        {
            if (XLogRecPtrIsInvalid(errarg->finish_lsn))
                errcontext("processing remote data for replication origin \"%s\" during message type \"%s\" for replication target relation \"%s.%s\" in transaction %u",
                           errarg->origin_name,
                           logicalrep_message_type(errarg->command),
                           errarg->rel->remoterel.nspname,
                           errarg->rel->remoterel.relname,
                           errarg->remote_xid);
            else
                errcontext("processing remote data for replication origin \"%s\" during message type \"%s\" for replication target relation \"%s.%s\" in transaction %u, finished at %X/%X",
                           errarg->origin_name,
                           logicalrep_message_type(errarg->command),
                           errarg->rel->remoterel.nspname,
                           errarg->rel->remoterel.relname,
                           errarg->remote_xid,
                           LSN_FORMAT_ARGS(errarg->finish_lsn));
        }
        else
        {
            if (XLogRecPtrIsInvalid(errarg->finish_lsn))
                errcontext("processing remote data for replication origin \"%s\" during message type \"%s\" for replication target relation \"%s.%s\" column \"%s\" in transaction %u",
                           errarg->origin_name,
                           logicalrep_message_type(errarg->command),
                           errarg->rel->remoterel.nspname,
                           errarg->rel->remoterel.relname,
                           errarg->rel->remoterel.attnames[errarg->remote_attnum],
                           errarg->remote_xid);
            else
                errcontext("processing remote data for replication origin \"%s\" during message type \"%s\" for replication target relation \"%s.%s\" column \"%s\" in transaction %u, finished at %X/%X",
                           errarg->origin_name,
                           logicalrep_message_type(errarg->command),
                           errarg->rel->remoterel.nspname,
                           errarg->rel->remoterel.relname,
                           errarg->rel->remoterel.attnames[errarg->remote_attnum],
                           errarg->remote_xid,
                           LSN_FORMAT_ARGS(errarg->finish_lsn));
        }
    }
}

References apply_error_callback_arg, Assert(), LogicalRepRelation::attnames, ApplyErrorCallbackArg::command, errcontext, ApplyErrorCallbackArg::finish_lsn, logicalrep_message_type(), LSN_FORMAT_ARGS, LogicalRepRelation::nspname, ApplyErrorCallbackArg::origin_name, ApplyErrorCallbackArg::rel, LogicalRepRelation::relname, ApplyErrorCallbackArg::remote_attnum, ApplyErrorCallbackArg::remote_xid, LogicalRepRelMapEntry::remoterel, TransactionIdIsValid, and XLogRecPtrIsInvalid.

Referenced by LogicalParallelApplyLoop(), and LogicalRepApplyLoop().

◆ apply_spooled_messages()

void apply_spooled_messages	(	FileSet *	stream_fileset,
		TransactionId	xid,
		XLogRecPtr	lsn
	)

Definition at line 2025 of file worker.c.

{
    int         nchanges;
    char        path[MAXPGPATH];
    char       *buffer = NULL;
    MemoryContext oldcxt;
    ResourceOwner oldowner;
    int         fileno;
    off_t       offset;
 
    if (!am_parallel_apply_worker())
        maybe_start_skipping_changes(lsn);
 
    /* Make sure we have an open transaction */
    begin_replication_step();
 
    /*
     * Allocate file handle and memory required to process all the messages in
     * TopTransactionContext to avoid them getting reset after each message is
     * processed.
     */
    oldcxt = MemoryContextSwitchTo(TopTransactionContext);
 
    /* Open the spool file for the committed/prepared transaction */
    changes_filename(path, MyLogicalRepWorker->subid, xid);
    elog(DEBUG1, "replaying changes from file \"%s\"", path);
 
    /*
     * Make sure the file is owned by the toplevel transaction so that the
     * file will not be accidentally closed when aborting a subtransaction.
     */
    oldowner = CurrentResourceOwner;
    CurrentResourceOwner = TopTransactionResourceOwner;
 
    stream_fd = BufFileOpenFileSet(stream_fileset, path, O_RDONLY, false);
 
    CurrentResourceOwner = oldowner;
 
    buffer = palloc(BLCKSZ);
 
    MemoryContextSwitchTo(oldcxt);
 
    remote_final_lsn = lsn;
 
    /*
     * Make sure the handle apply_dispatch methods are aware we're in a remote
     * transaction.
     */
    in_remote_transaction = true;
    pgstat_report_activity(STATE_RUNNING, NULL);
 
    end_replication_step();
 
    /*
     * Read the entries one by one and pass them through the same logic as in
     * apply_dispatch.
     */
    nchanges = 0;
    while (true)
    {
        StringInfoData s2;
        size_t      nbytes;
        int         len;
 
        CHECK_FOR_INTERRUPTS();
 
        /* read length of the on-disk record */
        nbytes = BufFileReadMaybeEOF(stream_fd, &len, sizeof(len), true);
 
        /* have we reached end of the file? */
        if (nbytes == 0)
            break;
 
        /* do we have a correct length? */
        if (len <= 0)
            elog(ERROR, "incorrect length %d in streaming transaction's changes file \"%s\"",
                 len, path);
 
        /* make sure we have sufficiently large buffer */
        buffer = repalloc(buffer, len);
 
        /* and finally read the data into the buffer */
        BufFileReadExact(stream_fd, buffer, len);
 
        BufFileTell(stream_fd, &fileno, &offset);
 
        /* init a stringinfo using the buffer and call apply_dispatch */
        initReadOnlyStringInfo(&s2, buffer, len);
 
        /* Ensure we are reading the data into our memory context. */
        oldcxt = MemoryContextSwitchTo(ApplyMessageContext);
 
        apply_dispatch(&s2);
 
        MemoryContextReset(ApplyMessageContext);
 
        MemoryContextSwitchTo(oldcxt);
 
        nchanges++;
 
        /*
         * It is possible the file has been closed because we have processed
         * the transaction end message like stream_commit in which case that
         * must be the last message.
         */
        if (!stream_fd)
        {
            ensure_last_message(stream_fileset, xid, fileno, offset);
            break;
        }
 
        if (nchanges % 1000 == 0)
            elog(DEBUG1, "replayed %d changes from file \"%s\"",
                 nchanges, path);
    }
 
    if (stream_fd)
        stream_close_file();
 
    elog(DEBUG1, "replayed %d (all) changes from file \"%s\"",
         nchanges, path);
 
    return;
}

References am_parallel_apply_worker(), apply_dispatch(), ApplyMessageContext, begin_replication_step(), BufFileOpenFileSet(), BufFileReadExact(), BufFileReadMaybeEOF(), BufFileTell(), changes_filename(), CHECK_FOR_INTERRUPTS, CurrentResourceOwner, DEBUG1, elog, end_replication_step(), ensure_last_message(), ERROR, in_remote_transaction, initReadOnlyStringInfo(), len, MAXPGPATH, maybe_start_skipping_changes(), MemoryContextReset(), MemoryContextSwitchTo(), MyLogicalRepWorker, palloc(), pgstat_report_activity(), remote_final_lsn, repalloc(), s2, STATE_RUNNING, stream_close_file(), stream_fd, LogicalRepWorker::subid, TopTransactionContext, and TopTransactionResourceOwner.

Referenced by apply_handle_stream_commit(), apply_handle_stream_prepare(), and pa_process_spooled_messages_if_required().

◆ DisableSubscriptionAndExit()

void DisableSubscriptionAndExit ( void )

Definition at line 4826 of file worker.c.

{
    /*
     * Emit the error message, and recover from the error state to an idle
     * state
     */
    HOLD_INTERRUPTS();
 
    EmitErrorReport();
    AbortOutOfAnyTransaction();
    FlushErrorState();
 
    RESUME_INTERRUPTS();
 
    /* Report the worker failed during either table synchronization or apply */
    pgstat_report_subscription_error(MyLogicalRepWorker->subid,
                                     !am_tablesync_worker());
 
    /* Disable the subscription */
    StartTransactionCommand();
    DisableSubscription(MySubscription->oid);
    CommitTransactionCommand();
 
    /* Ensure we remove no-longer-useful entry for worker's start time */
    if (am_leader_apply_worker())
        ApplyLauncherForgetWorkerStartTime(MyLogicalRepWorker->subid);
 
    /* Notify the subscription has been disabled and exit */
    ereport(LOG,
            errmsg("subscription \"%s\" has been disabled because of an error",
                   MySubscription->name));
 
    proc_exit(0);
}

References AbortOutOfAnyTransaction(), am_leader_apply_worker(), am_tablesync_worker(), ApplyLauncherForgetWorkerStartTime(), CommitTransactionCommand(), DisableSubscription(), EmitErrorReport(), ereport, errmsg(), FlushErrorState(), HOLD_INTERRUPTS, LOG, MyLogicalRepWorker, MySubscription, Subscription::name, Subscription::oid, pgstat_report_subscription_error(), proc_exit(), RESUME_INTERRUPTS, StartTransactionCommand(), and LogicalRepWorker::subid.

Referenced by start_apply(), and start_table_sync().

◆ InitializeLogRepWorker()

void InitializeLogRepWorker ( void )

Definition at line 4658 of file worker.c.

{
    MemoryContext oldctx;
 
    /* 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 worker for subscription %u will not start because the subscription was removed during startup",
                        MyLogicalRepWorker->subid)));
 
        /* Ensure we remove no-longer-useful entry for worker's start time */
        if (am_leader_apply_worker())
            ApplyLauncherForgetWorkerStartTime(MyLogicalRepWorker->subid);
 
        proc_exit(0);
    }
 
    MySubscriptionValid = true;
    MemoryContextSwitchTo(oldctx);
 
    if (!MySubscription->enabled)
    {
        ereport(LOG,
                (errmsg("logical replication worker for subscription \"%s\" will not start because the subscription was disabled during startup",
                        MySubscription->name)));
 
        apply_worker_exit();
    }
 
    /* Setup synchronous commit according to the user's wishes */
    SetConfigOption("synchronous_commit", MySubscription->synccommit,
                    PGC_BACKEND, PGC_S_OVERRIDE);
 
    /*
     * Keep us informed about subscription or role changes. Note that the
     * role's superuser privilege can be revoked.
     */
    CacheRegisterSyscacheCallback(SUBSCRIPTIONOID,
                                  subscription_change_cb,
                                  (Datum) 0);
 
    CacheRegisterSyscacheCallback(AUTHOID,
                                  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();
}

References ALLOCSET_DEFAULT_SIZES, AllocSetContextCreate, am_leader_apply_worker(), am_tablesync_worker(), apply_worker_exit(), ApplyContext, ApplyLauncherForgetWorkerStartTime(), BackgroundWorkerInitializeConnectionByOid(), CacheRegisterSyscacheCallback(), CommitTransactionCommand(), LogicalRepWorker::dbid, Subscription::enabled, ereport, errmsg(), get_rel_name(), GetSubscription(), LOG, MemoryContextSwitchTo(), MyLogicalRepWorker, MySubscription, MySubscriptionValid, Subscription::name, PGC_BACKEND, PGC_S_OVERRIDE, PGC_SUSET, proc_exit(), LogicalRepWorker::relid, SetConfigOption(), StartTransactionCommand(), LogicalRepWorker::subid, subscription_change_cb(), Subscription::synccommit, TopMemoryContext, and LogicalRepWorker::userid.

Referenced by ParallelApplyWorkerMain(), and SetupApplyOrSyncWorker().

◆ invalidate_syncing_table_states()

void invalidate_syncing_table_states	(	Datum	arg,
		int	cacheid,
		uint32	hashvalue
	)

Definition at line 280 of file tablesync.c.

{
    table_states_validity = SYNC_TABLE_STATE_NEEDS_REBUILD;
}

References SYNC_TABLE_STATE_NEEDS_REBUILD, and table_states_validity.

Referenced by ParallelApplyWorkerMain(), and SetupApplyOrSyncWorker().

◆ logicalrep_pa_worker_stop()

void logicalrep_pa_worker_stop ( ParallelApplyWorkerInfo * winfo )

Definition at line 630 of file launcher.c.

{
    int         slot_no;
    uint16      generation;
    LogicalRepWorker *worker;
 
    SpinLockAcquire(&winfo->shared->mutex);
    generation = winfo->shared->logicalrep_worker_generation;
    slot_no = winfo->shared->logicalrep_worker_slot_no;
    SpinLockRelease(&winfo->shared->mutex);
 
    Assert(slot_no >= 0 && slot_no < max_logical_replication_workers);
 
    /*
     * Detach from the error_mq_handle for the parallel apply worker before
     * stopping it. This prevents the leader apply worker from trying to
     * receive the message from the error queue that might already be detached
     * by the parallel apply worker.
     */
    if (winfo->error_mq_handle)
    {
        shm_mq_detach(winfo->error_mq_handle);
        winfo->error_mq_handle = NULL;
    }
 
    LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
 
    worker = &LogicalRepCtx->workers[slot_no];
    Assert(isParallelApplyWorker(worker));
 
    /*
     * Only stop the worker if the generation matches and the worker is alive.
     */
    if (worker->generation == generation && worker->proc)
        logicalrep_worker_stop_internal(worker, SIGINT);
 
    LWLockRelease(LogicalRepWorkerLock);
}

References Assert(), ParallelApplyWorkerInfo::error_mq_handle, LogicalRepWorker::generation, isParallelApplyWorker, ParallelApplyWorkerShared::logicalrep_worker_generation, ParallelApplyWorkerShared::logicalrep_worker_slot_no, logicalrep_worker_stop_internal(), LogicalRepCtx, LW_SHARED, LWLockAcquire(), LWLockRelease(), max_logical_replication_workers, ParallelApplyWorkerShared::mutex, LogicalRepWorker::proc, ParallelApplyWorkerInfo::shared, shm_mq_detach(), SpinLockAcquire, SpinLockRelease, and LogicalRepCtxStruct::workers.

Referenced by pa_free_worker().

◆ logicalrep_sync_worker_count()

int logicalrep_sync_worker_count ( Oid subid )

Definition at line 845 of file launcher.c.

{
    int         i;
    int         res = 0;
 
    Assert(LWLockHeldByMe(LogicalRepWorkerLock));
 
    /* Search for attached worker for a given subscription id. */
    for (i = 0; i < max_logical_replication_workers; i++)
    {
        LogicalRepWorker *w = &LogicalRepCtx->workers[i];
 
        if (isTablesyncWorker(w) && w->subid == subid)
            res++;
    }
 
    return res;
}

References Assert(), i, isTablesyncWorker, LogicalRepCtx, LWLockHeldByMe(), max_logical_replication_workers, LogicalRepWorker::subid, and LogicalRepCtxStruct::workers.

Referenced by logicalrep_worker_launch(), and process_syncing_tables_for_apply().

◆ logicalrep_worker_attach()

void logicalrep_worker_attach ( int slot )

Definition at line 704 of file launcher.c.

{
    /* Block concurrent access. */
    LWLockAcquire(LogicalRepWorkerLock, LW_EXCLUSIVE);
 
    Assert(slot >= 0 && slot < max_logical_replication_workers);
    MyLogicalRepWorker = &LogicalRepCtx->workers[slot];
 
    if (!MyLogicalRepWorker->in_use)
    {
        LWLockRelease(LogicalRepWorkerLock);
        ereport(ERROR,
                (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                 errmsg("logical replication worker slot %d is empty, cannot attach",
                        slot)));
    }
 
    if (MyLogicalRepWorker->proc)
    {
        LWLockRelease(LogicalRepWorkerLock);
        ereport(ERROR,
                (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                 errmsg("logical replication worker slot %d is already used by "
                        "another worker, cannot attach", slot)));
    }
 
    MyLogicalRepWorker->proc = MyProc;
    before_shmem_exit(logicalrep_worker_onexit, (Datum) 0);
 
    LWLockRelease(LogicalRepWorkerLock);
}

References Assert(), before_shmem_exit(), ereport, errcode(), errmsg(), ERROR, LogicalRepWorker::in_use, logicalrep_worker_onexit(), LogicalRepCtx, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), max_logical_replication_workers, MyLogicalRepWorker, MyProc, LogicalRepWorker::proc, and LogicalRepCtxStruct::workers.

Referenced by ParallelApplyWorkerMain(), and SetupApplyOrSyncWorker().

◆ logicalrep_worker_find()

LogicalRepWorker * logicalrep_worker_find	(	Oid	subid,
		Oid	relid,
		bool	only_running
	)

Definition at line 234 of file launcher.c.

{
    int         i;
    LogicalRepWorker *res = NULL;
 
    Assert(LWLockHeldByMe(LogicalRepWorkerLock));
 
    /* Search for attached worker for a given subscription id. */
    for (i = 0; i < max_logical_replication_workers; i++)
    {
        LogicalRepWorker *w = &LogicalRepCtx->workers[i];
 
        /* Skip parallel apply workers. */
        if (isParallelApplyWorker(w))
            continue;
 
        if (w->in_use && w->subid == subid && w->relid == relid &&
            (!only_running || w->proc))
        {
            res = w;
            break;
        }
    }
 
    return res;
}

References Assert(), i, LogicalRepWorker::in_use, isParallelApplyWorker, LogicalRepCtx, LWLockHeldByMe(), max_logical_replication_workers, LogicalRepWorker::proc, LogicalRepWorker::relid, LogicalRepWorker::subid, and LogicalRepCtxStruct::workers.

Referenced by ApplyLauncherMain(), logicalrep_worker_stop(), logicalrep_worker_wakeup(), process_syncing_tables_for_apply(), wait_for_relation_state_change(), and wait_for_worker_state_change().

◆ logicalrep_worker_launch()

bool logicalrep_worker_launch	(	LogicalRepWorkerType	wtype,
		Oid	dbid,
		Oid	subid,
		const char *	subname,
		Oid	userid,
		Oid	relid,
		dsm_handle	subworker_dsm
	)

Definition at line 297 of file launcher.c.

{
    BackgroundWorker bgw;
    BackgroundWorkerHandle *bgw_handle;
    uint16      generation;
    int         i;
    int         slot = 0;
    LogicalRepWorker *worker = NULL;
    int         nsyncworkers;
    int         nparallelapplyworkers;
    TimestampTz now;
    bool        is_tablesync_worker = (wtype == WORKERTYPE_TABLESYNC);
    bool        is_parallel_apply_worker = (wtype == WORKERTYPE_PARALLEL_APPLY);
 
    /*----------
     * Sanity checks:
     * - must be valid worker type
     * - tablesync workers are only ones to have relid
     * - parallel apply worker is the only kind of subworker
     */
    Assert(wtype != WORKERTYPE_UNKNOWN);
    Assert(is_tablesync_worker == OidIsValid(relid));
    Assert(is_parallel_apply_worker == (subworker_dsm != DSM_HANDLE_INVALID));
 
    ereport(DEBUG1,
            (errmsg_internal("starting logical replication worker for subscription \"%s\"",
                             subname)));
 
    /* Report this after the initial starting message for consistency. */
    if (max_active_replication_origins == 0)
        ereport(ERROR,
                (errcode(ERRCODE_CONFIGURATION_LIMIT_EXCEEDED),
                 errmsg("cannot start logical replication workers when \"max_active_replication_origins\"=0")));
 
    /*
     * We need to do the modification of the shared memory under lock so that
     * we have consistent view.
     */
    LWLockAcquire(LogicalRepWorkerLock, LW_EXCLUSIVE);
 
retry:
    /* Find unused worker slot. */
    for (i = 0; i < max_logical_replication_workers; i++)
    {
        LogicalRepWorker *w = &LogicalRepCtx->workers[i];
 
        if (!w->in_use)
        {
            worker = w;
            slot = i;
            break;
        }
    }
 
    nsyncworkers = logicalrep_sync_worker_count(subid);
 
    now = GetCurrentTimestamp();
 
    /*
     * If we didn't find a free slot, try to do garbage collection.  The
     * reason we do this is because if some worker failed to start up and its
     * parent has crashed while waiting, the in_use state was never cleared.
     */
    if (worker == NULL || nsyncworkers >= max_sync_workers_per_subscription)
    {
        bool        did_cleanup = false;
 
        for (i = 0; i < max_logical_replication_workers; i++)
        {
            LogicalRepWorker *w = &LogicalRepCtx->workers[i];
 
            /*
             * If the worker was marked in use but didn't manage to attach in
             * time, clean it up.
             */
            if (w->in_use && !w->proc &&
                TimestampDifferenceExceeds(w->launch_time, now,
                                           wal_receiver_timeout))
            {
                elog(WARNING,
                     "logical replication worker for subscription %u took too long to start; canceled",
                     w->subid);
 
                logicalrep_worker_cleanup(w);
                did_cleanup = true;
            }
        }
 
        if (did_cleanup)
            goto retry;
    }
 
    /*
     * We don't allow to invoke more sync workers once we have reached the
     * sync worker limit per subscription. So, just return silently as we
     * might get here because of an otherwise harmless race condition.
     */
    if (is_tablesync_worker && nsyncworkers >= max_sync_workers_per_subscription)
    {
        LWLockRelease(LogicalRepWorkerLock);
        return false;
    }
 
    nparallelapplyworkers = logicalrep_pa_worker_count(subid);
 
    /*
     * Return false if the number of parallel apply workers reached the limit
     * per subscription.
     */
    if (is_parallel_apply_worker &&
        nparallelapplyworkers >= max_parallel_apply_workers_per_subscription)
    {
        LWLockRelease(LogicalRepWorkerLock);
        return false;
    }
 
    /*
     * However if there are no more free worker slots, inform user about it
     * before exiting.
     */
    if (worker == NULL)
    {
        LWLockRelease(LogicalRepWorkerLock);
        ereport(WARNING,
                (errcode(ERRCODE_CONFIGURATION_LIMIT_EXCEEDED),
                 errmsg("out of logical replication worker slots"),
                 errhint("You might need to increase \"%s\".", "max_logical_replication_workers")));
        return false;
    }
 
    /* Prepare the worker slot. */
    worker->type = wtype;
    worker->launch_time = now;
    worker->in_use = true;
    worker->generation++;
    worker->proc = NULL;
    worker->dbid = dbid;
    worker->userid = userid;
    worker->subid = subid;
    worker->relid = relid;
    worker->relstate = SUBREL_STATE_UNKNOWN;
    worker->relstate_lsn = InvalidXLogRecPtr;
    worker->stream_fileset = NULL;
    worker->leader_pid = is_parallel_apply_worker ? MyProcPid : InvalidPid;
    worker->parallel_apply = is_parallel_apply_worker;
    worker->last_lsn = InvalidXLogRecPtr;
    TIMESTAMP_NOBEGIN(worker->last_send_time);
    TIMESTAMP_NOBEGIN(worker->last_recv_time);
    worker->reply_lsn = InvalidXLogRecPtr;
    TIMESTAMP_NOBEGIN(worker->reply_time);
 
    /* Before releasing lock, remember generation for future identification. */
    generation = worker->generation;
 
    LWLockRelease(LogicalRepWorkerLock);
 
    /* Register the new dynamic worker. */
    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");
 
    switch (worker->type)
    {
        case WORKERTYPE_APPLY:
            snprintf(bgw.bgw_function_name, BGW_MAXLEN, "ApplyWorkerMain");
            snprintf(bgw.bgw_name, BGW_MAXLEN,
                     "logical replication apply worker for subscription %u",
                     subid);
            snprintf(bgw.bgw_type, BGW_MAXLEN, "logical replication apply worker");
            break;
 
        case WORKERTYPE_PARALLEL_APPLY:
            snprintf(bgw.bgw_function_name, BGW_MAXLEN, "ParallelApplyWorkerMain");
            snprintf(bgw.bgw_name, BGW_MAXLEN,
                     "logical replication parallel apply worker for subscription %u",
                     subid);
            snprintf(bgw.bgw_type, BGW_MAXLEN, "logical replication parallel worker");
 
            memcpy(bgw.bgw_extra, &subworker_dsm, sizeof(dsm_handle));
            break;
 
        case WORKERTYPE_TABLESYNC:
            snprintf(bgw.bgw_function_name, BGW_MAXLEN, "TablesyncWorkerMain");
            snprintf(bgw.bgw_name, BGW_MAXLEN,
                     "logical replication tablesync worker for subscription %u sync %u",
                     subid,
                     relid);
            snprintf(bgw.bgw_type, BGW_MAXLEN, "logical replication tablesync worker");
            break;
 
        case WORKERTYPE_UNKNOWN:
            /* Should never happen. */
            elog(ERROR, "unknown worker type");
    }
 
    bgw.bgw_restart_time = BGW_NEVER_RESTART;
    bgw.bgw_notify_pid = MyProcPid;
    bgw.bgw_main_arg = Int32GetDatum(slot);
 
    if (!RegisterDynamicBackgroundWorker(&bgw, &bgw_handle))
    {
        /* Failed to start worker, so clean up the worker slot. */
        LWLockAcquire(LogicalRepWorkerLock, LW_EXCLUSIVE);
        Assert(generation == worker->generation);
        logicalrep_worker_cleanup(worker);
        LWLockRelease(LogicalRepWorkerLock);
 
        ereport(WARNING,
                (errcode(ERRCODE_CONFIGURATION_LIMIT_EXCEEDED),
                 errmsg("out of background worker slots"),
                 errhint("You might need to increase \"%s\".", "max_worker_processes")));
        return false;
    }
 
    /* Now wait until it attaches. */
    return WaitForReplicationWorkerAttach(worker, generation, bgw_handle);
}

Referenced by ApplyLauncherMain(), pa_launch_parallel_worker(), and process_syncing_tables_for_apply().

◆ logicalrep_worker_stop()

void logicalrep_worker_stop	(	Oid	subid,
		Oid	relid
	)

Definition at line 606 of file launcher.c.

{
    LogicalRepWorker *worker;
 
    LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
 
    worker = logicalrep_worker_find(subid, relid, false);
 
    if (worker)
    {
        Assert(!isParallelApplyWorker(worker));
        logicalrep_worker_stop_internal(worker, SIGTERM);
    }
 
    LWLockRelease(LogicalRepWorkerLock);
}

References Assert(), isParallelApplyWorker, logicalrep_worker_find(), logicalrep_worker_stop_internal(), LW_SHARED, LWLockAcquire(), and LWLockRelease().

Referenced by AlterSubscription_refresh(), and DropSubscription().

◆ logicalrep_worker_wakeup()

void logicalrep_worker_wakeup	(	Oid	subid,
		Oid	relid
	)

Definition at line 673 of file launcher.c.

{
    LogicalRepWorker *worker;
 
    LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
 
    worker = logicalrep_worker_find(subid, relid, true);
 
    if (worker)
        logicalrep_worker_wakeup_ptr(worker);
 
    LWLockRelease(LogicalRepWorkerLock);
}

References logicalrep_worker_find(), logicalrep_worker_wakeup_ptr(), LW_SHARED, LWLockAcquire(), and LWLockRelease().

Referenced by apply_handle_stream_start(), and finish_sync_worker().

◆ logicalrep_worker_wakeup_ptr()

void logicalrep_worker_wakeup_ptr ( LogicalRepWorker * worker )

Definition at line 693 of file launcher.c.

{
    Assert(LWLockHeldByMe(LogicalRepWorkerLock));
 
    SetLatch(&worker->proc->procLatch);
}

References Assert(), LWLockHeldByMe(), LogicalRepWorker::proc, PGPROC::procLatch, and SetLatch().

Referenced by AtEOXact_LogicalRepWorkers(), logicalrep_worker_wakeup(), process_syncing_tables_for_apply(), and wait_for_worker_state_change().

◆ logicalrep_workers_find()

List * logicalrep_workers_find	(	Oid	subid,
		bool	only_running,
		bool	acquire_lock
	)

Definition at line 266 of file launcher.c.

{
    int         i;
    List       *res = NIL;
 
    if (acquire_lock)
        LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
 
    Assert(LWLockHeldByMe(LogicalRepWorkerLock));
 
    /* Search for attached worker for a given subscription id. */
    for (i = 0; i < max_logical_replication_workers; i++)
    {
        LogicalRepWorker *w = &LogicalRepCtx->workers[i];
 
        if (w->in_use && w->subid == subid && (!only_running || w->proc))
            res = lappend(res, w);
    }
 
    if (acquire_lock)
        LWLockRelease(LogicalRepWorkerLock);
 
    return res;
}

References Assert(), i, LogicalRepWorker::in_use, lappend(), LogicalRepCtx, LW_SHARED, LWLockAcquire(), LWLockHeldByMe(), LWLockRelease(), max_logical_replication_workers, NIL, LogicalRepWorker::proc, LogicalRepWorker::subid, and LogicalRepCtxStruct::workers.

Referenced by AlterSubscription(), AtEOXact_LogicalRepWorkers(), DropSubscription(), and logicalrep_worker_detach().

◆ maybe_reread_subscription()

void maybe_reread_subscription ( void )

Definition at line 3968 of file worker.c.

{
    MemoryContext oldctx;
    Subscription *newsub;
    bool        started_tx = false;
 
    /* When cache state is valid there is nothing to do here. */
    if (MySubscriptionValid)
        return;
 
    /* This function might be called inside or outside of transaction. */
    if (!IsTransactionState())
    {
        StartTransactionCommand();
        started_tx = true;
    }
 
    /* Ensure allocations in permanent context. */
    oldctx = MemoryContextSwitchTo(ApplyContext);
 
    newsub = GetSubscription(MyLogicalRepWorker->subid, true);
 
    /*
     * Exit if the subscription was removed. This normally should not happen
     * as the worker gets killed during DROP SUBSCRIPTION.
     */
    if (!newsub)
    {
        ereport(LOG,
                (errmsg("logical replication worker for subscription \"%s\" will stop because the subscription was removed",
                        MySubscription->name)));
 
        /* Ensure we remove no-longer-useful entry for worker's start time */
        if (am_leader_apply_worker())
            ApplyLauncherForgetWorkerStartTime(MyLogicalRepWorker->subid);
 
        proc_exit(0);
    }
 
    /* Exit if the subscription was disabled. */
    if (!newsub->enabled)
    {
        ereport(LOG,
                (errmsg("logical replication worker for subscription \"%s\" will stop because the subscription was disabled",
                        MySubscription->name)));
 
        apply_worker_exit();
    }
 
    /* !slotname should never happen when enabled is true. */
    Assert(newsub->slotname);
 
    /* two-phase cannot be altered while the worker is running */
    Assert(newsub->twophasestate == MySubscription->twophasestate);
 
    /*
     * Exit if any parameter that affects the remote connection was changed.
     * The launcher will start a new worker but note that the parallel apply
     * worker won't restart if the streaming option's value is changed from
     * 'parallel' to any other value or the server decides not to stream the
     * in-progress transaction.
     */
    if (strcmp(newsub->conninfo, MySubscription->conninfo) != 0 ||
        strcmp(newsub->name, MySubscription->name) != 0 ||
        strcmp(newsub->slotname, MySubscription->slotname) != 0 ||
        newsub->binary != MySubscription->binary ||
        newsub->stream != MySubscription->stream ||
        newsub->passwordrequired != MySubscription->passwordrequired ||
        strcmp(newsub->origin, MySubscription->origin) != 0 ||
        newsub->owner != MySubscription->owner ||
        !equal(newsub->publications, MySubscription->publications))
    {
        if (am_parallel_apply_worker())
            ereport(LOG,
                    (errmsg("logical replication parallel apply worker for subscription \"%s\" will stop because of a parameter change",
                            MySubscription->name)));
        else
            ereport(LOG,
                    (errmsg("logical replication worker for subscription \"%s\" will restart because of a parameter change",
                            MySubscription->name)));
 
        apply_worker_exit();
    }
 
    /*
     * Exit if the subscription owner's superuser privileges have been
     * revoked.
     */
    if (!newsub->ownersuperuser && MySubscription->ownersuperuser)
    {
        if (am_parallel_apply_worker())
            ereport(LOG,
                    errmsg("logical replication parallel apply worker for subscription \"%s\" will stop because the subscription owner's superuser privileges have been revoked",
                           MySubscription->name));
        else
            ereport(LOG,
                    errmsg("logical replication worker for subscription \"%s\" will restart because the subscription owner's superuser privileges have been revoked",
                           MySubscription->name));
 
        apply_worker_exit();
    }
 
    /* Check for other changes that should never happen too. */
    if (newsub->dbid != MySubscription->dbid)
    {
        elog(ERROR, "subscription %u changed unexpectedly",
             MyLogicalRepWorker->subid);
    }
 
    /* Clean old subscription info and switch to new one. */
    FreeSubscription(MySubscription);
    MySubscription = newsub;
 
    MemoryContextSwitchTo(oldctx);
 
    /* Change synchronous commit according to the user's wishes */
    SetConfigOption("synchronous_commit", MySubscription->synccommit,
                    PGC_BACKEND, PGC_S_OVERRIDE);
 
    if (started_tx)
        CommitTransactionCommand();
 
    MySubscriptionValid = true;
}

References am_leader_apply_worker(), am_parallel_apply_worker(), apply_worker_exit(), ApplyContext, ApplyLauncherForgetWorkerStartTime(), Assert(), Subscription::binary, CommitTransactionCommand(), Subscription::conninfo, Subscription::dbid, elog, equal(), ereport, errmsg(), ERROR, FreeSubscription(), GetSubscription(), IsTransactionState(), LOG, MemoryContextSwitchTo(), MyLogicalRepWorker, MySubscription, MySubscriptionValid, Subscription::name, newsub(), Subscription::origin, Subscription::owner, Subscription::ownersuperuser, Subscription::passwordrequired, PGC_BACKEND, PGC_S_OVERRIDE, proc_exit(), Subscription::publications, SetConfigOption(), Subscription::slotname, StartTransactionCommand(), Subscription::stream, LogicalRepWorker::subid, Subscription::synccommit, and Subscription::twophasestate.

Referenced by apply_handle_commit_internal(), begin_replication_step(), LogicalRepApplyLoop(), and pa_can_start().

◆ pa_allocate_worker()

void pa_allocate_worker ( TransactionId xid )

Definition at line 470 of file applyparallelworker.c.

{
    bool        found;
    ParallelApplyWorkerInfo *winfo = NULL;
    ParallelApplyWorkerEntry *entry;
 
    if (!pa_can_start())
        return;
 
    winfo = pa_launch_parallel_worker();
    if (!winfo)
        return;
 
    /* First time through, initialize parallel apply worker state hashtable. */
    if (!ParallelApplyTxnHash)
    {
        HASHCTL     ctl;
 
        MemSet(&ctl, 0, sizeof(ctl));
        ctl.keysize = sizeof(TransactionId);
        ctl.entrysize = sizeof(ParallelApplyWorkerEntry);
        ctl.hcxt = ApplyContext;
 
        ParallelApplyTxnHash = hash_create("logical replication parallel apply workers hash",
                                           16, &ctl,
                                           HASH_ELEM | HASH_BLOBS | HASH_CONTEXT);
    }
 
    /* Create an entry for the requested transaction. */
    entry = hash_search(ParallelApplyTxnHash, &xid, HASH_ENTER, &found);
    if (found)
        elog(ERROR, "hash table corrupted");
 
    /* Update the transaction information in shared memory. */
    SpinLockAcquire(&winfo->shared->mutex);
    winfo->shared->xact_state = PARALLEL_TRANS_UNKNOWN;
    winfo->shared->xid = xid;
    SpinLockRelease(&winfo->shared->mutex);
 
    winfo->in_use = true;
    winfo->serialize_changes = false;
    entry->winfo = winfo;
}

References ApplyContext, ctl, elog, ERROR, HASH_BLOBS, HASH_CONTEXT, hash_create(), HASH_ELEM, HASH_ENTER, hash_search(), ParallelApplyWorkerInfo::in_use, MemSet, ParallelApplyWorkerShared::mutex, pa_can_start(), pa_launch_parallel_worker(), PARALLEL_TRANS_UNKNOWN, ParallelApplyTxnHash, ParallelApplyWorkerInfo::serialize_changes, ParallelApplyWorkerInfo::shared, SpinLockAcquire, SpinLockRelease, ParallelApplyWorkerEntry::winfo, ParallelApplyWorkerShared::xact_state, and ParallelApplyWorkerShared::xid.

Referenced by apply_handle_stream_start().

◆ pa_decr_and_wait_stream_block()

void pa_decr_and_wait_stream_block ( void )

Definition at line 1591 of file applyparallelworker.c.

{
    Assert(am_parallel_apply_worker());
 
    /*
     * It is only possible to not have any pending stream chunks when we are
     * applying spooled messages.
     */
    if (pg_atomic_read_u32(&MyParallelShared->pending_stream_count) == 0)
    {
        if (pa_has_spooled_message_pending())
            return;
 
        elog(ERROR, "invalid pending streaming chunk 0");
    }
 
    if (pg_atomic_sub_fetch_u32(&MyParallelShared->pending_stream_count, 1) == 0)
    {
        pa_lock_stream(MyParallelShared->xid, AccessShareLock);
        pa_unlock_stream(MyParallelShared->xid, AccessShareLock);
    }
}

References AccessShareLock, am_parallel_apply_worker(), Assert(), elog, ERROR, MyParallelShared, pa_has_spooled_message_pending(), pa_lock_stream(), pa_unlock_stream(), ParallelApplyWorkerShared::pending_stream_count, pg_atomic_read_u32(), pg_atomic_sub_fetch_u32(), and ParallelApplyWorkerShared::xid.

Referenced by apply_handle_stream_abort(), and apply_handle_stream_stop().

◆ pa_detach_all_error_mq()

void pa_detach_all_error_mq ( void )

Definition at line 622 of file applyparallelworker.c.

{
    ListCell   *lc;
 
    foreach(lc, ParallelApplyWorkerPool)
    {
        ParallelApplyWorkerInfo *winfo = (ParallelApplyWorkerInfo *) lfirst(lc);
 
        if (winfo->error_mq_handle)
        {
            shm_mq_detach(winfo->error_mq_handle);
            winfo->error_mq_handle = NULL;
        }
    }
}

References ParallelApplyWorkerInfo::error_mq_handle, lfirst, ParallelApplyWorkerPool, and shm_mq_detach().

Referenced by logicalrep_worker_detach().

◆ pa_find_worker()

ParallelApplyWorkerInfo * pa_find_worker ( TransactionId xid )

Definition at line 518 of file applyparallelworker.c.

{
    bool        found;
    ParallelApplyWorkerEntry *entry;
 
    if (!TransactionIdIsValid(xid))
        return NULL;
 
    if (!ParallelApplyTxnHash)
        return NULL;
 
    /* Return the cached parallel apply worker if valid. */
    if (stream_apply_worker)
        return stream_apply_worker;
 
    /* Find an entry for the requested transaction. */
    entry = hash_search(ParallelApplyTxnHash, &xid, HASH_FIND, &found);
    if (found)
    {
        /* The worker must not have exited.  */
        Assert(entry->winfo->in_use);
        return entry->winfo;
    }
 
    return NULL;
}

References Assert(), HASH_FIND, hash_search(), ParallelApplyWorkerInfo::in_use, ParallelApplyTxnHash, stream_apply_worker, TransactionIdIsValid, and ParallelApplyWorkerEntry::winfo.

Referenced by get_transaction_apply_action().

◆ pa_lock_stream()

void pa_lock_stream	(	TransactionId	xid,
		LOCKMODE	lockmode
	)

Definition at line 1540 of file applyparallelworker.c.

{
    LockApplyTransactionForSession(MyLogicalRepWorker->subid, xid,
                                   PARALLEL_APPLY_LOCK_STREAM, lockmode);
}

References LockApplyTransactionForSession(), MyLogicalRepWorker, PARALLEL_APPLY_LOCK_STREAM, and LogicalRepWorker::subid.

Referenced by apply_handle_stream_abort(), apply_handle_stream_stop(), pa_decr_and_wait_stream_block(), pa_process_spooled_messages_if_required(), and pa_switch_to_partial_serialize().

◆ pa_lock_transaction()

void pa_lock_transaction	(	TransactionId	xid,
		LOCKMODE	lockmode
	)

Definition at line 1573 of file applyparallelworker.c.

{
    LockApplyTransactionForSession(MyLogicalRepWorker->subid, xid,
                                   PARALLEL_APPLY_LOCK_XACT, lockmode);
}

References LockApplyTransactionForSession(), MyLogicalRepWorker, PARALLEL_APPLY_LOCK_XACT, and LogicalRepWorker::subid.

Referenced by apply_handle_stream_start(), and pa_wait_for_xact_finish().

◆ pa_reset_subtrans()

void pa_reset_subtrans ( void )

Definition at line 1402 of file applyparallelworker.c.

{
    /*
     * We don't need to free this explicitly as the allocated memory will be
     * freed at the transaction end.
     */
    subxactlist = NIL;
}

References NIL, and subxactlist.

Referenced by apply_handle_stream_commit(), apply_handle_stream_prepare(), and pa_stream_abort().

◆ pa_send_data()

bool pa_send_data	(	ParallelApplyWorkerInfo *	winfo,
		Size	nbytes,
		const void *	data
	)

Definition at line 1146 of file applyparallelworker.c.

{
    int         rc;
    shm_mq_result result;
    TimestampTz startTime = 0;
 
    Assert(!IsTransactionState());
    Assert(!winfo->serialize_changes);
 
    /*
     * We don't try to send data to parallel worker for 'immediate' mode. This
     * is primarily used for testing purposes.
     */
    if (unlikely(debug_logical_replication_streaming == DEBUG_LOGICAL_REP_STREAMING_IMMEDIATE))
        return false;
 
/*
 * This timeout is a bit arbitrary but testing revealed that it is sufficient
 * to send the message unless the parallel apply worker is waiting on some
 * lock or there is a serious resource crunch. See the comments atop this file
 * to know why we are using a non-blocking way to send the message.
 */
#define SHM_SEND_RETRY_INTERVAL_MS 1000
#define SHM_SEND_TIMEOUT_MS     (10000 - SHM_SEND_RETRY_INTERVAL_MS)
 
    for (;;)
    {
        result = shm_mq_send(winfo->mq_handle, nbytes, data, true, true);
 
        if (result == SHM_MQ_SUCCESS)
            return true;
        else if (result == SHM_MQ_DETACHED)
            ereport(ERROR,
                    (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                     errmsg("could not send data to shared-memory queue")));
 
        Assert(result == SHM_MQ_WOULD_BLOCK);
 
        /* Wait before retrying. */
        rc = WaitLatch(MyLatch,
                       WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
                       SHM_SEND_RETRY_INTERVAL_MS,
                       WAIT_EVENT_LOGICAL_APPLY_SEND_DATA);
 
        if (rc & WL_LATCH_SET)
        {
            ResetLatch(MyLatch);
            CHECK_FOR_INTERRUPTS();
        }
 
        if (startTime == 0)
            startTime = GetCurrentTimestamp();
        else if (TimestampDifferenceExceeds(startTime, GetCurrentTimestamp(),
                                            SHM_SEND_TIMEOUT_MS))
            return false;
    }
}

References Assert(), CHECK_FOR_INTERRUPTS, data, DEBUG_LOGICAL_REP_STREAMING_IMMEDIATE, debug_logical_replication_streaming, ereport, errcode(), errmsg(), ERROR, GetCurrentTimestamp(), IsTransactionState(), ParallelApplyWorkerInfo::mq_handle, MyLatch, ResetLatch(), ParallelApplyWorkerInfo::serialize_changes, SHM_MQ_DETACHED, shm_mq_send(), SHM_MQ_SUCCESS, SHM_MQ_WOULD_BLOCK, SHM_SEND_RETRY_INTERVAL_MS, SHM_SEND_TIMEOUT_MS, TimestampDifferenceExceeds(), unlikely, WaitLatch(), WL_EXIT_ON_PM_DEATH, WL_LATCH_SET, and WL_TIMEOUT.

Referenced by apply_handle_stream_abort(), apply_handle_stream_commit(), apply_handle_stream_prepare(), apply_handle_stream_start(), apply_handle_stream_stop(), and handle_streamed_transaction().

◆ pa_set_fileset_state()

void pa_set_fileset_state	(	ParallelApplyWorkerShared *	wshared,
		PartialFileSetState	fileset_state
	)

Definition at line 1498 of file applyparallelworker.c.

{
    SpinLockAcquire(&wshared->mutex);
    wshared->fileset_state = fileset_state;
 
    if (fileset_state == FS_SERIALIZE_DONE)
    {
        Assert(am_leader_apply_worker());
        Assert(MyLogicalRepWorker->stream_fileset);
        wshared->fileset = *MyLogicalRepWorker->stream_fileset;
    }
 
    SpinLockRelease(&wshared->mutex);
}

References am_leader_apply_worker(), Assert(), ParallelApplyWorkerShared::fileset, ParallelApplyWorkerShared::fileset_state, FS_SERIALIZE_DONE, ParallelApplyWorkerShared::mutex, MyLogicalRepWorker, SpinLockAcquire, SpinLockRelease, and LogicalRepWorker::stream_fileset.

Referenced by apply_handle_stream_abort(), apply_handle_stream_commit(), apply_handle_stream_prepare(), pa_process_spooled_messages_if_required(), and pa_switch_to_partial_serialize().

◆ pa_set_stream_apply_worker()

void pa_set_stream_apply_worker ( ParallelApplyWorkerInfo * winfo )

Definition at line 1334 of file applyparallelworker.c.

{
    stream_apply_worker = winfo;
}

References stream_apply_worker.

Referenced by apply_handle_stream_start(), and apply_handle_stream_stop().

◆ pa_set_xact_state()

void pa_set_xact_state	(	ParallelApplyWorkerShared *	wshared,
		ParallelTransState	xact_state
	)

Definition at line 1307 of file applyparallelworker.c.

{
    SpinLockAcquire(&wshared->mutex);
    wshared->xact_state = xact_state;
    SpinLockRelease(&wshared->mutex);
}

References ParallelApplyWorkerShared::mutex, SpinLockAcquire, SpinLockRelease, and ParallelApplyWorkerShared::xact_state.

Referenced by apply_handle_stream_commit(), apply_handle_stream_prepare(), apply_handle_stream_start(), and pa_stream_abort().

◆ pa_start_subtrans()

void pa_start_subtrans	(	TransactionId	current_xid,
		TransactionId	top_xid
	)

Definition at line 1362 of file applyparallelworker.c.

{
    if (current_xid != top_xid &&
        !list_member_xid(subxactlist, current_xid))
    {
        MemoryContext oldctx;
        char        spname[NAMEDATALEN];
 
        pa_savepoint_name(MySubscription->oid, current_xid,
                          spname, sizeof(spname));
 
        elog(DEBUG1, "defining savepoint %s in logical replication parallel apply worker", spname);
 
        /* We must be in transaction block to define the SAVEPOINT. */
        if (!IsTransactionBlock())
        {
            if (!IsTransactionState())
                StartTransactionCommand();
 
            BeginTransactionBlock();
            CommitTransactionCommand();
        }
 
        DefineSavepoint(spname);
 
        /*
         * CommitTransactionCommand is needed to start a subtransaction after
         * issuing a SAVEPOINT inside a transaction block (see
         * StartSubTransaction()).
         */
        CommitTransactionCommand();
 
        oldctx = MemoryContextSwitchTo(TopTransactionContext);
        subxactlist = lappend_xid(subxactlist, current_xid);
        MemoryContextSwitchTo(oldctx);
    }
}

References BeginTransactionBlock(), CommitTransactionCommand(), DEBUG1, DefineSavepoint(), elog, IsTransactionBlock(), IsTransactionState(), lappend_xid(), list_member_xid(), MemoryContextSwitchTo(), MySubscription, NAMEDATALEN, Subscription::oid, pa_savepoint_name(), StartTransactionCommand(), subxactlist, and TopTransactionContext.

Referenced by handle_streamed_transaction().

◆ pa_stream_abort()

void pa_stream_abort ( LogicalRepStreamAbortData * abort_data )

Definition at line 1416 of file applyparallelworker.c.

{
    TransactionId xid = abort_data->xid;
    TransactionId subxid = abort_data->subxid;
 
    /*
     * Update origin state so we can restart streaming from correct position
     * in case of crash.
     */
    replorigin_session_origin_lsn = abort_data->abort_lsn;
    replorigin_session_origin_timestamp = abort_data->abort_time;
 
    /*
     * If the two XIDs are the same, it's in fact abort of toplevel xact, so
     * just free the subxactlist.
     */
    if (subxid == xid)
    {
        pa_set_xact_state(MyParallelShared, PARALLEL_TRANS_FINISHED);
 
        /*
         * Release the lock as we might be processing an empty streaming
         * transaction in which case the lock won't be released during
         * transaction rollback.
         *
         * Note that it's ok to release the transaction lock before aborting
         * the transaction because even if the parallel apply worker dies due
         * to crash or some other reason, such a transaction would still be
         * considered aborted.
         */
        pa_unlock_transaction(xid, AccessExclusiveLock);
 
        AbortCurrentTransaction();
 
        if (IsTransactionBlock())
        {
            EndTransactionBlock(false);
            CommitTransactionCommand();
        }
 
        pa_reset_subtrans();
 
        pgstat_report_activity(STATE_IDLE, NULL);
    }
    else
    {
        /* OK, so it's a subxact. Rollback to the savepoint. */
        int         i;
        char        spname[NAMEDATALEN];
 
        pa_savepoint_name(MySubscription->oid, subxid, spname, sizeof(spname));
 
        elog(DEBUG1, "rolling back to savepoint %s in logical replication parallel apply worker", spname);
 
        /*
         * Search the subxactlist, determine the offset tracked for the
         * subxact, and truncate the list.
         *
         * Note that for an empty sub-transaction we won't find the subxid
         * here.
         */
        for (i = list_length(subxactlist) - 1; i >= 0; i--)
        {
            TransactionId xid_tmp = lfirst_xid(list_nth_cell(subxactlist, i));
 
            if (xid_tmp == subxid)
            {
                RollbackToSavepoint(spname);
                CommitTransactionCommand();
                subxactlist = list_truncate(subxactlist, i);
                break;
            }
        }
    }
}

References LogicalRepStreamAbortData::abort_lsn, LogicalRepStreamAbortData::abort_time, AbortCurrentTransaction(), AccessExclusiveLock, CommitTransactionCommand(), DEBUG1, elog, EndTransactionBlock(), i, IsTransactionBlock(), lfirst_xid, list_length(), list_nth_cell(), list_truncate(), MyParallelShared, MySubscription, NAMEDATALEN, Subscription::oid, pa_reset_subtrans(), pa_savepoint_name(), pa_set_xact_state(), pa_unlock_transaction(), PARALLEL_TRANS_FINISHED, pgstat_report_activity(), replorigin_session_origin_lsn, replorigin_session_origin_timestamp, RollbackToSavepoint(), STATE_IDLE, subxactlist, LogicalRepStreamAbortData::subxid, and LogicalRepStreamAbortData::xid.

Referenced by apply_handle_stream_abort().

◆ pa_switch_to_partial_serialize()

void pa_switch_to_partial_serialize	(	ParallelApplyWorkerInfo *	winfo,
		bool	stream_locked
	)

Definition at line 1211 of file applyparallelworker.c.

{
    ereport(LOG,
            (errmsg("logical replication apply worker will serialize the remaining changes of remote transaction %u to a file",
                    winfo->shared->xid)));
 
    /*
     * The parallel apply worker could be stuck for some reason (say waiting
     * on some lock by other backend), so stop trying to send data directly to
     * it and start serializing data to the file instead.
     */
    winfo->serialize_changes = true;
 
    /* Initialize the stream fileset. */
    stream_start_internal(winfo->shared->xid, true);
 
    /*
     * Acquires the stream lock if not already to make sure that the parallel
     * apply worker will wait for the leader to release the stream lock until
     * the end of the transaction.
     */
    if (!stream_locked)
        pa_lock_stream(winfo->shared->xid, AccessExclusiveLock);
 
    pa_set_fileset_state(winfo->shared, FS_SERIALIZE_IN_PROGRESS);
}

References AccessExclusiveLock, ereport, errmsg(), FS_SERIALIZE_IN_PROGRESS, LOG, pa_lock_stream(), pa_set_fileset_state(), ParallelApplyWorkerInfo::serialize_changes, ParallelApplyWorkerInfo::shared, stream_start_internal(), and ParallelApplyWorkerShared::xid.

Referenced by apply_handle_stream_abort(), apply_handle_stream_commit(), apply_handle_stream_prepare(), apply_handle_stream_start(), apply_handle_stream_stop(), and handle_streamed_transaction().

◆ pa_unlock_stream()

void pa_unlock_stream	(	TransactionId	xid,
		LOCKMODE	lockmode
	)

Definition at line 1547 of file applyparallelworker.c.

{
    UnlockApplyTransactionForSession(MyLogicalRepWorker->subid, xid,
                                     PARALLEL_APPLY_LOCK_STREAM, lockmode);
}

References MyLogicalRepWorker, PARALLEL_APPLY_LOCK_STREAM, LogicalRepWorker::subid, and UnlockApplyTransactionForSession().

Referenced by apply_handle_stream_abort(), apply_handle_stream_start(), pa_decr_and_wait_stream_block(), pa_process_spooled_messages_if_required(), and pa_xact_finish().

◆ pa_unlock_transaction()

void pa_unlock_transaction	(	TransactionId	xid,
		LOCKMODE	lockmode
	)

Definition at line 1580 of file applyparallelworker.c.

{
    UnlockApplyTransactionForSession(MyLogicalRepWorker->subid, xid,
                                     PARALLEL_APPLY_LOCK_XACT, lockmode);
}

References MyLogicalRepWorker, PARALLEL_APPLY_LOCK_XACT, LogicalRepWorker::subid, and UnlockApplyTransactionForSession().

Referenced by apply_handle_stream_commit(), apply_handle_stream_prepare(), pa_stream_abort(), and pa_wait_for_xact_finish().

◆ pa_xact_finish()

void pa_xact_finish	(	ParallelApplyWorkerInfo *	winfo,
		XLogRecPtr	remote_lsn
	)

Definition at line 1618 of file applyparallelworker.c.

{
    Assert(am_leader_apply_worker());
 
    /*
     * Unlock the shared object lock so that parallel apply worker can
     * continue to receive and apply changes.
     */
    pa_unlock_stream(winfo->shared->xid, AccessExclusiveLock);
 
    /*
     * Wait for that worker to finish. This is necessary to maintain commit
     * order which avoids failures due to transaction dependencies and
     * deadlocks.
     */
    pa_wait_for_xact_finish(winfo);
 
    if (!XLogRecPtrIsInvalid(remote_lsn))
        store_flush_position(remote_lsn, winfo->shared->last_commit_end);
 
    pa_free_worker(winfo);
}

References AccessExclusiveLock, am_leader_apply_worker(), Assert(), ParallelApplyWorkerShared::last_commit_end, pa_free_worker(), pa_unlock_stream(), pa_wait_for_xact_finish(), ParallelApplyWorkerInfo::shared, store_flush_position(), ParallelApplyWorkerShared::xid, and XLogRecPtrIsInvalid.

Referenced by apply_handle_stream_abort(), apply_handle_stream_commit(), and apply_handle_stream_prepare().

◆ process_syncing_tables()

void process_syncing_tables ( XLogRecPtr current_lsn )

Definition at line 666 of file tablesync.c.

{
    switch (MyLogicalRepWorker->type)
    {
        case WORKERTYPE_PARALLEL_APPLY:
 
            /*
             * Skip for parallel apply workers because they only operate on
             * tables that are in a READY state. See pa_can_start() and
             * should_apply_changes_for_rel().
             */
            break;
 
        case WORKERTYPE_TABLESYNC:
            process_syncing_tables_for_sync(current_lsn);
            break;
 
        case WORKERTYPE_APPLY:
            process_syncing_tables_for_apply(current_lsn);
            break;
 
        case WORKERTYPE_UNKNOWN:
            /* Should never happen. */
            elog(ERROR, "Unknown worker type");
    }
}

References elog, ERROR, MyLogicalRepWorker, process_syncing_tables_for_apply(), process_syncing_tables_for_sync(), LogicalRepWorker::type, WORKERTYPE_APPLY, WORKERTYPE_PARALLEL_APPLY, WORKERTYPE_TABLESYNC, and WORKERTYPE_UNKNOWN.

Referenced by apply_handle_commit(), apply_handle_commit_prepared(), apply_handle_prepare(), apply_handle_rollback_prepared(), apply_handle_stream_commit(), apply_handle_stream_prepare(), and LogicalRepApplyLoop().

◆ ReplicationOriginNameForLogicalRep()

void ReplicationOriginNameForLogicalRep	(	Oid	suboid,
		Oid	relid,
		char *	originname,
		Size	szoriginname
	)

Definition at line 428 of file worker.c.

{
    if (OidIsValid(relid))
    {
        /* Replication origin name for tablesync workers. */
        snprintf(originname, szoriginname, "pg_%u_%u", suboid, relid);
    }
    else
    {
        /* Replication origin name for non-tablesync workers. */
        snprintf(originname, szoriginname, "pg_%u", suboid);
    }
}

References OidIsValid, and snprintf.

Referenced by AlterSubscription(), AlterSubscription_refresh(), binary_upgrade_replorigin_advance(), CreateSubscription(), DropSubscription(), LogicalRepSyncTableStart(), ParallelApplyWorkerMain(), process_syncing_tables_for_apply(), process_syncing_tables_for_sync(), run_apply_worker(), and run_tablesync_worker().

◆ set_apply_error_context_origin()

void set_apply_error_context_origin ( char * originname )

Definition at line 5156 of file worker.c.

{
    apply_error_callback_arg.origin_name = MemoryContextStrdup(ApplyContext,
                                                               originname);
}

References apply_error_callback_arg, ApplyContext, MemoryContextStrdup(), and ApplyErrorCallbackArg::origin_name.

Referenced by ParallelApplyWorkerMain(), run_apply_worker(), and run_tablesync_worker().

◆ set_stream_options()

void set_stream_options	(	WalRcvStreamOptions *	options,
		char *	slotname,
		XLogRecPtr *	origin_startpos
	)

Definition at line 4444 of file worker.c.

{
    int         server_version;
 
    options->logical = true;
    options->startpoint = *origin_startpos;
    options->slotname = slotname;
 
    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);
}

References Subscription::binary, LOGICALREP_PROTO_STREAM_PARALLEL_VERSION_NUM, LOGICALREP_PROTO_STREAM_VERSION_NUM, LOGICALREP_PROTO_TWOPHASE_VERSION_NUM, LOGICALREP_PROTO_VERSION_NUM, LogRepWorkerWalRcvConn, MyLogicalRepWorker, MySubscription, Subscription::origin, LogicalRepWorker::parallel_apply, pstrdup(), Subscription::publications, server_version, Subscription::stream, and walrcv_server_version.

Referenced by run_apply_worker(), and run_tablesync_worker().

◆ SetupApplyOrSyncWorker()

void SetupApplyOrSyncWorker ( int worker_slot )

Definition at line 4752 of file worker.c.

{
    /* Attach to slot */
    logicalrep_worker_attach(worker_slot);
 
    Assert(am_tablesync_worker() || am_leader_apply_worker());
 
    /* 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);
 
    InitializeLogRepWorker();
 
    /*
     * Register a callback to reset the origin state before aborting any
     * pending transaction during shutdown (see ShutdownPostgres()). This will
     * avoid origin advancement for an in-complete transaction which could
     * otherwise lead to its loss as such a transaction won't be sent by the
     * server again.
     *
     * Note that even a LOG or DEBUG statement placed after setting the origin
     * state may process a shutdown signal before committing the current apply
     * operation. So, it is important to register such a callback here.
     */
    before_shmem_exit(replorigin_reset, (Datum) 0);
 
    /* Connect to the origin and start the replication. */
    elog(DEBUG1, "connecting to publisher using connection string \"%s\"",
         MySubscription->conninfo);
 
    /*
     * Setup callback for syscache so that we know when something changes in
     * the subscription relation state.
     */
    CacheRegisterSyscacheCallback(SUBSCRIPTIONRELMAP,
                                  invalidate_syncing_table_states,
                                  (Datum) 0);
}

References am_leader_apply_worker(), am_tablesync_worker(), Assert(), BackgroundWorkerUnblockSignals(), before_shmem_exit(), CacheRegisterSyscacheCallback(), Subscription::conninfo, DEBUG1, die, elog, GetCurrentTimestamp(), InitializeLogRepWorker(), invalidate_syncing_table_states(), LogicalRepWorker::last_recv_time, LogicalRepWorker::last_send_time, load_file(), logicalrep_worker_attach(), MyLogicalRepWorker, MySubscription, pqsignal, replorigin_reset(), LogicalRepWorker::reply_time, SIGHUP, and SignalHandlerForConfigReload().

Referenced by ApplyWorkerMain(), and TablesyncWorkerMain().

◆ start_apply()

void start_apply ( XLogRecPtr origin_startpos )

Definition at line 4513 of file worker.c.

{
    PG_TRY();
    {
        LogicalRepApplyLoop(origin_startpos);
    }
    PG_CATCH();
    {
        /*
         * Reset the origin state to prevent the advancement of origin
         * progress if we fail to apply. Otherwise, this will result in
         * transaction loss as that transaction won't be sent again by the
         * server.
         */
        replorigin_reset(0, (Datum) 0);
 
        if (MySubscription->disableonerr)
            DisableSubscriptionAndExit();
        else
        {
            /*
             * Report the worker failed while applying changes. Abort the
             * current transaction so that the stats message is sent in an
             * idle state.
             */
            AbortOutOfAnyTransaction();
            pgstat_report_subscription_error(MySubscription->oid, !am_tablesync_worker());
 
            PG_RE_THROW();
        }
    }
    PG_END_TRY();
}

References AbortOutOfAnyTransaction(), am_tablesync_worker(), Subscription::disableonerr, DisableSubscriptionAndExit(), LogicalRepApplyLoop(), MySubscription, Subscription::oid, PG_CATCH, PG_END_TRY, PG_RE_THROW, PG_TRY, pgstat_report_subscription_error(), and replorigin_reset().

Referenced by run_apply_worker(), and run_tablesync_worker().

◆ store_flush_position()

void store_flush_position	(	XLogRecPtr	remote_lsn,
		XLogRecPtr	local_lsn
	)

Definition at line 3539 of file worker.c.

{
    FlushPosition *flushpos;
 
    /*
     * Skip for parallel apply workers, because the lsn_mapping is maintained
     * by the leader apply worker.
     */
    if (am_parallel_apply_worker())
        return;
 
    /* Need to do this in permanent context */
    MemoryContextSwitchTo(ApplyContext);
 
    /* Track commit lsn  */
    flushpos = (FlushPosition *) palloc(sizeof(FlushPosition));
    flushpos->local_end = local_lsn;
    flushpos->remote_end = remote_lsn;
 
    dlist_push_tail(&lsn_mapping, &flushpos->node);
    MemoryContextSwitchTo(ApplyMessageContext);
}

References am_parallel_apply_worker(), ApplyContext, ApplyMessageContext, dlist_push_tail(), FlushPosition::local_end, lsn_mapping, MemoryContextSwitchTo(), FlushPosition::node, palloc(), and FlushPosition::remote_end.

Referenced by apply_handle_commit_internal(), apply_handle_commit_prepared(), apply_handle_prepare(), apply_handle_rollback_prepared(), apply_handle_stream_prepare(), and pa_xact_finish().

◆ stream_cleanup_files()

void stream_cleanup_files	(	Oid	subid,
		TransactionId	xid
	)

Definition at line 4311 of file worker.c.

{
    char        path[MAXPGPATH];
 
    /* Delete the changes file. */
    changes_filename(path, subid, xid);
    BufFileDeleteFileSet(MyLogicalRepWorker->stream_fileset, path, false);
 
    /* Delete the subxact file, if it exists. */
    subxact_filename(path, subid, xid);
    BufFileDeleteFileSet(MyLogicalRepWorker->stream_fileset, path, true);
}

References BufFileDeleteFileSet(), changes_filename(), MAXPGPATH, MyLogicalRepWorker, LogicalRepWorker::stream_fileset, and subxact_filename().

Referenced by apply_handle_stream_commit(), apply_handle_stream_prepare(), pa_free_worker_info(), and stream_abort_internal().

◆ stream_start_internal()

void stream_start_internal	(	TransactionId	xid,
		bool	first_segment
	)

Definition at line 1453 of file worker.c.

{
    begin_replication_step();
 
    /*
     * Initialize the worker's stream_fileset if we haven't yet. This will be
     * used for the entire duration of the worker so create it in a permanent
     * context. We create this on the very first streaming message from any
     * transaction and then use it for this and other streaming transactions.
     * Now, we could create a fileset at the start of the worker as well but
     * then we won't be sure that it will ever be used.
     */
    if (!MyLogicalRepWorker->stream_fileset)
    {
        MemoryContext oldctx;
 
        oldctx = MemoryContextSwitchTo(ApplyContext);
 
        MyLogicalRepWorker->stream_fileset = palloc(sizeof(FileSet));
        FileSetInit(MyLogicalRepWorker->stream_fileset);
 
        MemoryContextSwitchTo(oldctx);
    }
 
    /* Open the spool file for this transaction. */
    stream_open_file(MyLogicalRepWorker->subid, xid, first_segment);
 
    /* If this is not the first segment, open existing subxact file. */
    if (!first_segment)
        subxact_info_read(MyLogicalRepWorker->subid, xid);
 
    end_replication_step();
}

References ApplyContext, begin_replication_step(), end_replication_step(), FileSetInit(), MemoryContextSwitchTo(), MyLogicalRepWorker, palloc(), LogicalRepWorker::stream_fileset, stream_open_file(), LogicalRepWorker::subid, and subxact_info_read().

Referenced by apply_handle_stream_start(), pa_switch_to_partial_serialize(), and stream_open_and_write_change().

◆ stream_stop_internal()

void stream_stop_internal ( TransactionId xid )

Definition at line 1627 of file worker.c.

{
    /*
     * Serialize information about subxacts for the toplevel transaction, then
     * close the stream messages spool file.
     */
    subxact_info_write(MyLogicalRepWorker->subid, xid);
    stream_close_file();
 
    /* We must be in a valid transaction state */
    Assert(IsTransactionState());
 
    /* Commit the per-stream transaction */
    CommitTransactionCommand();
 
    /* Reset per-stream context */
    MemoryContextReset(LogicalStreamingContext);
}

References Assert(), CommitTransactionCommand(), IsTransactionState(), LogicalStreamingContext, MemoryContextReset(), MyLogicalRepWorker, stream_close_file(), LogicalRepWorker::subid, and subxact_info_write().

Referenced by apply_handle_stream_stop(), and stream_open_and_write_change().

◆ UpdateTwoPhaseState()

void UpdateTwoPhaseState	(	Oid	suboid,
		char	new_state
	)

Definition at line 1763 of file tablesync.c.

{
    Relation    rel;
    HeapTuple   tup;
    bool        nulls[Natts_pg_subscription];
    bool        replaces[Natts_pg_subscription];
    Datum       values[Natts_pg_subscription];
 
    Assert(new_state == LOGICALREP_TWOPHASE_STATE_DISABLED ||
           new_state == LOGICALREP_TWOPHASE_STATE_PENDING ||
           new_state == LOGICALREP_TWOPHASE_STATE_ENABLED);
 
    rel = table_open(SubscriptionRelationId, RowExclusiveLock);
    tup = SearchSysCacheCopy1(SUBSCRIPTIONOID, ObjectIdGetDatum(suboid));
    if (!HeapTupleIsValid(tup))
        elog(ERROR,
             "cache lookup failed for subscription oid %u",
             suboid);
 
    /* Form a new tuple. */
    memset(values, 0, sizeof(values));
    memset(nulls, false, sizeof(nulls));
    memset(replaces, false, sizeof(replaces));
 
    /* And update/set two_phase state */
    values[Anum_pg_subscription_subtwophasestate - 1] = CharGetDatum(new_state);
    replaces[Anum_pg_subscription_subtwophasestate - 1] = true;
 
    tup = heap_modify_tuple(tup, RelationGetDescr(rel),
                            values, nulls, replaces);
    CatalogTupleUpdate(rel, &tup->t_self, tup);
 
    heap_freetuple(tup);
    table_close(rel, RowExclusiveLock);
}

References Assert(), CatalogTupleUpdate(), CharGetDatum(), elog, ERROR, heap_freetuple(), heap_modify_tuple(), HeapTupleIsValid, ObjectIdGetDatum(), RelationGetDescr, RowExclusiveLock, SearchSysCacheCopy1, HeapTupleData::t_self, table_close(), table_open(), and values.

Referenced by CreateSubscription(), and run_apply_worker().

Variable Documentation

◆ apply_error_context_stack

PGDLLIMPORT ErrorContextCallback* apply_error_context_stack

extern

Definition at line 289 of file worker.c.

Referenced by LogicalRepApplyLoop(), and ProcessParallelApplyMessage().

◆ ApplyContext

PGDLLIMPORT MemoryContext ApplyContext

extern

Definition at line 292 of file worker.c.

Referenced by InitializeLogRepWorker(), LogicalParallelApplyLoop(), LogicalRepApplyLoop(), maybe_reread_subscription(), pa_allocate_worker(), pa_launch_parallel_worker(), send_feedback(), set_apply_error_context_origin(), start_table_sync(), store_flush_position(), and stream_start_internal().

◆ ApplyMessageContext

PGDLLIMPORT MemoryContext ApplyMessageContext

extern

Definition at line 291 of file worker.c.

Referenced by apply_spooled_messages(), begin_replication_step(), LogicalParallelApplyLoop(), LogicalRepApplyLoop(), and store_flush_position().

◆ in_remote_transaction

PGDLLIMPORT bool in_remote_transaction

extern

Definition at line 304 of file worker.c.

Referenced by apply_handle_begin(), apply_handle_begin_prepare(), apply_handle_commit_internal(), apply_handle_commit_prepared(), apply_handle_origin(), apply_handle_prepare(), apply_handle_rollback_prepared(), apply_handle_stream_prepare(), apply_spooled_messages(), LogicalRepApplyLoop(), and maybe_start_skipping_changes().

◆ InitializingApplyWorker

PGDLLIMPORT bool InitializingApplyWorker

extern

Definition at line 319 of file worker.c.

Referenced by ApplyWorkerMain(), logicalrep_worker_onexit(), and ParallelApplyWorkerMain().

◆ LogRepWorkerWalRcvConn

PGDLLIMPORT struct WalReceiverConn* LogRepWorkerWalRcvConn

extern

Definition at line 297 of file worker.c.

Referenced by copy_read_data(), copy_table(), fetch_remote_table_info(), logicalrep_worker_onexit(), LogicalRepApplyLoop(), LogicalRepSyncTableStart(), process_syncing_tables_for_sync(), run_apply_worker(), run_tablesync_worker(), send_feedback(), and set_stream_options().

◆ MyLogicalRepWorker

PGDLLIMPORT LogicalRepWorker* MyLogicalRepWorker

extern

Definition at line 54 of file launcher.c.

Referenced by am_leader_apply_worker(), am_parallel_apply_worker(), am_tablesync_worker(), apply_handle_stream_abort(), apply_handle_stream_commit(), apply_handle_stream_prepare(), apply_handle_stream_start(), apply_spooled_messages(), apply_worker_exit(), DisableSubscriptionAndExit(), ensure_last_message(), finish_sync_worker(), InitializeLogRepWorker(), IsLogicalWorker(), logicalrep_worker_attach(), logicalrep_worker_detach(), logicalrep_worker_onexit(), LogicalRepSyncTableStart(), maybe_reread_subscription(), pa_can_start(), pa_free_worker_info(), pa_launch_parallel_worker(), pa_lock_stream(), pa_lock_transaction(), pa_set_fileset_state(), pa_shutdown(), pa_unlock_stream(), pa_unlock_transaction(), ParallelApplyWorkerMain(), process_syncing_tables(), process_syncing_tables_for_apply(), process_syncing_tables_for_sync(), run_tablesync_worker(), set_stream_options(), SetupApplyOrSyncWorker(), should_apply_changes_for_rel(), stream_abort_internal(), stream_cleanup_files(), stream_open_file(), stream_start_internal(), stream_stop_internal(), subxact_info_read(), subxact_info_write(), UpdateWorkerStats(), wait_for_relation_state_change(), and wait_for_worker_state_change().

◆ MyParallelShared

PGDLLIMPORT ParallelApplyWorkerShared* MyParallelShared

extern

Definition at line 239 of file applyparallelworker.c.

Referenced by apply_handle_stream_commit(), apply_handle_stream_prepare(), apply_handle_stream_start(), pa_decr_and_wait_stream_block(), pa_get_fileset_state(), pa_process_spooled_messages_if_required(), pa_stream_abort(), and ParallelApplyWorkerMain().

◆ MySubscription

PGDLLIMPORT Subscription* MySubscription

extern

Definition at line 299 of file worker.c.

Referenced by apply_handle_commit_prepared(), apply_handle_delete(), apply_handle_insert(), apply_handle_prepare_internal(), apply_handle_rollback_prepared(), apply_handle_truncate(), apply_handle_update(), clear_subscription_skip_lsn(), copy_table(), DisableSubscriptionAndExit(), fetch_remote_table_info(), FetchTableStates(), finish_sync_worker(), InitializeLogRepWorker(), logicalrep_rel_open(), LogicalRepSyncTableStart(), maybe_reread_subscription(), maybe_start_skipping_changes(), pa_can_start(), pa_launch_parallel_worker(), pa_start_subtrans(), pa_stream_abort(), ParallelApplyWorkerMain(), process_syncing_tables_for_apply(), ProcessParallelApplyInterrupts(), ReportApplyConflict(), run_apply_worker(), run_tablesync_worker(), set_stream_options(), SetupApplyOrSyncWorker(), should_apply_changes_for_rel(), start_apply(), and start_table_sync().

Variables
PGDLLIMPORT MemoryContext	ApplyContext

PGDLLIMPORT MemoryContext	ApplyMessageContext

PGDLLIMPORT ErrorContextCallback *	apply_error_context_stack

PGDLLIMPORT ParallelApplyWorkerShared *	MyParallelShared

PGDLLIMPORT struct WalReceiverConn *	LogRepWorkerWalRcvConn

PGDLLIMPORT Subscription *	MySubscription

PGDLLIMPORT LogicalRepWorker *	MyLogicalRepWorker

PGDLLIMPORT bool	in_remote_transaction

PGDLLIMPORT bool	InitializingApplyWorker

Data Structures

Macros

Typedefs

Enumerations

Functions

Variables

Macro Definition Documentation

◆ isParallelApplyWorker

◆ isTablesyncWorker

Typedef Documentation

◆ LogicalRepWorker

◆ LogicalRepWorkerType

◆ ParallelApplyWorkerInfo

◆ ParallelApplyWorkerShared

◆ ParallelTransState

◆ PartialFileSetState

Enumeration Type Documentation

◆ LogicalRepWorkerType

◆ ParallelTransState

◆ PartialFileSetState

Function Documentation

◆ AllTablesyncsReady()

◆ am_leader_apply_worker()

◆ am_parallel_apply_worker()

◆ am_tablesync_worker()

◆ apply_dispatch()

◆ apply_error_callback()

◆ apply_spooled_messages()

◆ DisableSubscriptionAndExit()

◆ InitializeLogRepWorker()

◆ invalidate_syncing_table_states()

◆ logicalrep_pa_worker_stop()

◆ logicalrep_sync_worker_count()

◆ logicalrep_worker_attach()

◆ logicalrep_worker_find()

◆ logicalrep_worker_launch()

◆ logicalrep_worker_stop()

◆ logicalrep_worker_wakeup()

◆ logicalrep_worker_wakeup_ptr()

◆ logicalrep_workers_find()

◆ maybe_reread_subscription()

◆ pa_allocate_worker()

◆ pa_decr_and_wait_stream_block()

◆ pa_detach_all_error_mq()

◆ pa_find_worker()

◆ pa_lock_stream()

◆ pa_lock_transaction()

◆ pa_reset_subtrans()

◆ pa_send_data()

◆ pa_set_fileset_state()

◆ pa_set_stream_apply_worker()

◆ pa_set_xact_state()

◆ pa_start_subtrans()

◆ pa_stream_abort()

◆ pa_switch_to_partial_serialize()

◆ pa_unlock_stream()

◆ pa_unlock_transaction()

◆ pa_xact_finish()

◆ process_syncing_tables()

◆ ReplicationOriginNameForLogicalRep()

◆ set_apply_error_context_origin()

◆ set_stream_options()

◆ SetupApplyOrSyncWorker()

◆ start_apply()

◆ store_flush_position()

◆ stream_cleanup_files()

◆ stream_start_internal()

◆ stream_stop_internal()

◆ UpdateTwoPhaseState()

Variable Documentation

◆ apply_error_context_stack

◆ ApplyContext

◆ ApplyMessageContext

◆ in_remote_transaction

◆ InitializingApplyWorker

◆ LogRepWorkerWalRcvConn

◆ MyLogicalRepWorker

◆ MyParallelShared

◆ MySubscription