worker_internal.h File Reference

#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, SysCacheIdentifier 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_subsequences, 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)
static LogicalRepWorkerType	get_logical_worker_type (void)

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

Macro Definition Documentation

isParallelApplyWorker

#define isParallelApplyWorker

(

worker

)

Value:

                                       ((worker)->in_use && \
                                       (worker)->type == WORKERTYPE_PARALLEL_APPLY)

Definition at line 363 of file worker_internal.h.

{
    return isTableSyncWorker(MyLogicalRepWorker);
}
 
static inline bool
am_sequencesync_worker(void)
{
    return isSequenceSyncWorker(MyLogicalRepWorker);
}
 
static inline bool
am_leader_apply_worker(void)
{
    Assert(MyLogicalRepWorker->in_use);
    return (MyLogicalRepWorker->type == WORKERTYPE_APPLY);
}
 
static inline bool
am_parallel_apply_worker(void)
{
    Assert(MyLogicalRepWorker->in_use);
    return isParallelApplyWorker(MyLogicalRepWorker);
}
 
static inline LogicalRepWorkerType
get_logical_worker_type(void)
{
    Assert(MyLogicalRepWorker->in_use);
    return MyLogicalRepWorker->type;
}
 
#endif                          /* WORKER_INTERNAL_H */

isSequenceSyncWorker

#define isSequenceSyncWorker

(

worker

)

Value:

                                      ((worker)->in_use && \
                                      (worker)->type == WORKERTYPE_SEQUENCESYNC)

Definition at line 367 of file worker_internal.h.

isTableSyncWorker

#define isTableSyncWorker

(

worker

)

Value:

                                   ((worker)->in_use && \
                                   (worker)->type == WORKERTYPE_TABLESYNC)

Definition at line 365 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  )

extern

Definition at line 1597 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(), fb(), 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 383 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 390 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 377 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 371 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 )

extern

Definition at line 3781 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 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(), ERRCODE_PROTOCOL_VIOLATION, errmsg, ERROR, fb(), 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 )

extern

Definition at line 6221 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, ApplyErrorCallbackArg::command, errcontext, fb(), logicalrep_message_type(), LSN_FORMAT_ARGS, TransactionIdIsValid, and XLogRecPtrIsValid.

Referenced by LogicalParallelApplyLoop(), and LogicalRepApplyLoop().

apply_spooled_messages()

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

extern

Definition at line 2266 of file worker.c.

{
    int         nchanges;
    char        path[MAXPGPATH];
    char       *buffer = NULL;
    MemoryContext oldcxt;
    ResourceOwner oldowner;
    int         fileno;
    pgoff_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, fb(), 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  )

extern

Definition at line 6006 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);
 
    /* 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, and WARNING.

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

FetchRelationStates()

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

extern

Definition at line 203 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_object(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, fb(), foreach_ptr, get_rel_relkind(), GetSubscriptionRelations(), HasSubscriptionTables(), IsTransactionState(), lappend(), list_free_deep(), MemoryContextSwitchTo(), MySubscription, NIL, Subscription::oid, palloc_object, 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  )

extern

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

get_logical_worker_type()

static LogicalRepWorkerType get_logical_worker_type ( void  )

inlinestatic

Definition at line 397 of file worker_internal.h.

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

References Assert, LogicalRepWorker::in_use, MyLogicalRepWorker, and LogicalRepWorker::type.

Referenced by pgstat_report_subscription_error().

HasSubscriptionTablesCached()

bool HasSubscriptionTablesCached ( void  )

extern

Definition at line 1627 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(), fb(), FetchRelationStates(), and pgstat_report_stat().

Referenced by wait_for_local_flush().

InitializeLogRepWorker()

void InitializeLogRepWorker ( void  )

extern

Definition at line 5780 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, 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);
 
    /* Change wal_receiver_timeout according to the user's wishes */
    set_wal_receiver_timeout();
 
    /*
     * 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);
    /* Changes to foreign servers may affect subscriptions using SERVER. */
    CacheRegisterSyscacheCallback(FOREIGNSERVEROID,
                                  subscription_change_cb,
                                  (Datum) 0);
    /* Changes to user mappings may affect subscriptions using SERVER. */
    CacheRegisterSyscacheCallback(USERMAPPINGOID,
                                  subscription_change_cb,
                                  (Datum) 0);
 
    /*
     * Changes to FDW connection_function may affect subscriptions using
     * SERVER.
     */
    CacheRegisterSyscacheCallback(FOREIGNDATAWRAPPEROID,
                                  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();
 
    /*
     * Register a callback to reset the origin state before aborting any
     * pending transaction during shutdown (see ShutdownPostgres()). This will
     * avoid origin advancement for an incomplete 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.
     *
     * Register this callback here to ensure that all types of logical
     * replication workers that set up origins and apply remote transactions
     * are protected.
     */
    before_shmem_exit(on_exit_clear_xact_state, (Datum) 0);
}

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

