#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)

#define	isSequenceSyncWorker(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_SEQUENCESYNC , 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 (LogicalRepWorkerType wtype, 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, bool retain_dead_tuples)

void	logicalrep_worker_stop (LogicalRepWorkerType wtype, Oid subid, Oid relid)

void	logicalrep_pa_worker_stop (ParallelApplyWorkerInfo *winfo)

void	logicalrep_worker_wakeup (LogicalRepWorkerType wtype, Oid subid, Oid relid)

void	logicalrep_worker_wakeup_ptr (LogicalRepWorker *worker)

void	logicalrep_reset_seqsync_start_time (void)

int	logicalrep_sync_worker_count (Oid subid)

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

bool	AllTablesyncsReady (void)

bool	HasSubscriptionTablesCached (void)

void	UpdateTwoPhaseState (Oid suboid, char new_state)

void	ProcessSyncingTablesForSync (XLogRecPtr current_lsn)

void	ProcessSyncingTablesForApply (XLogRecPtr current_lsn)

void	ProcessSequencesForSync (void)

pg_noreturn void	FinishSyncWorker (void)

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

void	launch_sync_worker (LogicalRepWorkerType wtype, int nsyncworkers, Oid relid, TimestampTz *last_start_time)

void	ProcessSyncingRelations (XLogRecPtr current_lsn)

void	FetchRelationStates (bool has_pending_subtables, bool has_pending_sequences, bool *started_tx)

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_sequencesync_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:35

Definition at line 362 of file worker_internal.h.

◆ isSequenceSyncWorker

#define isSequenceSyncWorker ( worker )

Value:

((worker)->in_use && \

(worker)->type == WORKERTYPE_SEQUENCESYNC)

WORKERTYPE_SEQUENCESYNC

@ WORKERTYPE_SEQUENCESYNC

Definition: worker_internal.h:33

Definition at line 366 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 364 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_SEQUENCESYNC
WORKERTYPE_APPLY
WORKERTYPE_PARALLEL_APPLY

Definition at line 29 of file worker_internal.h.

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

◆ ParallelTransState

enum ParallelTransState

Enumerator
PARALLEL_TRANS_UNKNOWN
PARALLEL_TRANS_STARTED
PARALLEL_TRANS_FINISHED

Definition at line 120 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 143 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 1600 of file tablesync.c.

