worker_internal.h File Reference

#include <signal.h>
#include "access/xlogdefs.h"
#include "catalog/pg_subscription.h"
#include "datatype/timestamp.h"
#include "miscadmin.h"
#include "replication/logicalrelation.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)   ((worker)->leader_pid != InvalidPid)

Typedefs
typedef struct LogicalRepWorker	LogicalRepWorker
typedef enum ParallelTransState	ParallelTransState
typedef enum PartialFileSetState	PartialFileSetState
typedef struct ParallelApplyWorkerShared	ParallelApplyWorkerShared
typedef struct ParallelApplyWorkerInfo	ParallelApplyWorkerInfo

Enumerations
enum	ParallelTransState { PARALLEL_TRANS_UNKNOWN , PARALLEL_TRANS_STARTED , PARALLEL_TRANS_FINISHED }
enum	PartialFileSetState { FS_EMPTY , FS_SERIALIZE_IN_PROGRESS , FS_SERIALIZE_DONE , FS_READY }

Functions
void	logicalrep_worker_attach (int slot)
LogicalRepWorker *	logicalrep_worker_find (Oid subid, Oid relid, bool only_running)
List *	logicalrep_workers_find (Oid subid, bool only_running)
bool	logicalrep_worker_launch (Oid dbid, Oid subid, const char *subname, Oid userid, Oid relid, dsm_handle subworker_dsm)
void	logicalrep_worker_stop (Oid subid, Oid relid)
void	logicalrep_pa_worker_stop (ParallelApplyWorkerInfo *winfo)
void	logicalrep_worker_wakeup (Oid subid, Oid relid)
void	logicalrep_worker_wakeup_ptr (LogicalRepWorker *worker)
int	logicalrep_sync_worker_count (Oid subid)
void	ReplicationOriginNameForLogicalRep (Oid suboid, Oid relid, char *originname, Size szoriginname)
char *	LogicalRepSyncTableStart (XLogRecPtr *origin_startpos)
bool	AllTablesyncsReady (void)
void	UpdateTwoPhaseState (Oid suboid, char new_state)
void	process_syncing_tables (XLogRecPtr current_lsn)
void	invalidate_syncing_table_states (Datum arg, int cacheid, uint32 hashvalue)
void	stream_start_internal (TransactionId xid, bool first_segment)
void	stream_stop_internal (TransactionId xid)
void	apply_spooled_messages (FileSet *stream_fileset, TransactionId xid, XLogRecPtr lsn)
void	apply_dispatch (StringInfo s)
void	maybe_reread_subscription (void)
void	stream_cleanup_files (Oid subid, TransactionId xid)
void	InitializeApplyWorker (void)
void	store_flush_position (XLogRecPtr remote_lsn, XLogRecPtr local_lsn)
void	apply_error_callback (void *arg)
void	set_apply_error_context_origin (char *originname)
void	pa_allocate_worker (TransactionId xid)
ParallelApplyWorkerInfo *	pa_find_worker (TransactionId xid)
void	pa_detach_all_error_mq (void)
bool	pa_send_data (ParallelApplyWorkerInfo *winfo, Size nbytes, const void *data)
void	pa_switch_to_partial_serialize (ParallelApplyWorkerInfo *winfo, bool stream_locked)
void	pa_set_xact_state (ParallelApplyWorkerShared *wshared, ParallelTransState xact_state)
void	pa_set_stream_apply_worker (ParallelApplyWorkerInfo *winfo)
void	pa_start_subtrans (TransactionId current_xid, TransactionId top_xid)
void	pa_reset_subtrans (void)
void	pa_stream_abort (LogicalRepStreamAbortData *abort_data)
void	pa_set_fileset_state (ParallelApplyWorkerShared *wshared, PartialFileSetState fileset_state)
void	pa_lock_stream (TransactionId xid, LOCKMODE lockmode)
void	pa_unlock_stream (TransactionId xid, LOCKMODE lockmode)
void	pa_lock_transaction (TransactionId xid, LOCKMODE lockmode)
void	pa_unlock_transaction (TransactionId xid, LOCKMODE lockmode)
void	pa_decr_and_wait_stream_block (void)
void	pa_xact_finish (ParallelApplyWorkerInfo *winfo, XLogRecPtr remote_lsn)
static bool	am_tablesync_worker (void)
static bool	am_leader_apply_worker (void)
static bool	am_parallel_apply_worker (void)

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

Macro Definition Documentation

isParallelApplyWorker

#define isParallelApplyWorker

(

worker

)

((worker)->leader_pid != InvalidPid)

Definition at line 308 of file worker_internal.h.

Typedef Documentation

LogicalRepWorker

typedef struct LogicalRepWorker LogicalRepWorker

ParallelApplyWorkerInfo

typedef struct ParallelApplyWorkerInfo ParallelApplyWorkerInfo

ParallelApplyWorkerShared

typedef struct ParallelApplyWorkerShared ParallelApplyWorkerShared

ParallelTransState

typedef enum ParallelTransState ParallelTransState

PartialFileSetState

typedef enum PartialFileSetState PartialFileSetState

Enumeration Type Documentation

ParallelTransState

enum ParallelTransState

Enumerator
PARALLEL_TRANS_UNKNOWN
PARALLEL_TRANS_STARTED
PARALLEL_TRANS_FINISHED

Definition at line 93 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 116 of file worker_internal.h.

{
    FS_EMPTY,
    FS_SERIALIZE_IN_PROGRESS,
    FS_SERIALIZE_DONE,
    FS_READY
} PartialFileSetState;

Function Documentation

AllTablesyncsReady()

bool AllTablesyncsReady