Referenced by ParallelApplyWorkerMain(), and SetupApplyOrSyncWorker().

InvalidateSyncingRelStates()

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

extern

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  )

extern

Definition at line 118 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, fb(), 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 )

extern

Definition at line 680 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, fb(), 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  )

extern

Definition at line 873 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 fb(), 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 )

extern

Definition at line 928 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, fb(), 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 )

extern

Definition at line 758 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, fb(), 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  )

extern

Definition at line 259 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, fb(), 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  )

extern

Definition at line 325 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);
}

References Assert, BGW_MAXLEN, BGW_NEVER_RESTART, BGWORKER_BACKEND_DATABASE_CONNECTION, BGWORKER_SHMEM_ACCESS, BgWorkerStart_RecoveryFinished, ReplicationSlot::data, LogicalRepWorker::dbid, DEBUG1, DSM_HANDLE_INVALID, elog, ereport, errcode(), errhint(), errmsg, errmsg_internal(), ERROR, fb(), LogicalRepWorker::generation, GetCurrentTimestamp(), i, LogicalRepWorker::in_use, Int32GetDatum(), InvalidPid, InvalidTransactionId, InvalidXLogRecPtr, LogicalRepWorker::last_lsn, LogicalRepWorker::last_recv_time, LogicalRepWorker::last_send_time, LogicalRepWorker::last_seqsync_start_time, LogicalRepWorker::launch_time, LogicalRepWorker::leader_pid, logicalrep_pa_worker_count(), logicalrep_sync_worker_count(), logicalrep_worker_cleanup(), LogicalRepCtx, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), max_active_replication_origins, max_logical_replication_workers, max_parallel_apply_workers_per_subscription, max_sync_workers_per_subscription, MAXPGPATH, MyProcPid, MyReplicationSlot, now(), OidIsValid, LogicalRepWorker::oldest_nonremovable_xid, LogicalRepWorker::parallel_apply, LogicalRepWorker::proc, RegisterDynamicBackgroundWorker(), LogicalRepWorker::relid, LogicalRepWorker::relstate, LogicalRepWorker::relstate_lsn, LogicalRepWorker::reply_lsn, LogicalRepWorker::reply_time, snprintf, LogicalRepWorker::stream_fileset, LogicalRepWorker::subid, subname, TIMESTAMP_NOBEGIN, TimestampDifferenceExceeds(), LogicalRepWorker::type, LogicalRepWorker::userid, WaitForReplicationWorkerAttach(), wal_receiver_timeout, WARNING, LogicalRepCtxStruct::workers, WORKERTYPE_APPLY, WORKERTYPE_PARALLEL_APPLY, WORKERTYPE_SEQUENCESYNC, WORKERTYPE_TABLESYNC, WORKERTYPE_UNKNOWN, and ReplicationSlotPersistentData::xmin.

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

logicalrep_worker_stop()

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

extern

Definition at line 653 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, fb(), 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  )

extern

Definition at line 724 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, fb(), 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 )

extern

Definition at line 747 of file launcher.c.

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

References Assert, fb(), 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  )

extern

Definition at line 294 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, fb(), 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  )

extern

Definition at line 5044 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, 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);
 
    /* Change wal_receiver_timeout according to the user's wishes */
    set_wal_receiver_timeout();
 
    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, fb(), 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, set_wal_receiver_timeout(), 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 )

extern

Definition at line 474 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, fb(), 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  )

extern

Definition at line 1601 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  )

extern

Definition at line 626 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, fb(), lfirst, ParallelApplyWorkerPool, and shm_mq_detach().

Referenced by logicalrep_worker_detach().

pa_find_worker()

ParallelApplyWorkerInfo * pa_find_worker ( TransactionId  xid )

extern

Definition at line 522 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, fb(), 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  )

extern

Definition at line 1550 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  )

extern

Definition at line 1583 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  )

extern

Definition at line 1412 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  )

extern

Definition at line 1156 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, fb(), 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  )

extern

Definition at line 1508 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, fb(), FS_SERIALIZE_DONE, 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 )

extern

Definition at line 1344 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  )

extern

Definition at line 1317 of file applyparallelworker.c.

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

References fb(), SpinLockAcquire(), and SpinLockRelease().

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  )

extern

Definition at line 1372 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, fb(), 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 )

extern