{
    bool        started_tx;
    bool        has_tables;
 
    /* We need up-to-date sync state info for subscription tables here. */
    FetchRelationStates(&has_tables, NULL, &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_tables && (table_states_not_ready == NIL);
}

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

Referenced by pa_can_start(), ProcessSyncingTablesForApply(), run_apply_worker(), and wait_for_local_flush().

◆ am_leader_apply_worker()

static bool am_leader_apply_worker ( void )

inlinestatic

Definition at line 382 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(), can_advance_nonremovable_xid(), DisableSubscriptionAndExit(), FindDeletedTupleInLocalRel(), 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 389 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_sequencesync_worker()

static bool am_sequencesync_worker ( void )

inlinestatic

Definition at line 376 of file worker_internal.h.

{
    return isSequenceSyncWorker(MyLogicalRepWorker);
}

References isSequenceSyncWorker, and MyLogicalRepWorker.

Referenced by FinishSyncWorker(), InitializeLogRepWorker(), SetupApplyOrSyncWorker(), and start_sequence_sync().

◆ am_tablesync_worker()

static bool am_tablesync_worker ( void )

inlinestatic

Definition at line 370 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(), FinishSyncWorker(), InitializeLogRepWorker(), SetupApplyOrSyncWorker(), and start_table_sync().

◆ apply_dispatch()

void apply_dispatch ( StringInfo s )

Definition at line 3775 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 6159 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 (!XLogRecPtrIsValid(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/%08X",
                       errarg->origin_name,
                       logicalrep_message_type(errarg->command),
                       errarg->remote_xid,
                       LSN_FORMAT_ARGS(errarg->finish_lsn));
    }
    else
    {
        if (errarg->remote_attnum < 0)
        {
            if (!XLogRecPtrIsValid(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/%08X",
                           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 (!XLogRecPtrIsValid(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/%08X",
                           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 XLogRecPtrIsValid.

Referenced by LogicalParallelApplyLoop(), and LogicalRepApplyLoop().

◆ apply_spooled_messages()

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

Definition at line 2260 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 5943 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 sequence synchronization, table
     * synchronization, or apply.
     */
    pgstat_report_subscription_error(MyLogicalRepWorker->subid,
                                     MyLogicalRepWorker->type);
 
    /* Disable the subscription */
    StartTransactionCommand();
 
    /*
     * Updating pg_subscription might involve TOAST table access, so ensure we
     * have a valid snapshot.
     */
    PushActiveSnapshot(GetTransactionSnapshot());
 
    DisableSubscription(MySubscription->oid);
    PopActiveSnapshot();
    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));
 
    /*
     * Skip the track_commit_timestamp check when disabling the worker due to
     * an error, as verifying commit timestamps is unnecessary in this
     * context.
     */
    CheckSubDeadTupleRetention(false, true, WARNING,
                               MySubscription->retaindeadtuples,
                               MySubscription->retentionactive, false);
 
    proc_exit(0);
}

References AbortOutOfAnyTransaction(), am_leader_apply_worker(), ApplyLauncherForgetWorkerStartTime(), CheckSubDeadTupleRetention(), CommitTransactionCommand(), DisableSubscription(), EmitErrorReport(), ereport, errmsg(), FlushErrorState(), GetTransactionSnapshot(), HOLD_INTERRUPTS, LOG, MyLogicalRepWorker, MySubscription, Subscription::name, Subscription::oid, pgstat_report_subscription_error(), PopActiveSnapshot(), proc_exit(), PushActiveSnapshot(), RESUME_INTERRUPTS, Subscription::retaindeadtuples, Subscription::retentionactive, StartTransactionCommand(), LogicalRepWorker::subid, LogicalRepWorker::type, and WARNING.

Referenced by start_apply(), start_sequence_sync(), and start_table_sync().

◆ FetchRelationStates()

void FetchRelationStates	(	bool *	has_pending_subtables,
		bool *	has_pending_sequences,
		bool *	started_tx
	)

Definition at line 202 of file syncutils.c.

{
    /*
     * has_subtables and has_subsequences_non_ready are declared as static,
     * since the same value can be used until the system table is invalidated.
     */
    static bool has_subtables = false;
    static bool has_subsequences_non_ready = false;
 
    *started_tx = false;
 
    if (relation_states_validity != SYNC_RELATIONS_STATE_VALID)
    {
        MemoryContext oldctx;
        List       *rstates;
        SubscriptionRelState *rstate;
 
        relation_states_validity = SYNC_RELATIONS_STATE_REBUILD_STARTED;
        has_subsequences_non_ready = false;
 
        /* Clean the old lists. */
        list_free_deep(table_states_not_ready);
        table_states_not_ready = NIL;
 
        if (!IsTransactionState())
        {
            StartTransactionCommand();
            *started_tx = true;
        }
 
        /* Fetch tables and sequences that are in non-READY state. */
        rstates = GetSubscriptionRelations(MySubscription->oid, true, true,
                                           true);
 
        /* Allocate the tracking info in a permanent memory context. */
        oldctx = MemoryContextSwitchTo(CacheMemoryContext);
        foreach_ptr(SubscriptionRelState, subrel, rstates)
        {
            if (get_rel_relkind(subrel->relid) == RELKIND_SEQUENCE)
                has_subsequences_non_ready = true;
            else
            {
                rstate = palloc(sizeof(SubscriptionRelState));
                memcpy(rstate, subrel, sizeof(SubscriptionRelState));
                table_states_not_ready = lappend(table_states_not_ready,
                                                 rstate);
            }
        }
        MemoryContextSwitchTo(oldctx);
 
        /*
         * Does the subscription have tables?
         *
         * If there were not-READY tables found then we know it does. But if
         * table_states_not_ready was empty we still need to check again to
         * see if there are 0 tables.
         */
        has_subtables = (table_states_not_ready != NIL) ||
            HasSubscriptionTables(MySubscription->oid);
 
        /*
         * If the subscription relation cache has been invalidated since we
         * entered this routine, we still use and return the relations we just
         * finished constructing, to avoid infinite loops, but we leave the
         * table states marked as stale so that we'll rebuild it again on next
         * access. Otherwise, we mark the table states as valid.
         */
        if (relation_states_validity == SYNC_RELATIONS_STATE_REBUILD_STARTED)
            relation_states_validity = SYNC_RELATIONS_STATE_VALID;
    }
 
    if (has_pending_subtables)
        *has_pending_subtables = has_subtables;
 
    if (has_pending_subsequences)
        *has_pending_subsequences = has_subsequences_non_ready;
}

References CacheMemoryContext, foreach_ptr, get_rel_relkind(), GetSubscriptionRelations(), HasSubscriptionTables(), IsTransactionState(), lappend(), list_free_deep(), MemoryContextSwitchTo(), MySubscription, NIL, Subscription::oid, palloc(), relation_states_validity, StartTransactionCommand(), SYNC_RELATIONS_STATE_REBUILD_STARTED, SYNC_RELATIONS_STATE_VALID, and table_states_not_ready.

Referenced by AllTablesyncsReady(), HasSubscriptionTablesCached(), ProcessSequencesForSync(), and ProcessSyncingTablesForApply().

◆ FinishSyncWorker()

pg_noreturn void FinishSyncWorker ( void )

Definition at line 50 of file syncutils.c.

{
    Assert(am_sequencesync_worker() || am_tablesync_worker());
 
    /*
     * Commit any outstanding transaction. This is the usual case, unless
     * there was nothing to do for the table.
     */
    if (IsTransactionState())
    {
        CommitTransactionCommand();
        pgstat_report_stat(true);
    }
 
    /* And flush all writes. */
    XLogFlush(GetXLogWriteRecPtr());
 
    if (am_sequencesync_worker())
    {
        ereport(LOG,
                errmsg("logical replication sequence synchronization worker for subscription \"%s\" has finished",
                       MySubscription->name));
 
        /*
         * Reset last_seqsync_start_time, so that next time a sequencesync
         * worker is needed it can be started promptly.
         */
        logicalrep_reset_seqsync_start_time();
    }
    else
    {
        StartTransactionCommand();
        ereport(LOG,
                errmsg("logical replication table synchronization worker for subscription \"%s\", table \"%s\" has finished",
                       MySubscription->name,
                       get_rel_name(MyLogicalRepWorker->relid)));
        CommitTransactionCommand();
 
        /* Find the leader apply worker and signal it. */
        logicalrep_worker_wakeup(WORKERTYPE_APPLY, MyLogicalRepWorker->subid,
                                 InvalidOid);
    }
 
    /* Stop gracefully */
    proc_exit(0);
}

References am_sequencesync_worker(), am_tablesync_worker(), Assert(), CommitTransactionCommand(), ereport, errmsg(), get_rel_name(), GetXLogWriteRecPtr(), InvalidOid, IsTransactionState(), LOG, logicalrep_reset_seqsync_start_time(), logicalrep_worker_wakeup(), MyLogicalRepWorker, MySubscription, Subscription::name, pgstat_report_stat(), proc_exit(), LogicalRepWorker::relid, StartTransactionCommand(), LogicalRepWorker::subid, WORKERTYPE_APPLY, and XLogFlush().

Referenced by LogicalRepSyncTableStart(), ProcessSyncingTablesForSync(), SequenceSyncWorkerMain(), and TableSyncWorkerMain().

◆ HasSubscriptionTablesCached()

bool HasSubscriptionTablesCached ( void )

Definition at line 1630 of file tablesync.c.

{
    bool        started_tx;
    bool        has_tables;
 
    /* We need up-to-date subscription tables info here */
    FetchRelationStates(&has_tables, NULL, &started_tx);
 
    if (started_tx)
    {
        CommitTransactionCommand();
        pgstat_report_stat(true);
    }
 
    return has_tables;
}

References CommitTransactionCommand(), FetchRelationStates(), and pgstat_report_stat().

Referenced by wait_for_local_flush().

◆ InitializeLogRepWorker()

void InitializeLogRepWorker ( void )

Definition at line 5737 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);
 
    /*
     * Lock the subscription to prevent it from being concurrently dropped,
     * then re-verify its existence. After the initialization, the worker will
     * be terminated gracefully if the subscription is dropped.
     */
    LockSharedObject(SubscriptionRelationId, MyLogicalRepWorker->subid, 0,
                     AccessShareLock);
    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();
    }
 
    /*
     * Restart the worker if retain_dead_tuples was enabled during startup.
     *
     * At this point, the replication slot used for conflict detection might
     * not exist yet, or could be dropped soon if the launcher perceives
     * retain_dead_tuples as disabled. To avoid unnecessary tracking of
     * oldest_nonremovable_xid when the slot is absent or at risk of being
     * dropped, a restart is initiated.
     *
     * The oldest_nonremovable_xid should be initialized only when the
     * subscription's retention is active before launching the worker. See
     * logicalrep_worker_launch.
     */
    if (am_leader_apply_worker() &&
        MySubscription->retaindeadtuples &&
        MySubscription->retentionactive &&
        !TransactionIdIsValid(MyLogicalRepWorker->oldest_nonremovable_xid))
    {
        ereport(LOG,
                errmsg("logical replication worker for subscription \"%s\" will restart because the option %s was enabled during startup",
                       MySubscription->name, "retain_dead_tuples"));
 
        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 if (am_sequencesync_worker())
        ereport(LOG,
                errmsg("logical replication sequence synchronization worker for subscription \"%s\" has started",
                       MySubscription->name));
    else
        ereport(LOG,
                errmsg("logical replication apply worker for subscription \"%s\" has started",
                       MySubscription->name));
 
    CommitTransactionCommand();
}

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

Referenced by ParallelApplyWorkerMain(), and SetupApplyOrSyncWorker().

◆ InvalidateSyncingRelStates()

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

Definition at line 101 of file syncutils.c.

{
    relation_states_validity = SYNC_RELATIONS_STATE_NEEDS_REBUILD;
}

References relation_states_validity, and SYNC_RELATIONS_STATE_NEEDS_REBUILD.

Referenced by ParallelApplyWorkerMain(), and SetupApplyOrSyncWorker().

◆ launch_sync_worker()

void launch_sync_worker	(	LogicalRepWorkerType	wtype,
		int	nsyncworkers,
		Oid	relid,
		TimestampTz *	last_start_time
	)

Definition at line 117 of file syncutils.c.

{
    TimestampTz now;
 
    Assert((wtype == WORKERTYPE_TABLESYNC && OidIsValid(relid)) ||
           (wtype == WORKERTYPE_SEQUENCESYNC && !OidIsValid(relid)));
 
    /* If there is a free sync worker slot, start a new sync worker */
    if (nsyncworkers >= max_sync_workers_per_subscription)
        return;
 
    now = GetCurrentTimestamp();
 
    if (!(*last_start_time) ||
        TimestampDifferenceExceeds(*last_start_time, now,
                                   wal_retrieve_retry_interval))
    {
        /*
         * Set the last_start_time even if we fail to start the worker, so
         * that we won't retry until wal_retrieve_retry_interval has elapsed.
         */
        *last_start_time = now;
        (void) logicalrep_worker_launch(wtype,
                                        MyLogicalRepWorker->dbid,
                                        MySubscription->oid,
                                        MySubscription->name,
                                        MyLogicalRepWorker->userid,
                                        relid, DSM_HANDLE_INVALID, false);
    }
}

References Assert(), LogicalRepWorker::dbid, DSM_HANDLE_INVALID, GetCurrentTimestamp(), logicalrep_worker_launch(), max_sync_workers_per_subscription, MyLogicalRepWorker, MySubscription, Subscription::name, now(), Subscription::oid, OidIsValid, TimestampDifferenceExceeds(), LogicalRepWorker::userid, wal_retrieve_retry_interval, WORKERTYPE_SEQUENCESYNC, and WORKERTYPE_TABLESYNC.

Referenced by ProcessSequencesForSync(), and ProcessSyncingTablesForApply().

◆ logicalrep_pa_worker_stop()

void logicalrep_pa_worker_stop ( ParallelApplyWorkerInfo * winfo )

Definition at line 679 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, SIGUSR2);
 
    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(), SIGUSR2, SpinLockAcquire, SpinLockRelease, and LogicalRepCtxStruct::workers.

Referenced by pa_free_worker().

◆ logicalrep_reset_seqsync_start_time()

void logicalrep_reset_seqsync_start_time ( void )

Definition at line 872 of file launcher.c.

{
    LogicalRepWorker *worker;
 
    /*
     * The apply worker can't access last_seqsync_start_time concurrently, so
     * it is okay to use SHARED lock here. See ProcessSequencesForSync().
     */
    LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
 
    worker = logicalrep_worker_find(WORKERTYPE_APPLY,
                                    MyLogicalRepWorker->subid, InvalidOid,
                                    true);
    if (worker)
        worker->last_seqsync_start_time = 0;
 
    LWLockRelease(LogicalRepWorkerLock);
}

References InvalidOid, LogicalRepWorker::last_seqsync_start_time, logicalrep_worker_find(), LW_SHARED, LWLockAcquire(), LWLockRelease(), MyLogicalRepWorker, LogicalRepWorker::subid, and WORKERTYPE_APPLY.

Referenced by FinishSyncWorker().

◆ logicalrep_sync_worker_count()

int logicalrep_sync_worker_count ( Oid subid )

Definition at line 927 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 (w->subid == subid && (isTableSyncWorker(w) || isSequenceSyncWorker(w)))
            res++;
    }
 
    return res;
}

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