(

void

)

Definition at line 1580 of file tablesync.c.

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

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

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

am_leader_apply_worker()

static bool am_leader_apply_worker ( void  )

inlinestatic

Definition at line 317 of file worker_internal.h.

{
    return (!am_tablesync_worker() &&
            !isParallelApplyWorker(MyLogicalRepWorker));
}

References am_tablesync_worker(), isParallelApplyWorker, and MyLogicalRepWorker.

Referenced by logicalrep_worker_detach(), pa_can_start(), pa_set_fileset_state(), and pa_xact_finish().

am_parallel_apply_worker()

static bool am_parallel_apply_worker ( void  )

inlinestatic

Definition at line 324 of file worker_internal.h.

{
    return isParallelApplyWorker(MyLogicalRepWorker);
}

References isParallelApplyWorker, and MyLogicalRepWorker.

Referenced by apply_spooled_messages(), apply_worker_exit(), clear_subscription_skip_lsn(), DisableSubscriptionAndExit(), get_transaction_apply_action(), get_worker_name(), InitializeApplyWorker(), IsLogicalParallelApplyWorker(), maybe_reread_subscription(), pa_decr_and_wait_stream_block(), pa_free_worker(), pa_get_fileset_state(), process_syncing_tables(), should_apply_changes_for_rel(), and store_flush_position().

am_tablesync_worker()

static bool am_tablesync_worker ( void  )

inlinestatic

Definition at line 311 of file worker_internal.h.

{
    return OidIsValid(MyLogicalRepWorker->relid);
}

References MyLogicalRepWorker, OidIsValid, and LogicalRepWorker::relid.

Referenced by am_leader_apply_worker(), apply_handle_begin_prepare(), apply_handle_origin(), apply_handle_stream_prepare(), apply_worker_exit(), ApplyWorkerMain(), DisableSubscriptionAndExit(), get_worker_name(), InitializeApplyWorker(), maybe_reread_subscription(), process_syncing_tables(), should_apply_changes_for_rel(), start_apply(), and start_table_sync().

apply_dispatch()

void apply_dispatch

(

StringInfo

s

)

Definition at line 3285 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 \"%c\"", action)));
    }
 
    /* Reset the current command */
    apply_error_callback_arg.command = saved_command;
}

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

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

apply_error_callback()

void apply_error_callback

(

void *

arg

)

Definition at line 4916 of file worker.c.

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

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

Referenced by LogicalParallelApplyLoop(), and LogicalRepApplyLoop().

apply_spooled_messages()

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

Definition at line 2025 of file worker.c.

{
    StringInfoData s2;
    int         nchanges;
    char        path[MAXPGPATH];
    char       *buffer = NULL;
    MemoryContext oldcxt;
    ResourceOwner oldowner;
    int         fileno;
    off_t       offset;
 
    if (!am_parallel_apply_worker())
        maybe_start_skipping_changes(lsn);
 
    /* Make sure we have an open transaction */
    begin_replication_step();
 
    /*
     * Allocate file handle and memory required to process all the messages in
     * TopTransactionContext to avoid them getting reset after each message is
     * processed.
     */
    oldcxt = MemoryContextSwitchTo(TopTransactionContext);
 
    /* Open the spool file for the committed/prepared transaction */
    changes_filename(path, MyLogicalRepWorker->subid, xid);
    elog(DEBUG1, "replaying changes from file \"%s\"", path);
 
    /*
     * Make sure the file is owned by the toplevel transaction so that the
     * file will not be accidentally closed when aborting a subtransaction.
     */
    oldowner = CurrentResourceOwner;
    CurrentResourceOwner = TopTransactionResourceOwner;
 
    stream_fd = BufFileOpenFileSet(stream_fileset, path, O_RDONLY, false);
 
    CurrentResourceOwner = oldowner;
 
    buffer = palloc(BLCKSZ);
    initStringInfo(&s2);
 
    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)
    {
        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);
 
        /* copy the buffer to the stringinfo and call apply_dispatch */
        resetStringInfo(&s2);
        appendBinaryStringInfo(&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(), appendBinaryStringInfo(), apply_dispatch(), ApplyMessageContext, begin_replication_step(), BufFileOpenFileSet(), BufFileReadExact(), BufFileReadMaybeEOF(), BufFileTell(), changes_filename(), CHECK_FOR_INTERRUPTS, CurrentResourceOwner, DEBUG1, elog(), end_replication_step(), ensure_last_message(), ERROR, in_remote_transaction, initStringInfo(), len, MAXPGPATH, maybe_start_skipping_changes(), MemoryContextReset(), MemoryContextSwitchTo(), MyLogicalRepWorker, palloc(), pgstat_report_activity(), remote_final_lsn, repalloc(), resetStringInfo(), 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().

InitializeApplyWorker()

void InitializeApplyWorker

(

void

)

Definition at line 4445 of file worker.c.

{
    MemoryContext oldctx;
 
    /* Run as replica session replication role. */
    SetConfigOption("session_replication_role", "replica",
                    PGC_SUSET, PGC_S_OVERRIDE);
 
    /* Connect to our database. */
    BackgroundWorkerInitializeConnectionByOid(MyLogicalRepWorker->dbid,
                                              MyLogicalRepWorker->userid,
                                              0);
 
    /*
     * Set always-secure search path, so malicious users can't redirect user
     * code (e.g. pg_index.indexprs).
     */
    SetConfigOption("search_path", "", PGC_SUSET, PGC_S_OVERRIDE);
 
    /* Load the subscription into persistent memory context. */
    ApplyContext = AllocSetContextCreate(TopMemoryContext,
                                         "ApplyContext",
                                         ALLOCSET_DEFAULT_SIZES);
    StartTransactionCommand();
    oldctx = MemoryContextSwitchTo(ApplyContext);
 
    MySubscription = GetSubscription(MyLogicalRepWorker->subid, true);
    if (!MySubscription)
    {
        ereport(LOG,
        /* translator: %s is the name of logical replication worker */
                (errmsg("%s for subscription %u will not start because the subscription was removed during startup",
                        get_worker_name(), MyLogicalRepWorker->subid)));
 
        /* Ensure we remove no-longer-useful entry for worker's start time */
        if (!am_tablesync_worker() && !am_parallel_apply_worker())
            ApplyLauncherForgetWorkerStartTime(MyLogicalRepWorker->subid);
        proc_exit(0);
    }
 
    MySubscriptionValid = true;
    MemoryContextSwitchTo(oldctx);
 
    if (!MySubscription->enabled)
    {
        ereport(LOG,
        /* translator: first %s is the name of logical replication worker */
                (errmsg("%s for subscription \"%s\" will not start because the subscription was disabled during startup",
                        get_worker_name(), MySubscription->name)));
 
        apply_worker_exit();
    }
 
    /* Setup synchronous commit according to the user's wishes */
    SetConfigOption("synchronous_commit", MySubscription->synccommit,
                    PGC_BACKEND, PGC_S_OVERRIDE);
 
    /* Keep us informed about subscription changes. */
    CacheRegisterSyscacheCallback(SUBSCRIPTIONOID,
                                  subscription_change_cb,
                                  (Datum) 0);
 
    if (am_tablesync_worker())
        ereport(LOG,
                (errmsg("logical replication table synchronization worker for subscription \"%s\", table \"%s\" has started",
                        MySubscription->name,
                        get_rel_name(MyLogicalRepWorker->relid))));
    else
        ereport(LOG,
        /* translator: first %s is the name of logical replication worker */
                (errmsg("%s for subscription \"%s\" has started",
                        get_worker_name(), MySubscription->name)));
 
    CommitTransactionCommand();
}

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

Referenced by ApplyWorkerMain(), and ParallelApplyWorkerMain().

invalidate_syncing_table_states()

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

Definition at line 272 of file tablesync.c.

{
    table_states_valid = false;
}

References table_states_valid.

Referenced by ApplyWorkerMain(), and ParallelApplyWorkerMain().

logicalrep_pa_worker_stop()

void logicalrep_pa_worker_stop

(

ParallelApplyWorkerInfo *

winfo

)

Definition at line 618 of file launcher.c.

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

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

Referenced by pa_free_worker().

logicalrep_sync_worker_count()

int logicalrep_sync_worker_count

(

Oid

subid

)

Definition at line 832 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 && OidIsValid(w->relid))
            res++;
    }
 
    return res;
}

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