Definition at line 1426 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_xact_state.origin_lsn = abort_data->abort_lsn;
    replorigin_xact_state.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 AbortCurrentTransaction(), AccessExclusiveLock, CommitTransactionCommand(), DEBUG1, elog, EndTransactionBlock(), fb(), i, IsTransactionBlock(), lfirst_xid, list_length(), list_nth_cell(), list_truncate(), MyParallelShared, MySubscription, NAMEDATALEN, Subscription::oid, ReplOriginXactState::origin_lsn, ReplOriginXactState::origin_timestamp, pa_reset_subtrans(), pa_savepoint_name(), pa_set_xact_state(), pa_unlock_transaction(), PARALLEL_TRANS_FINISHED, pgstat_report_activity(), replorigin_xact_state, RollbackToSavepoint(), STATE_IDLE, and subxactlist.

Referenced by apply_handle_stream_abort().

pa_switch_to_partial_serialize()

void pa_switch_to_partial_serialize ( ParallelApplyWorkerInfo *  winfo, bool  stream_locked  )

extern

Definition at line 1221 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, fb(), 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  )

extern

Definition at line 1557 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  )

extern

Definition at line 1590 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  )

extern

Definition at line 1628 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  )

extern

Definition at line 95 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(), fb(), 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 )

extern

Definition at line 156 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, fb(), 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 )

extern

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, fb(), FetchRelationStates(), get_rel_relkind(), HASH_BLOBS, hash_create(), hash_destroy(), HASH_ELEM, HASH_ENTER, hash_search(), IsTransactionState(), HASHCTL::keysize, 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(), proc_exit(), SubscriptionRelState::relid, 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 )

extern

Definition at line 246 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_xact_clear(true);
 
        /*
         * 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(), fb(), FinishSyncWorker(), 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_reset(), replorigin_xact_clear(), SpinLockAcquire(), SpinLockRelease(), StartTransactionCommand(), LogicalRepWorker::subid, UpdateSubscriptionRelState(), and walrcv_endstreaming.

Referenced by ProcessSyncingRelations().

ReplicationOriginNameForLogicalRep()

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

extern

Definition at line 647 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 fb(), 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 )

extern

Definition at line 6363 of file worker.c.

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

References apply_error_callback_arg, ApplyContext, fb(), 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  )

extern

Definition at line 5558 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, fb(), 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 )

extern

Definition at line 5946 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);
    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();
 
    /* 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(), CacheRegisterSyscacheCallback(), Subscription::conninfo, DEBUG1, elog, fb(), GetCurrentTimestamp(), InitializeLogRepWorker(), InvalidateSyncingRelStates(), LogicalRepWorker::last_recv_time, LogicalRepWorker::last_send_time, load_file(), logicalrep_worker_attach(), MyLogicalRepWorker, MySubscription, pqsignal, LogicalRepWorker::reply_time, SIGHUP, and SignalHandlerForConfigReload().

Referenced by ApplyWorkerMain(), SequenceSyncWorkerMain(), and TableSyncWorkerMain().

start_apply()

void start_apply ( XLogRecPtr  origin_startpos )

extern

Definition at line 5627 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_xact_clear(true);
 
        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);
 
            PG_RE_THROW();
        }
    }
    PG_END_TRY();
}

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

Referenced by run_apply_worker(), and run_tablesync_worker().

store_flush_position()

void store_flush_position ( XLogRecPtr  remote_lsn, XLogRecPtr  local_lsn  )

extern

Definition at line 3945 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 = palloc_object(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(), fb(), lsn_mapping, MemoryContextSwitchTo(), and palloc_object.

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  )

extern

Definition at line 5425 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  )

extern

Definition at line 1693 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_object(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(), fb(), FileSetInit(), MemoryContextSwitchTo(), MyLogicalRepWorker, palloc_object, 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 )

extern

Definition at line 1868 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  )

extern

Definition at line 1648 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, fb(), heap_freetuple(), heap_modify_tuple(), HeapTupleIsValid, ObjectIdGetDatum(), RelationGetDescr, RowExclusiveLock, SearchSysCacheCopy1, 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 473 of file worker.c.

Referenced by LogicalRepApplyLoop(), and ProcessParallelApplyMessage().

ApplyContext

PGDLLIMPORT MemoryContext ApplyContext

extern

Definition at line 476 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 475 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 488 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 503 of file worker.c.

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

LogRepWorkerWalRcvConn

PGDLLIMPORT struct WalReceiverConn* LogRepWorkerWalRcvConn

extern

Definition at line 481 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 57 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(), get_logical_worker_type(), 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(), 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 242 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 483 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(), set_wal_receiver_timeout(), 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 127 of file tablesync.c.

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