Referenced by logicalrep_worker_launch(), ProcessSequencesForSync(), and ProcessSyncingTablesForApply().

◆ logicalrep_worker_attach()

void logicalrep_worker_attach ( int slot )

Definition at line 757 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	(	LogicalRepWorkerType	wtype,
		Oid	subid,
		Oid	relid,
		bool	only_running
	)

Definition at line 258 of file launcher.c.

{
    int         i;
    LogicalRepWorker *res = NULL;
 
    /* relid must be valid only for table sync workers */
    Assert((wtype == WORKERTYPE_TABLESYNC) == OidIsValid(relid));
    Assert(LWLockHeldByMe(LogicalRepWorkerLock));
 
    /* Search for an attached worker that matches the specified criteria. */
    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 &&
            w->type == wtype && (!only_running || w->proc))
        {
            res = w;
            break;
        }
    }
 
    return res;
}

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

Referenced by ApplyLauncherMain(), FindDeletedTupleInLocalRel(), logicalrep_reset_seqsync_start_time(), logicalrep_worker_stop(), logicalrep_worker_wakeup(), ProcessSequencesForSync(), ProcessSyncingTablesForApply(), wait_for_table_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,
		bool	retain_dead_tuples
	)

Definition at line 324 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_sequencesync_worker = (wtype == WORKERTYPE_SEQUENCESYNC);
    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
     * - The replication slot used in conflict detection is created when
     *   retain_dead_tuples is enabled
     */
    Assert(wtype != WORKERTYPE_UNKNOWN);
    Assert(is_tablesync_worker == OidIsValid(relid));
    Assert(is_parallel_apply_worker == (subworker_dsm != DSM_HANDLE_INVALID));
    Assert(!retain_dead_tuples || MyReplicationSlot);
 
    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\" is 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 || is_sequencesync_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->oldest_nonremovable_xid = retain_dead_tuples
        ? MyReplicationSlot->data.xmin
        : InvalidTransactionId;
    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);
    worker->last_seqsync_start_time = 0;
 
    /* 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_SEQUENCESYNC:
            snprintf(bgw.bgw_function_name, BGW_MAXLEN, "SequenceSyncWorkerMain");
            snprintf(bgw.bgw_name, BGW_MAXLEN,
                     "logical replication sequencesync worker for subscription %u",
                     subid);
            snprintf(bgw.bgw_type, BGW_MAXLEN, "logical replication sequencesync worker");
            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(), launch_sync_worker(), and pa_launch_parallel_worker().

◆ logicalrep_worker_stop()

void logicalrep_worker_stop	(	LogicalRepWorkerType	wtype,
		Oid	subid,
		Oid	relid
	)

Definition at line 652 of file launcher.c.

{
    LogicalRepWorker *worker;
 
    /* relid must be valid only for table sync workers */
    Assert((wtype == WORKERTYPE_TABLESYNC) == OidIsValid(relid));
 
    LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
 
    worker = logicalrep_worker_find(wtype, 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(), LWLockRelease(), OidIsValid, and WORKERTYPE_TABLESYNC.

Referenced by AlterSubscription_refresh(), and DropSubscription().

◆ logicalrep_worker_wakeup()

void logicalrep_worker_wakeup	(	LogicalRepWorkerType	wtype,
		Oid	subid,
		Oid	relid
	)

Definition at line 723 of file launcher.c.

{
    LogicalRepWorker *worker;
 
    /* relid must be valid only for table sync workers */
    Assert((wtype == WORKERTYPE_TABLESYNC) == OidIsValid(relid));
 
    LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
 
    worker = logicalrep_worker_find(wtype, subid, relid, true);
 
    if (worker)
        logicalrep_worker_wakeup_ptr(worker);
 
    LWLockRelease(LogicalRepWorkerLock);
}

References Assert(), logicalrep_worker_find(), logicalrep_worker_wakeup_ptr(), LW_SHARED, LWLockAcquire(), LWLockRelease(), OidIsValid, and WORKERTYPE_TABLESYNC.

Referenced by apply_handle_stream_start(), and FinishSyncWorker().

◆ logicalrep_worker_wakeup_ptr()

void logicalrep_worker_wakeup_ptr ( LogicalRepWorker * worker )

Definition at line 746 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(), ProcessSyncingTablesForApply(), and wait_for_worker_state_change().

◆ logicalrep_workers_find()

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

Definition at line 293 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 5038 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 471 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 1599 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 623 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 519 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 1548 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 1581 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 1410 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 1154 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 1506 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 1342 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 1315 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 1370 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 1424 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 1219 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 1555 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 1588 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 1626 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 (XLogRecPtrIsValid(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 XLogRecPtrIsValid.

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

◆ ProcessSequencesForSync()

void ProcessSequencesForSync ( void )

Definition at line 94 of file sequencesync.c.

{
    LogicalRepWorker *sequencesync_worker;
    int         nsyncworkers;
    bool        has_pending_sequences;
    bool        started_tx;
 
    FetchRelationStates(NULL, &has_pending_sequences, &started_tx);
 
    if (started_tx)
    {
        CommitTransactionCommand();
        pgstat_report_stat(true);
    }
 
    if (!has_pending_sequences)
        return;
 
    LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
 
    /* Check if there is a sequencesync worker already running? */
    sequencesync_worker = logicalrep_worker_find(WORKERTYPE_SEQUENCESYNC,
                                                 MyLogicalRepWorker->subid,
                                                 InvalidOid, true);
    if (sequencesync_worker)
    {
        LWLockRelease(LogicalRepWorkerLock);
        return;
    }
 
    /*
     * Count running sync workers for this subscription, while we have the
     * lock.
     */
    nsyncworkers = logicalrep_sync_worker_count(MyLogicalRepWorker->subid);
    LWLockRelease(LogicalRepWorkerLock);
 
    /*
     * It is okay to read/update last_seqsync_start_time here in apply worker
     * as we have already ensured that sync worker doesn't exist.
     */
    launch_sync_worker(WORKERTYPE_SEQUENCESYNC, nsyncworkers, InvalidOid,
                       &MyLogicalRepWorker->last_seqsync_start_time);
}

References CommitTransactionCommand(), FetchRelationStates(), InvalidOid, LogicalRepWorker::last_seqsync_start_time, launch_sync_worker(), logicalrep_sync_worker_count(), logicalrep_worker_find(), LW_SHARED, LWLockAcquire(), LWLockRelease(), MyLogicalRepWorker, pgstat_report_stat(), LogicalRepWorker::subid, and WORKERTYPE_SEQUENCESYNC.

Referenced by ProcessSyncingRelations().

◆ ProcessSyncingRelations()

void ProcessSyncingRelations ( XLogRecPtr current_lsn )

Definition at line 155 of file syncutils.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:
            ProcessSyncingTablesForSync(current_lsn);
            break;
 
        case WORKERTYPE_APPLY:
            ProcessSyncingTablesForApply(current_lsn);
            ProcessSequencesForSync();
            break;
 
        case WORKERTYPE_SEQUENCESYNC:
            /* Should never happen. */
            elog(ERROR, "sequence synchronization worker is not expected to process relations");
            break;
 
        case WORKERTYPE_UNKNOWN:
            /* Should never happen. */
            elog(ERROR, "Unknown worker type");
    }
}

References elog, ERROR, MyLogicalRepWorker, ProcessSequencesForSync(), ProcessSyncingTablesForApply(), ProcessSyncingTablesForSync(), LogicalRepWorker::type, WORKERTYPE_APPLY, WORKERTYPE_PARALLEL_APPLY, WORKERTYPE_SEQUENCESYNC, 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().

◆ ProcessSyncingTablesForApply()

void ProcessSyncingTablesForApply ( XLogRecPtr current_lsn )

Definition at line 368 of file tablesync.c.

{
    struct tablesync_start_time_mapping
    {
        Oid         relid;
        TimestampTz last_start_time;
    };
    static HTAB *last_start_times = NULL;
    ListCell   *lc;
    bool        started_tx;
    bool        should_exit = false;
    Relation    rel = NULL;
 
    Assert(!IsTransactionState());
 
    /* We need up-to-date sync state info for subscription tables here. */
    FetchRelationStates(NULL, NULL, &started_tx);
 
    /*
     * Prepare a hash table for tracking last start times of workers, to avoid
     * immediate restarts.  We don't need it if there are no tables that need
     * syncing.
     */
    if (table_states_not_ready != NIL && !last_start_times)
    {
        HASHCTL     ctl;
 
        ctl.keysize = sizeof(Oid);
        ctl.entrysize = sizeof(struct tablesync_start_time_mapping);
        last_start_times = hash_create("Logical replication table sync worker start times",
                                       256, &ctl, HASH_ELEM | HASH_BLOBS);
    }
 
    /*
     * Clean up the hash table when we're done with all tables (just to
     * release the bit of memory).
     */
    else if (table_states_not_ready == NIL && last_start_times)
    {
        hash_destroy(last_start_times);
        last_start_times = NULL;
    }
 
    /*
     * Process all tables that are being synchronized.
     */
    foreach(lc, table_states_not_ready)
    {
        SubscriptionRelState *rstate = (SubscriptionRelState *) lfirst(lc);
 
        if (!started_tx)
        {
            StartTransactionCommand();
            started_tx = true;
        }
 
        Assert(get_rel_relkind(rstate->relid) != RELKIND_SEQUENCE);
 
        if (rstate->state == SUBREL_STATE_SYNCDONE)
        {
            /*
             * Apply has caught up to the position where the table sync has
             * finished.  Mark the table as ready so that the apply will just
             * continue to replicate it normally.
             */
            if (current_lsn >= rstate->lsn)
            {
                char        originname[NAMEDATALEN];
 
                rstate->state = SUBREL_STATE_READY;
                rstate->lsn = current_lsn;
 
                /*
                 * Remove the tablesync origin tracking if exists.
                 *
                 * There is a chance that the user is concurrently performing
                 * refresh for the subscription where we remove the table
                 * state and its origin or the tablesync worker would have
                 * already removed this origin. We can't rely on tablesync
                 * worker to remove the origin tracking as if there is any
                 * error while dropping we won't restart it to drop the
                 * origin. So passing missing_ok = true.
                 *
                 * Lock the subscription and origin in the same order as we
                 * are doing during DDL commands to avoid deadlocks. See
                 * AlterSubscription_refresh.
                 */
                LockSharedObject(SubscriptionRelationId, MyLogicalRepWorker->subid,
                                 0, AccessShareLock);
 
                if (!rel)
                    rel = table_open(SubscriptionRelRelationId, RowExclusiveLock);
 
                ReplicationOriginNameForLogicalRep(MyLogicalRepWorker->subid,
                                                   rstate->relid,
                                                   originname,
                                                   sizeof(originname));
                replorigin_drop_by_name(originname, true, false);
 
                /*
                 * Update the state to READY only after the origin cleanup.
                 */
                UpdateSubscriptionRelState(MyLogicalRepWorker->subid,
                                           rstate->relid, rstate->state,
                                           rstate->lsn, true);
            }
        }
        else
        {
            LogicalRepWorker *syncworker;
 
            /*
             * Look for a sync worker for this relation.
             */
            LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
 
            syncworker = logicalrep_worker_find(WORKERTYPE_TABLESYNC,
                                                MyLogicalRepWorker->subid,
                                                rstate->relid, false);
 
            if (syncworker)
            {
                /* Found one, update our copy of its state */
                SpinLockAcquire(&syncworker->relmutex);
                rstate->state = syncworker->relstate;
                rstate->lsn = syncworker->relstate_lsn;
                if (rstate->state == SUBREL_STATE_SYNCWAIT)
                {
                    /*
                     * Sync worker is waiting for apply.  Tell sync worker it
                     * can catchup now.
                     */
                    syncworker->relstate = SUBREL_STATE_CATCHUP;
                    syncworker->relstate_lsn =
                        Max(syncworker->relstate_lsn, current_lsn);
                }
                SpinLockRelease(&syncworker->relmutex);
 
                /* If we told worker to catch up, wait for it. */
                if (rstate->state == SUBREL_STATE_SYNCWAIT)
                {
                    /* Signal the sync worker, as it may be waiting for us. */
                    if (syncworker->proc)
                        logicalrep_worker_wakeup_ptr(syncworker);
 
                    /* Now safe to release the LWLock */
                    LWLockRelease(LogicalRepWorkerLock);
 
                    if (started_tx)
                    {
                        /*
                         * We must commit the existing transaction to release
                         * the existing locks before entering a busy loop.
                         * This is required to avoid any undetected deadlocks
                         * due to any existing lock as deadlock detector won't
                         * be able to detect the waits on the latch.
                         *
                         * Also close any tables prior to the commit.
                         */
                        if (rel)
                        {
                            table_close(rel, NoLock);
                            rel = NULL;
                        }
                        CommitTransactionCommand();
                        pgstat_report_stat(false);
                    }
 
                    /*
                     * Enter busy loop and wait for synchronization worker to
                     * reach expected state (or die trying).
                     */
                    StartTransactionCommand();
                    started_tx = true;
 
                    wait_for_table_state_change(rstate->relid,
                                                SUBREL_STATE_SYNCDONE);
                }
                else
                    LWLockRelease(LogicalRepWorkerLock);
            }
            else
            {
                /*
                 * If there is no sync worker for this table yet, count
                 * running sync workers for this subscription, while we have
                 * the lock.
                 */
                int         nsyncworkers =
                    logicalrep_sync_worker_count(MyLogicalRepWorker->subid);
                struct tablesync_start_time_mapping *hentry;
                bool        found;
 
                /* Now safe to release the LWLock */
                LWLockRelease(LogicalRepWorkerLock);
 
                hentry = hash_search(last_start_times, &rstate->relid,
                                     HASH_ENTER, &found);
                if (!found)
                    hentry->last_start_time = 0;
 
                launch_sync_worker(WORKERTYPE_TABLESYNC, nsyncworkers,
                                   rstate->relid, &hentry->last_start_time);
            }
        }
    }
 
    /* Close table if opened */
    if (rel)
        table_close(rel, NoLock);
 
 
    if (started_tx)
    {
        /*
         * Even when the two_phase mode is requested by the user, it remains
         * as 'pending' until all tablesyncs have reached READY state.
         *
         * When this happens, we restart the apply worker and (if the
         * conditions are still ok) then the two_phase tri-state will become
         * 'enabled' at that time.
         *
         * 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)
        {
            CommandCounterIncrement();  /* make updates visible */
            if (AllTablesyncsReady())
            {
                ereport(LOG,
                        (errmsg("logical replication apply worker for subscription \"%s\" will restart so that two_phase can be enabled",
                                MySubscription->name)));
                should_exit = true;
            }
        }
 
        CommitTransactionCommand();
        pgstat_report_stat(true);
    }
 
    if (should_exit)
    {
        /*
         * Reset the last-start time for this worker so that the launcher will
         * restart it without waiting for wal_retrieve_retry_interval.
         */
        ApplyLauncherForgetWorkerStartTime(MySubscription->oid);
 
        proc_exit(0);
    }
}