Referenced by logicalrep_worker_launch(), and process_syncing_tables_for_apply().

logicalrep_worker_attach()

void logicalrep_worker_attach

(

int

slot

)

Definition at line 692 of file launcher.c.

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

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

Referenced by ApplyWorkerMain(), and ParallelApplyWorkerMain().

logicalrep_worker_find()

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

Definition at line 249 of file launcher.c.

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

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

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

logicalrep_worker_launch()

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

Definition at line 306 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_parallel_apply_worker = (subworker_dsm != DSM_HANDLE_INVALID);
 
    /* Sanity check - tablesync worker cannot be a subworker */
    Assert(!(is_parallel_apply_worker && OidIsValid(relid)));
 
    ereport(DEBUG1,
            (errmsg_internal("starting logical replication worker for subscription \"%s\"",
                             subname)));
 
    /* Report this after the initial starting message for consistency. */
    if (max_replication_slots == 0)
        ereport(ERROR,
                (errcode(ERRCODE_CONFIGURATION_LIMIT_EXCEEDED),
                 errmsg("cannot start logical replication workers when max_replication_slots = 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 (OidIsValid(relid) && 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 max_logical_replication_workers.")));
        return false;
    }
 
    /* Prepare the worker slot. */
    worker->launch_time = now;
    worker->in_use = true;
    worker->generation++;
    worker->proc = NULL;
    worker->dbid = dbid;
    worker->userid = userid;
    worker->subid = subid;
    worker->relid = relid;
    worker->relstate = SUBREL_STATE_UNKNOWN;
    worker->relstate_lsn = InvalidXLogRecPtr;
    worker->stream_fileset = NULL;
    worker->leader_pid = is_parallel_apply_worker ? MyProcPid : InvalidPid;
    worker->parallel_apply = is_parallel_apply_worker;
    worker->last_lsn = InvalidXLogRecPtr;
    TIMESTAMP_NOBEGIN(worker->last_send_time);
    TIMESTAMP_NOBEGIN(worker->last_recv_time);
    worker->reply_lsn = InvalidXLogRecPtr;
    TIMESTAMP_NOBEGIN(worker->reply_time);
 
    /* Before releasing lock, remember generation for future identification. */
    generation = worker->generation;
 
    LWLockRelease(LogicalRepWorkerLock);
 
    /* Register the new dynamic worker. */
    memset(&bgw, 0, sizeof(bgw));
    bgw.bgw_flags = BGWORKER_SHMEM_ACCESS |
        BGWORKER_BACKEND_DATABASE_CONNECTION;
    bgw.bgw_start_time = BgWorkerStart_RecoveryFinished;
    snprintf(bgw.bgw_library_name, BGW_MAXLEN, "postgres");
 
    if (is_parallel_apply_worker)
        snprintf(bgw.bgw_function_name, BGW_MAXLEN, "ParallelApplyWorkerMain");
    else
        snprintf(bgw.bgw_function_name, BGW_MAXLEN, "ApplyWorkerMain");
 
    if (OidIsValid(relid))
        snprintf(bgw.bgw_name, BGW_MAXLEN,
                 "logical replication worker for subscription %u sync %u", subid, relid);
    else if (is_parallel_apply_worker)
        snprintf(bgw.bgw_name, BGW_MAXLEN,
                 "logical replication parallel apply worker for subscription %u", subid);
    else
        snprintf(bgw.bgw_name, BGW_MAXLEN,
                 "logical replication apply worker for subscription %u", subid);
 
    if (is_parallel_apply_worker)
        snprintf(bgw.bgw_type, BGW_MAXLEN, "logical replication parallel worker");
    else
        snprintf(bgw.bgw_type, BGW_MAXLEN, "logical replication worker");
 
    bgw.bgw_restart_time = BGW_NEVER_RESTART;
    bgw.bgw_notify_pid = MyProcPid;
    bgw.bgw_main_arg = Int32GetDatum(slot);
 
    if (is_parallel_apply_worker)
        memcpy(bgw.bgw_extra, &subworker_dsm, sizeof(dsm_handle));
 
    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 max_worker_processes.")));
        return false;
    }
 
    /* Now wait until it attaches. */
    return WaitForReplicationWorkerAttach(worker, generation, bgw_handle);
}