References AccessShareLock, AllTablesyncsReady(), ApplyLauncherForgetWorkerStartTime(), Assert(), CommandCounterIncrement(), CommitTransactionCommand(), ctl, ereport, errmsg(), FetchRelationStates(), get_rel_relkind(), HASH_BLOBS, hash_create(), hash_destroy(), HASH_ELEM, HASH_ENTER, hash_search(), IsTransactionState(), last_start_times, launch_sync_worker(), lfirst, LockSharedObject(), LOG, logicalrep_sync_worker_count(), logicalrep_worker_find(), logicalrep_worker_wakeup_ptr(), SubscriptionRelState::lsn, LW_SHARED, LWLockAcquire(), LWLockRelease(), Max, MyLogicalRepWorker, MySubscription, Subscription::name, NAMEDATALEN, NIL, NoLock, Subscription::oid, pgstat_report_stat(), LogicalRepWorker::proc, proc_exit(), SubscriptionRelState::relid, LogicalRepWorker::relmutex, LogicalRepWorker::relstate, LogicalRepWorker::relstate_lsn, ReplicationOriginNameForLogicalRep(), replorigin_drop_by_name(), RowExclusiveLock, SpinLockAcquire, SpinLockRelease, StartTransactionCommand(), SubscriptionRelState::state, LogicalRepWorker::subid, table_close(), table_open(), table_states_not_ready, Subscription::twophasestate, UpdateSubscriptionRelState(), wait_for_table_state_change(), and WORKERTYPE_TABLESYNC.