References Assert(), BackgroundWorker::bgw_extra, BackgroundWorker::bgw_flags, BackgroundWorker::bgw_function_name, BackgroundWorker::bgw_library_name, BackgroundWorker::bgw_main_arg, BGW_MAXLEN, BackgroundWorker::bgw_name, BGW_NEVER_RESTART, BackgroundWorker::bgw_notify_pid, BackgroundWorker::bgw_restart_time, BackgroundWorker::bgw_start_time, BackgroundWorker::bgw_type, BGWORKER_BACKEND_DATABASE_CONNECTION, BGWORKER_SHMEM_ACCESS, BgWorkerStart_RecoveryFinished, LogicalRepWorker::dbid, DEBUG1, DSM_HANDLE_INVALID, elog(), ereport, errcode(), errhint(), errmsg(), errmsg_internal(), ERROR, LogicalRepWorker::generation, GetCurrentTimestamp(), i, LogicalRepWorker::in_use, Int32GetDatum(), InvalidPid, InvalidXLogRecPtr, LogicalRepWorker::last_lsn, LogicalRepWorker::last_recv_time, LogicalRepWorker::last_send_time, LogicalRepWorker::launch_time, LogicalRepWorker::leader_pid, logicalrep_pa_worker_count(), logicalrep_sync_worker_count(), logicalrep_worker_cleanup(), LogicalRepCtx, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), max_logical_replication_workers, max_parallel_apply_workers_per_subscription, max_replication_slots, max_sync_workers_per_subscription, MyProcPid, now(), OidIsValid, 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::userid, WaitForReplicationWorkerAttach(), wal_receiver_timeout, WARNING, and LogicalRepCtxStruct::workers.

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

logicalrep_worker_stop()

void logicalrep_worker_stop	(	Oid	subid,
		Oid	relid
	)

Definition at line 594 of file launcher.c.

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

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

Referenced by AlterSubscription_refresh(), and DropSubscription().

logicalrep_worker_wakeup()

void logicalrep_worker_wakeup	(	Oid	subid,
		Oid	relid
	)

Definition at line 661 of file launcher.c.

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

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

Referenced by apply_handle_stream_start(), and pg_attribute_noreturn().

logicalrep_worker_wakeup_ptr()

void logicalrep_worker_wakeup_ptr

(

LogicalRepWorker *

worker

)

Definition at line 681 of file launcher.c.

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

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

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

logicalrep_workers_find()

List* logicalrep_workers_find	(	Oid	subid,
		bool	only_running
	)

Definition at line 281 of file launcher.c.

{
    int         i;
    List       *res = NIL;
 
    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);
    }
 
    return res;
}

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

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

LogicalRepSyncTableStart()

char* LogicalRepSyncTableStart

(

XLogRecPtr *

origin_startpos

)

Definition at line 1244 of file tablesync.c.