Referenced by ProcessSyncingRelations().

◆ ProcessSyncingTablesForSync()

void ProcessSyncingTablesForSync ( XLogRecPtr current_lsn )

Definition at line 244 of file tablesync.c.

{
    SpinLockAcquire(&MyLogicalRepWorker->relmutex);
 
    if (MyLogicalRepWorker->relstate == SUBREL_STATE_CATCHUP &&
        current_lsn >= MyLogicalRepWorker->relstate_lsn)
    {
        TimeLineID  tli;
        char        syncslotname[NAMEDATALEN] = {0};
        char        originname[NAMEDATALEN] = {0};
 
        MyLogicalRepWorker->relstate = SUBREL_STATE_SYNCDONE;
        MyLogicalRepWorker->relstate_lsn = current_lsn;
 
        SpinLockRelease(&MyLogicalRepWorker->relmutex);
 
        /*
         * UpdateSubscriptionRelState must be called within a transaction.
         */
        if (!IsTransactionState())
            StartTransactionCommand();
 
        UpdateSubscriptionRelState(MyLogicalRepWorker->subid,
                                   MyLogicalRepWorker->relid,
                                   MyLogicalRepWorker->relstate,
                                   MyLogicalRepWorker->relstate_lsn,
                                   false);
 
        /*
         * End streaming so that LogRepWorkerWalRcvConn can be used to drop
         * the slot.
         */
        walrcv_endstreaming(LogRepWorkerWalRcvConn, &tli);
 
        /*
         * Cleanup the tablesync slot.
         *
         * This has to be done after updating the state because otherwise if
         * there is an error while doing the database operations we won't be
         * able to rollback dropped slot.
         */
        ReplicationSlotNameForTablesync(MyLogicalRepWorker->subid,
                                        MyLogicalRepWorker->relid,
                                        syncslotname,
                                        sizeof(syncslotname));
 
        /*
         * It is important to give an error if we are unable to drop the slot,
         * otherwise, it won't be dropped till the corresponding subscription
         * is dropped. So passing missing_ok = false.
         */
        ReplicationSlotDropAtPubNode(LogRepWorkerWalRcvConn, syncslotname, false);
 
        CommitTransactionCommand();
        pgstat_report_stat(false);
 
        /*
         * Start a new transaction to clean up the tablesync origin tracking.
         * This transaction will be ended within the FinishSyncWorker(). Now,
         * even, if we fail to remove this here, the apply worker will ensure
         * to clean it up afterward.
         *
         * We need to do this after the table state is set to SYNCDONE.
         * Otherwise, if an error occurs while performing the database
         * operation, the worker will be restarted and the in-memory state of
         * replication progress (remote_lsn) won't be rolled-back which would
         * have been cleared before restart. So, the restarted worker will use
         * invalid replication progress state resulting in replay of
         * transactions that have already been applied.
         */
        StartTransactionCommand();
 
        ReplicationOriginNameForLogicalRep(MyLogicalRepWorker->subid,
                                           MyLogicalRepWorker->relid,
                                           originname,
                                           sizeof(originname));
 
        /*
         * Resetting the origin session removes the ownership of the slot.
         * This is needed to allow the origin to be dropped.
         */
        replorigin_session_reset();
        replorigin_session_origin = InvalidRepOriginId;
        replorigin_session_origin_lsn = InvalidXLogRecPtr;
        replorigin_session_origin_timestamp = 0;
 
        /*
         * Drop the tablesync's origin tracking if exists.
         *
         * There is a chance that the user is concurrently performing refresh
         * for the subscription where we remove the table state and its origin
         * or the apply worker would have removed this origin. So passing
         * missing_ok = true.
         */
        replorigin_drop_by_name(originname, true, false);
 
        FinishSyncWorker();
    }
    else
        SpinLockRelease(&MyLogicalRepWorker->relmutex);
}