{
    char       *slotname;
    char       *err;
    char        relstate;
    XLogRecPtr  relstate_lsn;
    Relation    rel;
    AclResult   aclresult;
    WalRcvExecResult *res;
    char        originname[NAMEDATALEN];
    RepOriginId originid;
    bool        must_use_password;
 
    /* Check the state of the table synchronization. */
    StartTransactionCommand();
    relstate = GetSubscriptionRelState(MyLogicalRepWorker->subid,
                                       MyLogicalRepWorker->relid,
                                       &relstate_lsn);
    CommitTransactionCommand();
 
    SpinLockAcquire(&MyLogicalRepWorker->relmutex);
    MyLogicalRepWorker->relstate = relstate;
    MyLogicalRepWorker->relstate_lsn = relstate_lsn;
    SpinLockRelease(&MyLogicalRepWorker->relmutex);
 
    /*
     * If synchronization is already done or no longer necessary, exit now
     * that we've updated shared memory state.
     */
    switch (relstate)
    {
        case SUBREL_STATE_SYNCDONE:
        case SUBREL_STATE_READY:
        case SUBREL_STATE_UNKNOWN:
            finish_sync_worker();   /* doesn't return */
    }
 
    /* Calculate the name of the tablesync slot. */
    slotname = (char *) palloc(NAMEDATALEN);
    ReplicationSlotNameForTablesync(MySubscription->oid,
                                    MyLogicalRepWorker->relid,
                                    slotname,
                                    NAMEDATALEN);
 
    /* Is the use of a password mandatory? */
    must_use_password = MySubscription->passwordrequired &&
        !superuser_arg(MySubscription->owner);
 
    /*
     * Here we use the slot name instead of the subscription name as the
     * application_name, so that it is different from the leader apply worker,
     * so that synchronous replication can distinguish them.
     */
    LogRepWorkerWalRcvConn =
        walrcv_connect(MySubscription->conninfo, true,
                       must_use_password,
                       slotname, &err);
    if (LogRepWorkerWalRcvConn == NULL)
        ereport(ERROR,
                (errcode(ERRCODE_CONNECTION_FAILURE),
                 errmsg("could not connect to the publisher: %s", err)));
 
    Assert(MyLogicalRepWorker->relstate == SUBREL_STATE_INIT ||
           MyLogicalRepWorker->relstate == SUBREL_STATE_DATASYNC ||
           MyLogicalRepWorker->relstate == SUBREL_STATE_FINISHEDCOPY);
 
    /* Assign the origin tracking record name. */
    ReplicationOriginNameForLogicalRep(MySubscription->oid,
                                       MyLogicalRepWorker->relid,
                                       originname,
                                       sizeof(originname));
 
    if (MyLogicalRepWorker->relstate == SUBREL_STATE_DATASYNC)
    {
        /*
         * We have previously errored out before finishing the copy so the
         * replication slot might exist. We want to remove the slot if it
         * already exists and proceed.
         *
         * XXX We could also instead try to drop the slot, last time we failed
         * but for that, we might need to clean up the copy state as it might
         * be in the middle of fetching the rows. Also, if there is a network
         * breakdown then it wouldn't have succeeded so trying it next time
         * seems like a better bet.
         */
        ReplicationSlotDropAtPubNode(LogRepWorkerWalRcvConn, slotname, true);
    }
    else if (MyLogicalRepWorker->relstate == SUBREL_STATE_FINISHEDCOPY)
    {
        /*
         * The COPY phase was previously done, but tablesync then crashed
         * before it was able to finish normally.
         */
        StartTransactionCommand();
 
        /*
         * The origin tracking name must already exist. It was created first
         * time this tablesync was launched.
         */
        originid = replorigin_by_name(originname, false);
        replorigin_session_setup(originid, 0);
        replorigin_session_origin = originid;
        *origin_startpos = replorigin_session_get_progress(false);
 
        CommitTransactionCommand();
 
        goto copy_table_done;
    }
 
    SpinLockAcquire(&MyLogicalRepWorker->relmutex);
    MyLogicalRepWorker->relstate = SUBREL_STATE_DATASYNC;
    MyLogicalRepWorker->relstate_lsn = InvalidXLogRecPtr;
    SpinLockRelease(&MyLogicalRepWorker->relmutex);
 
    /* Update the state and make it visible to others. */
    StartTransactionCommand();
    UpdateSubscriptionRelState(MyLogicalRepWorker->subid,
                               MyLogicalRepWorker->relid,
                               MyLogicalRepWorker->relstate,
                               MyLogicalRepWorker->relstate_lsn);
    CommitTransactionCommand();
    pgstat_report_stat(true);
 
    StartTransactionCommand();
 
    /*
     * Use a standard write lock here. It might be better to disallow access
     * to the table while it's being synchronized. But we don't want to block
     * the main apply process from working and it has to open the relation in
     * RowExclusiveLock when remapping remote relation id to local one.
     */
    rel = table_open(MyLogicalRepWorker->relid, RowExclusiveLock);
 
    /*
     * Check that our table sync worker has permission to insert into the
     * target table.
     */
    aclresult = pg_class_aclcheck(RelationGetRelid(rel), GetUserId(),
                                  ACL_INSERT);
    if (aclresult != ACLCHECK_OK)
        aclcheck_error(aclresult,
                       get_relkind_objtype(rel->rd_rel->relkind),
                       RelationGetRelationName(rel));
 
    /*
     * COPY FROM does not honor RLS policies.  That is not a problem for
     * subscriptions owned by roles with BYPASSRLS privilege (or superuser,
     * who has it implicitly), but other roles should not be able to
     * circumvent RLS.  Disallow logical replication into RLS enabled
     * relations for such roles.
     */
    if (check_enable_rls(RelationGetRelid(rel), InvalidOid, false) == RLS_ENABLED)
        ereport(ERROR,
                (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
                 errmsg("user \"%s\" cannot replicate into relation with row-level security enabled: \"%s\"",
                        GetUserNameFromId(GetUserId(), true),
                        RelationGetRelationName(rel))));
 
    /*
     * Start a transaction in the remote node in REPEATABLE READ mode.  This
     * ensures that both the replication slot we create (see below) and the
     * COPY are consistent with each other.
     */
    res = walrcv_exec(LogRepWorkerWalRcvConn,
                      "BEGIN READ ONLY ISOLATION LEVEL REPEATABLE READ",
                      0, NULL);
    if (res->status != WALRCV_OK_COMMAND)
        ereport(ERROR,
                (errcode(ERRCODE_CONNECTION_FAILURE),
                 errmsg("table copy could not start transaction on publisher: %s",
                        res->err)));
    walrcv_clear_result(res);
 
    /*
     * Create a new permanent logical decoding slot. This slot will be used
     * for the catchup phase after COPY is done, so tell it to use the
     * snapshot to make the final data consistent.
     */
    walrcv_create_slot(LogRepWorkerWalRcvConn,
                       slotname, false /* permanent */ , false /* two_phase */ ,
                       CRS_USE_SNAPSHOT, origin_startpos);
 
    /*
     * Setup replication origin tracking. The purpose of doing this before the
     * copy is to avoid doing the copy again due to any error in setting up
     * origin tracking.
     */
    originid = replorigin_by_name(originname, true);
    if (!OidIsValid(originid))
    {
        /*
         * Origin tracking does not exist, so create it now.
         *
         * Then advance to the LSN got from walrcv_create_slot. This is WAL
         * logged for the purpose of recovery. Locks are to prevent the
         * replication origin from vanishing while advancing.
         */
        originid = replorigin_create(originname);
 
        LockRelationOid(ReplicationOriginRelationId, RowExclusiveLock);
        replorigin_advance(originid, *origin_startpos, InvalidXLogRecPtr,
                           true /* go backward */ , true /* WAL log */ );
        UnlockRelationOid(ReplicationOriginRelationId, RowExclusiveLock);
 
        replorigin_session_setup(originid, 0);
        replorigin_session_origin = originid;
    }
    else
    {
        ereport(ERROR,
                (errcode(ERRCODE_DUPLICATE_OBJECT),
                 errmsg("replication origin \"%s\" already exists",
                        originname)));
    }
 
    /* Now do the initial data copy */
    PushActiveSnapshot(GetTransactionSnapshot());
    copy_table(rel);
    PopActiveSnapshot();
 
    res = walrcv_exec(LogRepWorkerWalRcvConn, "COMMIT", 0, NULL);
    if (res->status != WALRCV_OK_COMMAND)
        ereport(ERROR,
                (errcode(ERRCODE_CONNECTION_FAILURE),
                 errmsg("table copy could not finish transaction on publisher: %s",
                        res->err)));
    walrcv_clear_result(res);
 
    table_close(rel, NoLock);
 
    /* Make the copy visible. */
    CommandCounterIncrement();
 
    /*
     * Update the persisted state to indicate the COPY phase is done; make it
     * visible to others.
     */
    UpdateSubscriptionRelState(MyLogicalRepWorker->subid,
                               MyLogicalRepWorker->relid,
                               SUBREL_STATE_FINISHEDCOPY,
                               MyLogicalRepWorker->relstate_lsn);
 
    CommitTransactionCommand();
 
copy_table_done:
 
    elog(DEBUG1,
         "LogicalRepSyncTableStart: '%s' origin_startpos lsn %X/%X",
         originname, LSN_FORMAT_ARGS(*origin_startpos));
 
    /*
     * We are done with the initial data synchronization, update the state.
     */
    SpinLockAcquire(&MyLogicalRepWorker->relmutex);
    MyLogicalRepWorker->relstate = SUBREL_STATE_SYNCWAIT;
    MyLogicalRepWorker->relstate_lsn = *origin_startpos;
    SpinLockRelease(&MyLogicalRepWorker->relmutex);
 
    /*
     * Finally, wait until the leader apply worker tells us to catch up and
     * then return to let LogicalRepApplyLoop do it.
     */
    wait_for_worker_state_change(SUBREL_STATE_CATCHUP);
    return slotname;
}

References ACL_INSERT, aclcheck_error(), ACLCHECK_OK, Assert(), check_enable_rls(), CommandCounterIncrement(), CommitTransactionCommand(), Subscription::conninfo, copy_table(), CRS_USE_SNAPSHOT, DEBUG1, elog(), ereport, err(), errcode(), ERRCODE_DUPLICATE_OBJECT, errmsg(), ERROR, get_relkind_objtype(), GetSubscriptionRelState(), GetTransactionSnapshot(), GetUserId(), GetUserNameFromId(), InvalidOid, InvalidXLogRecPtr, LockRelationOid(), LogRepWorkerWalRcvConn, LSN_FORMAT_ARGS, MyLogicalRepWorker, MySubscription, NAMEDATALEN, NoLock, Subscription::oid, OidIsValid, Subscription::owner, palloc(), Subscription::passwordrequired, pg_class_aclcheck(), pgstat_report_stat(), PopActiveSnapshot(), PushActiveSnapshot(), RelationData::rd_rel, RelationGetRelationName, RelationGetRelid, LogicalRepWorker::relid, LogicalRepWorker::relmutex, LogicalRepWorker::relstate, LogicalRepWorker::relstate_lsn, ReplicationOriginNameForLogicalRep(), ReplicationSlotDropAtPubNode(), ReplicationSlotNameForTablesync(), replorigin_advance(), replorigin_by_name(), replorigin_create(), replorigin_session_get_progress(), replorigin_session_origin, replorigin_session_setup(), res, RLS_ENABLED, RowExclusiveLock, SpinLockAcquire, SpinLockRelease, StartTransactionCommand(), LogicalRepWorker::subid, superuser_arg(), table_close(), table_open(), UnlockRelationOid(), UpdateSubscriptionRelState(), wait_for_worker_state_change(), walrcv_clear_result(), walrcv_connect, walrcv_create_slot, walrcv_exec, and WALRCV_OK_COMMAND.

Referenced by start_table_sync().

maybe_reread_subscription()

void maybe_reread_subscription

(

void

)

Definition at line 3872 of file worker.c.