References CommitTransactionCommand(), FinishSyncWorker(), InvalidRepOriginId, InvalidXLogRecPtr, IsTransactionState(), LogRepWorkerWalRcvConn, MyLogicalRepWorker, NAMEDATALEN, pgstat_report_stat(), LogicalRepWorker::relid, LogicalRepWorker::relmutex, LogicalRepWorker::relstate, LogicalRepWorker::relstate_lsn, ReplicationOriginNameForLogicalRep(), ReplicationSlotDropAtPubNode(), ReplicationSlotNameForTablesync(), replorigin_drop_by_name(), replorigin_session_origin, replorigin_session_origin_lsn, replorigin_session_origin_timestamp, replorigin_session_reset(), SpinLockAcquire, SpinLockRelease, StartTransactionCommand(), LogicalRepWorker::subid, UpdateSubscriptionRelState(), and walrcv_endstreaming.

Referenced by ProcessSyncingRelations().

◆ ReplicationOriginNameForLogicalRep()

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

Definition at line 641 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(), ProcessSyncingTablesForApply(), ProcessSyncingTablesForSync(), run_apply_worker(), and run_tablesync_worker().

◆ set_apply_error_context_origin()

void set_apply_error_context_origin ( char * originname )

Definition at line 6301 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 5514 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 5869 of file worker.c.