{
    MemoryContext oldctx;
    Subscription *newsub;
    bool        started_tx = false;
 
    /* When cache state is valid there is nothing to do here. */
    if (MySubscriptionValid)
        return;
 
    /* This function might be called inside or outside of transaction. */
    if (!IsTransactionState())
    {
        StartTransactionCommand();
        started_tx = true;
    }
 
    /* Ensure allocations in permanent context. */
    oldctx = MemoryContextSwitchTo(ApplyContext);
 
    newsub = GetSubscription(MyLogicalRepWorker->subid, true);
 
    /*
     * Exit if the subscription was removed. This normally should not happen
     * as the worker gets killed during DROP SUBSCRIPTION.
     */
    if (!newsub)
    {
        ereport(LOG,
        /* translator: first %s is the name of logical replication worker */
                (errmsg("%s for subscription \"%s\" will stop because the subscription was removed",
                        get_worker_name(), MySubscription->name)));
 
        /* Ensure we remove no-longer-useful entry for worker's start time */
        if (!am_tablesync_worker() && !am_parallel_apply_worker())
            ApplyLauncherForgetWorkerStartTime(MyLogicalRepWorker->subid);
        proc_exit(0);
    }
 
    /* Exit if the subscription was disabled. */
    if (!newsub->enabled)
    {
        ereport(LOG,
        /* translator: first %s is the name of logical replication worker */
                (errmsg("%s for subscription \"%s\" will stop because the subscription was disabled",
                        get_worker_name(), MySubscription->name)));
 
        apply_worker_exit();
    }
 
    /* !slotname should never happen when enabled is true. */
    Assert(newsub->slotname);
 
    /* two-phase should not be altered */
    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,
            /* translator: first %s is the name of logical replication worker */
                    (errmsg("%s for subscription \"%s\" will restart because of a parameter change",
                            get_worker_name(), MySubscription->name)));
 
        apply_worker_exit();
    }
 
    /* Check for other changes that should never happen too. */
    if (newsub->dbid != MySubscription->dbid)
    {
        elog(ERROR, "subscription %u changed unexpectedly",
             MyLogicalRepWorker->subid);
    }
 
    /* Clean old subscription info and switch to new one. */
    FreeSubscription(MySubscription);
    MySubscription = newsub;
 
    MemoryContextSwitchTo(oldctx);
 
    /* Change synchronous commit according to the user's wishes */
    SetConfigOption("synchronous_commit", MySubscription->synccommit,
                    PGC_BACKEND, PGC_S_OVERRIDE);
 
    if (started_tx)
        CommitTransactionCommand();
 
    MySubscriptionValid = true;
}

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

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

pa_allocate_worker()

void pa_allocate_worker

(

TransactionId

xid

)

Definition at line 469 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;
    entry->xid = xid;
}

References ApplyContext, elog(), HASHCTL::entrysize, ERROR, HASH_BLOBS, HASH_CONTEXT, hash_create(), HASH_ELEM, HASH_ENTER, hash_search(), HASHCTL::hcxt, ParallelApplyWorkerInfo::in_use, HASHCTL::keysize, 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, ParallelApplyWorkerEntry::xid, and ParallelApplyWorkerShared::xid.

Referenced by apply_handle_stream_start().

pa_decr_and_wait_stream_block()

void pa_decr_and_wait_stream_block

(

void

)

Definition at line 1594 of file applyparallelworker.c.

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

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

Referenced by apply_handle_stream_abort(), and apply_handle_stream_stop().

pa_detach_all_error_mq()

void pa_detach_all_error_mq

(

void

)

Definition at line 622 of file applyparallelworker.c.

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

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

Referenced by logicalrep_worker_detach().

pa_find_worker()

ParallelApplyWorkerInfo* pa_find_worker

(

TransactionId

xid

)

Definition at line 518 of file applyparallelworker.c.

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

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

Referenced by get_transaction_apply_action().

pa_lock_stream()

void pa_lock_stream	(	TransactionId	xid,
		LOCKMODE	lockmode
	)

Definition at line 1543 of file applyparallelworker.c.

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

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

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

pa_lock_transaction()

void pa_lock_transaction	(	TransactionId	xid,
		LOCKMODE	lockmode
	)

Definition at line 1576 of file applyparallelworker.c.

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

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

Referenced by apply_handle_stream_start(), and pa_wait_for_xact_finish().

pa_reset_subtrans()

void pa_reset_subtrans

(

void

)

Definition at line 1405 of file applyparallelworker.c.

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

References NIL, and subxactlist.

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

pa_send_data()

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