{
    /* Attach to slot */
    logicalrep_worker_attach(worker_slot);
 
    Assert(am_tablesync_worker() || am_sequencesync_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,
                                  InvalidateSyncingRelStates,
                                  (Datum) 0);
}

References am_leader_apply_worker(), am_sequencesync_worker(), am_tablesync_worker(), Assert(), BackgroundWorkerUnblockSignals(), before_shmem_exit(), CacheRegisterSyscacheCallback(), Subscription::conninfo, DEBUG1, die, elog, GetCurrentTimestamp(), InitializeLogRepWorker(), InvalidateSyncingRelStates(), 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(), SequenceSyncWorkerMain(), and TableSyncWorkerMain().

◆ start_apply()

void start_apply ( XLogRecPtr origin_startpos )

Definition at line 5583 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,
                                             MyLogicalRepWorker->type);
 
            PG_RE_THROW();
        }
    }
    PG_END_TRY();
}

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

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 3939 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 5381 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 1687 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 1862 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 1651 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 469 of file worker.c.

Referenced by LogicalRepApplyLoop(), and ProcessParallelApplyMessage().

◆ ApplyContext

PGDLLIMPORT MemoryContext ApplyContext

extern

Definition at line 472 of file worker.c.

Referenced by copy_sequences(), InitializeLogRepWorker(), LogicalParallelApplyLoop(), LogicalRepApplyLoop(), LogicalRepSyncSequences(), maybe_reread_subscription(), pa_allocate_worker(), pa_launch_parallel_worker(), request_publisher_status(), 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 471 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 484 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 499 of file worker.c.

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

◆ LogRepWorkerWalRcvConn

PGDLLIMPORT struct WalReceiverConn* LogRepWorkerWalRcvConn

extern

Definition at line 477 of file worker.c.

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

◆ MyLogicalRepWorker

PGDLLIMPORT LogicalRepWorker* MyLogicalRepWorker

extern

Definition at line 56 of file launcher.c.

Referenced by am_leader_apply_worker(), am_parallel_apply_worker(), am_sequencesync_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(), FindDeletedTupleInLocalRel(), FinishSyncWorker(), get_candidate_xid(), InitializeLogRepWorker(), IsLogicalWorker(), launch_sync_worker(), logicalrep_reset_seqsync_start_time(), logicalrep_worker_attach(), logicalrep_worker_detach(), logicalrep_worker_onexit(), LogicalRepSyncSequences(), 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(), ProcessSequencesForSync(), ProcessSyncingRelations(), ProcessSyncingTablesForApply(), ProcessSyncingTablesForSync(), run_tablesync_worker(), set_stream_options(), SetupApplyOrSyncWorker(), should_apply_changes_for_rel(), start_apply(), stop_conflict_info_retention(), stream_abort_internal(), stream_cleanup_files(), stream_open_file(), stream_start_internal(), stream_stop_internal(), subxact_info_read(), subxact_info_write(), UpdateWorkerStats(), wait_for_local_flush(), wait_for_table_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 479 of file worker.c.

Referenced by adjust_xid_advance_interval(), apply_handle_commit_prepared(), apply_handle_delete(), apply_handle_insert(), apply_handle_prepare_internal(), apply_handle_rollback_prepared(), apply_handle_truncate(), apply_handle_update(), can_advance_nonremovable_xid(), clear_subscription_skip_lsn(), copy_sequence(), copy_sequences(), copy_table(), DisableSubscriptionAndExit(), fetch_remote_table_info(), FetchRelationStates(), FindDeletedTupleInLocalRel(), FinishSyncWorker(), InitializeLogRepWorker(), launch_sync_worker(), logicalrep_rel_open(), LogicalRepApplyLoop(), LogicalRepSyncSequences(), LogicalRepSyncTableStart(), maybe_reread_subscription(), maybe_start_skipping_changes(), pa_can_start(), pa_launch_parallel_worker(), pa_start_subtrans(), pa_stream_abort(), ParallelApplyWorkerMain(), ProcessParallelApplyInterrupts(), ProcessSyncingTablesForApply(), report_sequence_errors(), ReportApplyConflict(), resume_conflict_info_retention(), run_apply_worker(), run_tablesync_worker(), set_stream_options(), SetupApplyOrSyncWorker(), should_apply_changes_for_rel(), should_stop_conflict_info_retention(), start_apply(), start_sequence_sync(), start_table_sync(), stop_conflict_info_retention(), update_retention_status(), and wait_for_local_flush().

◆ table_states_not_ready

PGDLLIMPORT List* table_states_not_ready

extern

Definition at line 125 of file tablesync.c.

Referenced by AllTablesyncsReady(), FetchRelationStates(), and ProcessSyncingTablesForApply().

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

PGDLLIMPORT List *	table_states_not_ready

Data Structures

Macros

Typedefs

Enumerations

Functions

Variables

Macro Definition Documentation

◆ isParallelApplyWorker

◆ isSequenceSyncWorker

◆ 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_sequencesync_worker()

◆ am_tablesync_worker()

◆ apply_dispatch()

◆ apply_error_callback()

◆ apply_spooled_messages()

◆ DisableSubscriptionAndExit()

◆ FetchRelationStates()

◆ FinishSyncWorker()

◆ HasSubscriptionTablesCached()

◆ InitializeLogRepWorker()

◆ InvalidateSyncingRelStates()

◆ launch_sync_worker()

◆ logicalrep_pa_worker_stop()

◆ logicalrep_reset_seqsync_start_time()

◆ 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()

◆ ProcessSequencesForSync()

◆ ProcessSyncingRelations()

◆ ProcessSyncingTablesForApply()

◆ ProcessSyncingTablesForSync()

◆ 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