Definition at line 1149 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(logical_replication_mode == LOGICAL_REP_MODE_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, ereport, errcode(), errmsg(), ERROR, GetCurrentTimestamp(), IsTransactionState(), LOGICAL_REP_MODE_IMMEDIATE, logical_replication_mode, 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, WAIT_EVENT_LOGICAL_APPLY_SEND_DATA, WaitLatch(), WL_EXIT_ON_PM_DEATH, WL_LATCH_SET, and WL_TIMEOUT.

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

pa_set_fileset_state()

void pa_set_fileset_state	(	ParallelApplyWorkerShared *	wshared,
		PartialFileSetState	fileset_state
	)

Definition at line 1501 of file applyparallelworker.c.

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

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

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

pa_set_stream_apply_worker()

void pa_set_stream_apply_worker

(

ParallelApplyWorkerInfo *

winfo

)

Definition at line 1337 of file applyparallelworker.c.

{
    stream_apply_worker = winfo;
}

References stream_apply_worker.

Referenced by apply_handle_stream_start(), and apply_handle_stream_stop().

pa_set_xact_state()

void pa_set_xact_state	(	ParallelApplyWorkerShared *	wshared,
		ParallelTransState	xact_state
	)

Definition at line 1310 of file applyparallelworker.c.

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

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

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

pa_start_subtrans()

void pa_start_subtrans	(	TransactionId	current_xid,
		TransactionId	top_xid
	)

Definition at line 1365 of file applyparallelworker.c.

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

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

Referenced by handle_streamed_transaction().

pa_stream_abort()

void pa_stream_abort

(

LogicalRepStreamAbortData *

abort_data

)

Definition at line 1419 of file applyparallelworker.c.

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

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

Referenced by apply_handle_stream_abort().

pa_switch_to_partial_serialize()

void pa_switch_to_partial_serialize	(	ParallelApplyWorkerInfo *	winfo,
		bool	stream_locked
	)

Definition at line 1214 of file applyparallelworker.c.

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

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

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

pa_unlock_stream()

void pa_unlock_stream	(	TransactionId	xid,
		LOCKMODE	lockmode
	)

Definition at line 1550 of file applyparallelworker.c.

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

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

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

pa_unlock_transaction()

void pa_unlock_transaction	(	TransactionId	xid,
		LOCKMODE	lockmode
	)

Definition at line 1583 of file applyparallelworker.c.

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

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

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

pa_xact_finish()

void pa_xact_finish	(	ParallelApplyWorkerInfo *	winfo,
		XLogRecPtr	remote_lsn
	)

Definition at line 1621 of file applyparallelworker.c.

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

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

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

process_syncing_tables()

void process_syncing_tables

(

XLogRecPtr

current_lsn

)

Definition at line 647 of file tablesync.c.

{
    /*
     * 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().
     */
    if (am_parallel_apply_worker())
        return;
 
    if (am_tablesync_worker())
        process_syncing_tables_for_sync(current_lsn);
    else
        process_syncing_tables_for_apply(current_lsn);
}

References am_parallel_apply_worker(), am_tablesync_worker(), process_syncing_tables_for_apply(), and process_syncing_tables_for_sync().

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

ReplicationOriginNameForLogicalRep()

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

Definition at line 461 of file worker.c.

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

References OidIsValid, and snprintf.

Referenced by AlterSubscription(), AlterSubscription_refresh(), ApplyWorkerMain(), CreateSubscription(), DropSubscription(), LogicalRepSyncTableStart(), ParallelApplyWorkerMain(), process_syncing_tables_for_apply(), and process_syncing_tables_for_sync().

set_apply_error_context_origin()

void set_apply_error_context_origin

(

char *

originname

)

Definition at line 5058 of file worker.c.

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

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

Referenced by ApplyWorkerMain(), and ParallelApplyWorkerMain().

store_flush_position()

void store_flush_position	(	XLogRecPtr	remote_lsn,
		XLogRecPtr	local_lsn
	)

Definition at line 3449 of file worker.c.

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

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

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

stream_cleanup_files()

void stream_cleanup_files	(	Oid	subid,
		TransactionId	xid
	)

Definition at line 4199 of file worker.c.

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

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

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

stream_start_internal()

void stream_start_internal	(	TransactionId	xid,
		bool	first_segment
	)

Definition at line 1453 of file worker.c.

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

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

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

stream_stop_internal()

void stream_stop_internal

(

TransactionId

xid

)

Definition at line 1627 of file worker.c.

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

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

Referenced by apply_handle_stream_stop(), and stream_open_and_write_change().

UpdateTwoPhaseState()

void UpdateTwoPhaseState	(	Oid	suboid,
		char	new_state
	)

Definition at line 1605 of file tablesync.c.

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

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

Referenced by ApplyWorkerMain(), and CreateSubscription().

Variable Documentation

apply_error_context_stack

PGDLLIMPORT ErrorContextCallback* apply_error_context_stack

extern

Definition at line 305 of file worker.c.

Referenced by HandleParallelApplyMessage(), and LogicalRepApplyLoop().

ApplyContext

PGDLLIMPORT MemoryContext ApplyContext

extern

Definition at line 308 of file worker.c.

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

ApplyMessageContext

PGDLLIMPORT MemoryContext ApplyMessageContext

extern

Definition at line 307 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 320 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 335 of file worker.c.

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

LogRepWorkerWalRcvConn

PGDLLIMPORT struct WalReceiverConn* LogRepWorkerWalRcvConn

extern

Definition at line 313 of file worker.c.

Referenced by ApplyWorkerMain(), copy_read_data(), copy_table(), fetch_remote_table_info(), logicalrep_worker_onexit(), LogicalRepApplyLoop(), LogicalRepSyncTableStart(), process_syncing_tables_for_sync(), and send_feedback().

MyLogicalRepWorker

PGDLLIMPORT LogicalRepWorker* MyLogicalRepWorker

extern

Definition at line 61 of file launcher.c.

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

MyParallelShared

PGDLLIMPORT ParallelApplyWorkerShared* MyParallelShared

extern

Definition at line 239 of file applyparallelworker.c.

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

MySubscription

PGDLLIMPORT Subscription* MySubscription

extern

Definition at line 315 of file worker.c.

Referenced by apply_handle_commit_prepared(), apply_handle_delete(), apply_handle_insert(), apply_handle_prepare_internal(), apply_handle_rollback_prepared(), apply_handle_truncate(), apply_handle_update(), ApplyWorkerMain(), clear_subscription_skip_lsn(), copy_table(), DisableSubscriptionAndExit(), fetch_remote_table_info(), FetchTableStates(), InitializeApplyWorker(), logicalrep_rel_open(), LogicalRepSyncTableStart(), maybe_reread_subscription(), maybe_start_skipping_changes(), pa_can_start(), pa_launch_parallel_worker(), pa_start_subtrans(), pa_stream_abort(), ParallelApplyWorkerMain(), pg_attribute_noreturn(), process_syncing_tables_for_apply(), ProcessParallelApplyInterrupts(), should_apply_changes_for_rel(), start_apply(), and start_table_sync().