PostgreSQL Source Code git master
All Data Structures Namespaces Files Functions Variables Typedefs Enumerations Enumerator Macros Pages
worker_internal.h File Reference
Include dependency graph for worker_internal.h:
This graph shows which files directly or indirectly include this file:

Go to the source code of this file.

Data Structures

struct  LogicalRepWorker
 
struct  ParallelApplyWorkerShared
 
struct  ParallelApplyWorkerInfo
 

Macros

#define isParallelApplyWorker(worker)
 
#define isTablesyncWorker(worker)
 

Typedefs

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

Enumerations

enum  LogicalRepWorkerType { WORKERTYPE_UNKNOWN = 0 , WORKERTYPE_TABLESYNC , WORKERTYPE_APPLY , WORKERTYPE_PARALLEL_APPLY }
 
enum  ParallelTransState { PARALLEL_TRANS_UNKNOWN , PARALLEL_TRANS_STARTED , PARALLEL_TRANS_FINISHED }
 
enum  PartialFileSetState { FS_EMPTY , FS_SERIALIZE_IN_PROGRESS , FS_SERIALIZE_DONE , FS_READY }
 

Functions

void logicalrep_worker_attach (int slot)
 
LogicalRepWorkerlogicalrep_worker_find (Oid subid, Oid relid, bool only_running)
 
Listlogicalrep_workers_find (Oid subid, bool only_running, bool acquire_lock)
 
bool logicalrep_worker_launch (LogicalRepWorkerType wtype, Oid dbid, Oid subid, const char *subname, Oid userid, Oid relid, dsm_handle subworker_dsm)
 
void logicalrep_worker_stop (Oid subid, Oid relid)
 
void logicalrep_pa_worker_stop (ParallelApplyWorkerInfo *winfo)
 
void logicalrep_worker_wakeup (Oid subid, Oid relid)
 
void logicalrep_worker_wakeup_ptr (LogicalRepWorker *worker)
 
int logicalrep_sync_worker_count (Oid subid)
 
void ReplicationOriginNameForLogicalRep (Oid suboid, Oid relid, char *originname, Size szoriginname)
 
bool AllTablesyncsReady (void)
 
void UpdateTwoPhaseState (Oid suboid, char new_state)
 
void process_syncing_tables (XLogRecPtr current_lsn)
 
void invalidate_syncing_table_states (Datum arg, int cacheid, uint32 hashvalue)
 
void stream_start_internal (TransactionId xid, bool first_segment)
 
void stream_stop_internal (TransactionId xid)
 
void apply_spooled_messages (FileSet *stream_fileset, TransactionId xid, XLogRecPtr lsn)
 
void apply_dispatch (StringInfo s)
 
void maybe_reread_subscription (void)
 
void stream_cleanup_files (Oid subid, TransactionId xid)
 
void set_stream_options (WalRcvStreamOptions *options, char *slotname, XLogRecPtr *origin_startpos)
 
void start_apply (XLogRecPtr origin_startpos)
 
void InitializeLogRepWorker (void)
 
void SetupApplyOrSyncWorker (int worker_slot)
 
void DisableSubscriptionAndExit (void)
 
void store_flush_position (XLogRecPtr remote_lsn, XLogRecPtr local_lsn)
 
void apply_error_callback (void *arg)
 
void set_apply_error_context_origin (char *originname)
 
void pa_allocate_worker (TransactionId xid)
 
ParallelApplyWorkerInfopa_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 ErrorContextCallbackapply_error_context_stack
 
PGDLLIMPORT ParallelApplyWorkerSharedMyParallelShared
 
PGDLLIMPORT struct WalReceiverConnLogRepWorkerWalRcvConn
 
PGDLLIMPORT SubscriptionMySubscription
 
PGDLLIMPORT LogicalRepWorkerMyLogicalRepWorker
 
PGDLLIMPORT bool in_remote_transaction
 
PGDLLIMPORT bool InitializingApplyWorker
 

Macro Definition Documentation

◆ isParallelApplyWorker

#define isParallelApplyWorker (   worker)
Value:
((worker)->in_use && \
(worker)->type == WORKERTYPE_PARALLEL_APPLY)
@ WORKERTYPE_PARALLEL_APPLY

Definition at line 329 of file worker_internal.h.

◆ isTablesyncWorker

#define isTablesyncWorker (   worker)
Value:
((worker)->in_use && \
(worker)->type == WORKERTYPE_TABLESYNC)
@ WORKERTYPE_TABLESYNC

Definition at line 331 of file worker_internal.h.

Typedef Documentation

◆ LogicalRepWorker

◆ LogicalRepWorkerType

◆ ParallelApplyWorkerInfo

◆ ParallelApplyWorkerShared

◆ ParallelTransState

◆ PartialFileSetState

Enumeration Type Documentation

◆ LogicalRepWorkerType

Enumerator
WORKERTYPE_UNKNOWN 
WORKERTYPE_TABLESYNC 
WORKERTYPE_APPLY 
WORKERTYPE_PARALLEL_APPLY 

Definition at line 29 of file worker_internal.h.

◆ ParallelTransState

Enumerator
PARALLEL_TRANS_UNKNOWN 
PARALLEL_TRANS_STARTED 
PARALLEL_TRANS_FINISHED 

Definition at line 103 of file worker_internal.h.

104{
ParallelTransState
@ PARALLEL_TRANS_UNKNOWN
@ PARALLEL_TRANS_STARTED
@ PARALLEL_TRANS_FINISHED

◆ PartialFileSetState

Enumerator
FS_EMPTY 
FS_SERIALIZE_IN_PROGRESS 
FS_SERIALIZE_DONE 
FS_READY 

Definition at line 126 of file worker_internal.h.

127{
128 FS_EMPTY,
131 FS_READY,
PartialFileSetState
@ FS_EMPTY
@ FS_SERIALIZE_DONE
@ FS_READY
@ FS_SERIALIZE_IN_PROGRESS

Function Documentation

◆ AllTablesyncsReady()

bool AllTablesyncsReady ( void  )

Definition at line 1739 of file tablesync.c.

1740{
1741 bool started_tx = false;
1742 bool has_subrels = false;
1743
1744 /* We need up-to-date sync state info for subscription tables here. */
1745 has_subrels = FetchTableStates(&started_tx);
1746
1747 if (started_tx)
1748 {
1750 pgstat_report_stat(true);
1751 }
1752
1753 /*
1754 * Return false when there are no tables in subscription or not all tables
1755 * are in ready state; true otherwise.
1756 */
1757 return has_subrels && (table_states_not_ready == NIL);
1758}
#define NIL
Definition: pg_list.h:68
long pgstat_report_stat(bool force)
Definition: pgstat.c:692
static List * table_states_not_ready
Definition: tablesync.c:134
static bool FetchTableStates(bool *started_tx)
Definition: tablesync.c:1580
void CommitTransactionCommand(void)
Definition: xact.c:3149

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

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

◆ am_leader_apply_worker()

static bool am_leader_apply_worker ( void  )
inlinestatic

◆ am_parallel_apply_worker()

◆ am_tablesync_worker()

static bool am_tablesync_worker ( void  )
inlinestatic

◆ apply_dispatch()

void apply_dispatch ( StringInfo  s)

Definition at line 3364 of file worker.c.

3365{
3367 LogicalRepMsgType saved_command;
3368
3369 /*
3370 * Set the current command being applied. Since this function can be
3371 * called recursively when applying spooled changes, save the current
3372 * command.
3373 */
3374 saved_command = apply_error_callback_arg.command;
3376
3377 switch (action)
3378 {
3381 break;
3382
3385 break;
3386
3389 break;
3390
3393 break;
3394
3397 break;
3398
3401 break;
3402
3405 break;
3406
3409 break;
3410
3413 break;
3414
3416
3417 /*
3418 * Logical replication does not use generic logical messages yet.
3419 * Although, it could be used by other applications that use this
3420 * output plugin.
3421 */
3422 break;
3423
3426 break;
3427
3430 break;
3431
3434 break;
3435
3438 break;
3439
3442 break;
3443
3446 break;
3447
3450 break;
3451
3454 break;
3455
3458 break;
3459
3460 default:
3461 ereport(ERROR,
3462 (errcode(ERRCODE_PROTOCOL_VIOLATION),
3463 errmsg("invalid logical replication message type \"??? (%d)\"", action)));
3464 }
3465
3466 /* Reset the current command */
3467 apply_error_callback_arg.command = saved_command;
3468}
static void apply_handle_stream_prepare(StringInfo s)
Definition: worker.c:1284
static void apply_handle_type(StringInfo s)
Definition: worker.c:2345
static void apply_handle_truncate(StringInfo s)
Definition: worker.c:3236
static void apply_handle_update(StringInfo s)
Definition: worker.c:2542
static void apply_handle_stream_commit(StringInfo s)
Definition: worker.c:2152
static void apply_handle_commit_prepared(StringInfo s)
Definition: worker.c:1177
static ApplyErrorCallbackArg apply_error_callback_arg
Definition: worker.c:279
static void apply_handle_delete(StringInfo s)
Definition: worker.c:2752
static void apply_handle_begin(StringInfo s)
Definition: worker.c:989
static void apply_handle_commit(StringInfo s)
Definition: worker.c:1014
static void apply_handle_stream_abort(StringInfo s)
Definition: worker.c:1833
static void apply_handle_relation(StringInfo s)
Definition: worker.c:2322
static void apply_handle_prepare(StringInfo s)
Definition: worker.c:1106
static void apply_handle_rollback_prepared(StringInfo s)
Definition: worker.c:1226
static void apply_handle_stream_stop(StringInfo s)
Definition: worker.c:1647
static void apply_handle_origin(StringInfo s)
Definition: worker.c:1429
static void apply_handle_begin_prepare(StringInfo s)
Definition: worker.c:1040
static void apply_handle_stream_start(StringInfo s)
Definition: worker.c:1488
static void apply_handle_insert(StringInfo s)
Definition: worker.c:2392
int errcode(int sqlerrcode)
Definition: elog.c:853
int errmsg(const char *fmt,...)
Definition: elog.c:1070
#define ERROR
Definition: elog.h:39
#define ereport(elevel,...)
Definition: elog.h:149
LogicalRepMsgType
Definition: logicalproto.h:58
@ LOGICAL_REP_MSG_INSERT
Definition: logicalproto.h:62
@ LOGICAL_REP_MSG_TRUNCATE
Definition: logicalproto.h:65
@ LOGICAL_REP_MSG_STREAM_STOP
Definition: logicalproto.h:74
@ LOGICAL_REP_MSG_BEGIN
Definition: logicalproto.h:59
@ LOGICAL_REP_MSG_STREAM_PREPARE
Definition: logicalproto.h:77
@ LOGICAL_REP_MSG_STREAM_ABORT
Definition: logicalproto.h:76
@ LOGICAL_REP_MSG_BEGIN_PREPARE
Definition: logicalproto.h:69
@ LOGICAL_REP_MSG_STREAM_START
Definition: logicalproto.h:73
@ LOGICAL_REP_MSG_COMMIT
Definition: logicalproto.h:60
@ LOGICAL_REP_MSG_PREPARE
Definition: logicalproto.h:70
@ LOGICAL_REP_MSG_RELATION
Definition: logicalproto.h:66
@ LOGICAL_REP_MSG_MESSAGE
Definition: logicalproto.h:68
@ LOGICAL_REP_MSG_ROLLBACK_PREPARED
Definition: logicalproto.h:72
@ LOGICAL_REP_MSG_COMMIT_PREPARED
Definition: logicalproto.h:71
@ LOGICAL_REP_MSG_TYPE
Definition: logicalproto.h:67
@ LOGICAL_REP_MSG_DELETE
Definition: logicalproto.h:64
@ LOGICAL_REP_MSG_STREAM_COMMIT
Definition: logicalproto.h:75
@ LOGICAL_REP_MSG_ORIGIN
Definition: logicalproto.h:61
@ LOGICAL_REP_MSG_UPDATE
Definition: logicalproto.h:63
int pq_getmsgbyte(StringInfo msg)
Definition: pqformat.c:399
LogicalRepMsgType command
Definition: worker.c:222

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 4995 of file worker.c.

4996{
4998 int elevel;
4999
5001 return;
5002
5003 Assert(errarg->origin_name);
5004
5005 elevel = geterrlevel();
5006
5007 /*
5008 * Reset the origin state to prevent the advancement of origin progress if
5009 * we fail to apply. Otherwise, this will result in transaction loss as
5010 * that transaction won't be sent again by the server.
5011 */
5012 if (elevel >= ERROR)
5013 replorigin_reset(0, (Datum) 0);
5014
5015 if (errarg->rel == NULL)
5016 {
5017 if (!TransactionIdIsValid(errarg->remote_xid))
5018 errcontext("processing remote data for replication origin \"%s\" during message type \"%s\"",
5019 errarg->origin_name,
5021 else if (XLogRecPtrIsInvalid(errarg->finish_lsn))
5022 errcontext("processing remote data for replication origin \"%s\" during message type \"%s\" in transaction %u",
5023 errarg->origin_name,
5025 errarg->remote_xid);
5026 else
5027 errcontext("processing remote data for replication origin \"%s\" during message type \"%s\" in transaction %u, finished at %X/%X",
5028 errarg->origin_name,
5030 errarg->remote_xid,
5031 LSN_FORMAT_ARGS(errarg->finish_lsn));
5032 }
5033 else
5034 {
5035 if (errarg->remote_attnum < 0)
5036 {
5037 if (XLogRecPtrIsInvalid(errarg->finish_lsn))
5038 errcontext("processing remote data for replication origin \"%s\" during message type \"%s\" for replication target relation \"%s.%s\" in transaction %u",
5039 errarg->origin_name,
5041 errarg->rel->remoterel.nspname,
5042 errarg->rel->remoterel.relname,
5043 errarg->remote_xid);
5044 else
5045 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",
5046 errarg->origin_name,
5048 errarg->rel->remoterel.nspname,
5049 errarg->rel->remoterel.relname,
5050 errarg->remote_xid,
5051 LSN_FORMAT_ARGS(errarg->finish_lsn));
5052 }
5053 else
5054 {
5055 if (XLogRecPtrIsInvalid(errarg->finish_lsn))
5056 errcontext("processing remote data for replication origin \"%s\" during message type \"%s\" for replication target relation \"%s.%s\" column \"%s\" in transaction %u",
5057 errarg->origin_name,
5059 errarg->rel->remoterel.nspname,
5060 errarg->rel->remoterel.relname,
5061 errarg->rel->remoterel.attnames[errarg->remote_attnum],
5062 errarg->remote_xid);
5063 else
5064 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",
5065 errarg->origin_name,
5067 errarg->rel->remoterel.nspname,
5068 errarg->rel->remoterel.relname,
5069 errarg->rel->remoterel.attnames[errarg->remote_attnum],
5070 errarg->remote_xid,
5071 LSN_FORMAT_ARGS(errarg->finish_lsn));
5072 }
5073 }
5074}
static void replorigin_reset(int code, Datum arg)
Definition: worker.c:4724
int geterrlevel(void)
Definition: elog.c:1578
#define errcontext
Definition: elog.h:196
uintptr_t Datum
Definition: postgres.h:64
const char * logicalrep_message_type(LogicalRepMsgType action)
Definition: proto.c:1196
TransactionId remote_xid
Definition: worker.c:227
XLogRecPtr finish_lsn
Definition: worker.c:228
LogicalRepRelMapEntry * rel
Definition: worker.c:223
LogicalRepRelation remoterel
#define TransactionIdIsValid(xid)
Definition: transam.h:41
#define LSN_FORMAT_ARGS(lsn)
Definition: xlogdefs.h:43
#define XLogRecPtrIsInvalid(r)
Definition: xlogdefs.h:29

References apply_error_callback_arg, Assert, LogicalRepRelation::attnames, ApplyErrorCallbackArg::command, errcontext, ERROR, ApplyErrorCallbackArg::finish_lsn, geterrlevel(), logicalrep_message_type(), LSN_FORMAT_ARGS, LogicalRepRelation::nspname, ApplyErrorCallbackArg::origin_name, ApplyErrorCallbackArg::rel, LogicalRepRelation::relname, ApplyErrorCallbackArg::remote_attnum, ApplyErrorCallbackArg::remote_xid, LogicalRepRelMapEntry::remoterel, replorigin_reset(), 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 2022 of file worker.c.

2024{
2025 int nchanges;
2026 char path[MAXPGPATH];
2027 char *buffer = NULL;
2028 MemoryContext oldcxt;
2029 ResourceOwner oldowner;
2030 int fileno;
2031 off_t offset;
2032
2035
2036 /* Make sure we have an open transaction */
2038
2039 /*
2040 * Allocate file handle and memory required to process all the messages in
2041 * TopTransactionContext to avoid them getting reset after each message is
2042 * processed.
2043 */
2045
2046 /* Open the spool file for the committed/prepared transaction */
2048 elog(DEBUG1, "replaying changes from file \"%s\"", path);
2049
2050 /*
2051 * Make sure the file is owned by the toplevel transaction so that the
2052 * file will not be accidentally closed when aborting a subtransaction.
2053 */
2054 oldowner = CurrentResourceOwner;
2056
2057 stream_fd = BufFileOpenFileSet(stream_fileset, path, O_RDONLY, false);
2058
2059 CurrentResourceOwner = oldowner;
2060
2061 buffer = palloc(BLCKSZ);
2062
2063 MemoryContextSwitchTo(oldcxt);
2064
2065 remote_final_lsn = lsn;
2066
2067 /*
2068 * Make sure the handle apply_dispatch methods are aware we're in a remote
2069 * transaction.
2070 */
2071 in_remote_transaction = true;
2073
2075
2076 /*
2077 * Read the entries one by one and pass them through the same logic as in
2078 * apply_dispatch.
2079 */
2080 nchanges = 0;
2081 while (true)
2082 {
2084 size_t nbytes;
2085 int len;
2086
2088
2089 /* read length of the on-disk record */
2090 nbytes = BufFileReadMaybeEOF(stream_fd, &len, sizeof(len), true);
2091
2092 /* have we reached end of the file? */
2093 if (nbytes == 0)
2094 break;
2095
2096 /* do we have a correct length? */
2097 if (len <= 0)
2098 elog(ERROR, "incorrect length %d in streaming transaction's changes file \"%s\"",
2099 len, path);
2100
2101 /* make sure we have sufficiently large buffer */
2102 buffer = repalloc(buffer, len);
2103
2104 /* and finally read the data into the buffer */
2105 BufFileReadExact(stream_fd, buffer, len);
2106
2107 BufFileTell(stream_fd, &fileno, &offset);
2108
2109 /* init a stringinfo using the buffer and call apply_dispatch */
2110 initReadOnlyStringInfo(&s2, buffer, len);
2111
2112 /* Ensure we are reading the data into our memory context. */
2114
2116
2118
2119 MemoryContextSwitchTo(oldcxt);
2120
2121 nchanges++;
2122
2123 /*
2124 * It is possible the file has been closed because we have processed
2125 * the transaction end message like stream_commit in which case that
2126 * must be the last message.
2127 */
2128 if (!stream_fd)
2129 {
2130 ensure_last_message(stream_fileset, xid, fileno, offset);
2131 break;
2132 }
2133
2134 if (nchanges % 1000 == 0)
2135 elog(DEBUG1, "replayed %d changes from file \"%s\"",
2136 nchanges, path);
2137 }
2138
2139 if (stream_fd)
2141
2142 elog(DEBUG1, "replayed %d (all) changes from file \"%s\"",
2143 nchanges, path);
2144
2145 return;
2146}
static void begin_replication_step(void)
Definition: worker.c:506
static void end_replication_step(void)
Definition: worker.c:529
MemoryContext ApplyMessageContext
Definition: worker.c:291
static void changes_filename(char *path, Oid subid, TransactionId xid)
Definition: worker.c:4286
static BufFile * stream_fd
Definition: worker.c:340
bool in_remote_transaction
Definition: worker.c:304
void apply_dispatch(StringInfo s)
Definition: worker.c:3364
static void ensure_last_message(FileSet *stream_fileset, TransactionId xid, int fileno, off_t offset)
Definition: worker.c:1990
static XLogRecPtr remote_final_lsn
Definition: worker.c:305
static void maybe_start_skipping_changes(XLogRecPtr finish_lsn)
Definition: worker.c:4865
static void stream_close_file(void)
Definition: worker.c:4369
void pgstat_report_activity(BackendState state, const char *cmd_str)
@ STATE_RUNNING
BufFile * BufFileOpenFileSet(FileSet *fileset, const char *name, int mode, bool missing_ok)
Definition: buffile.c:291
void BufFileReadExact(BufFile *file, void *ptr, size_t size)
Definition: buffile.c:654
void BufFileTell(BufFile *file, int *fileno, off_t *offset)
Definition: buffile.c:833
size_t BufFileReadMaybeEOF(BufFile *file, void *ptr, size_t size, bool eofOK)
Definition: buffile.c:664
#define DEBUG1
Definition: elog.h:30
#define elog(elevel,...)
Definition: elog.h:225
LogicalRepWorker * MyLogicalRepWorker
Definition: launcher.c:54
void MemoryContextReset(MemoryContext context)
Definition: mcxt.c:383
MemoryContext TopTransactionContext
Definition: mcxt.c:154
void * repalloc(void *pointer, Size size)
Definition: mcxt.c:1541
void * palloc(Size size)
Definition: mcxt.c:1317
#define CHECK_FOR_INTERRUPTS()
Definition: miscadmin.h:122
static MemoryContext MemoryContextSwitchTo(MemoryContext context)
Definition: palloc.h:124
#define MAXPGPATH
const void size_t len
char * s2
ResourceOwner TopTransactionResourceOwner
Definition: resowner.c:167
ResourceOwner CurrentResourceOwner
Definition: resowner.c:165
static void initReadOnlyStringInfo(StringInfo str, char *data, int len)
Definition: stringinfo.h:130
static bool am_parallel_apply_worker(void)

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

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

◆ DisableSubscriptionAndExit()

void DisableSubscriptionAndExit ( void  )

Definition at line 4807 of file worker.c.

4808{
4809 /*
4810 * Emit the error message, and recover from the error state to an idle
4811 * state
4812 */
4814
4818
4820
4821 /* Report the worker failed during either table synchronization or apply */
4824
4825 /* Disable the subscription */
4829
4830 /* Ensure we remove no-longer-useful entry for worker's start time */
4833
4834 /* Notify the subscription has been disabled and exit */
4835 ereport(LOG,
4836 errmsg("subscription \"%s\" has been disabled because of an error",
4838
4839 proc_exit(0);
4840}
Subscription * MySubscription
Definition: worker.c:299
void EmitErrorReport(void)
Definition: elog.c:1687
void FlushErrorState(void)
Definition: elog.c:1867
#define LOG
Definition: elog.h:31
void proc_exit(int code)
Definition: ipc.c:104
void ApplyLauncherForgetWorkerStartTime(Oid subid)
Definition: launcher.c:1072
#define RESUME_INTERRUPTS()
Definition: miscadmin.h:135
#define HOLD_INTERRUPTS()
Definition: miscadmin.h:133
void DisableSubscription(Oid subid)
void pgstat_report_subscription_error(Oid subid, bool is_apply_error)
static bool am_tablesync_worker(void)
static bool am_leader_apply_worker(void)
void StartTransactionCommand(void)
Definition: xact.c:3051
void AbortOutOfAnyTransaction(void)
Definition: xact.c:4854

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

Referenced by start_apply(), and start_table_sync().

◆ InitializeLogRepWorker()

void InitializeLogRepWorker ( void  )

Definition at line 4639 of file worker.c.

4640{
4641 MemoryContext oldctx;
4642
4643 /* Run as replica session replication role. */
4644 SetConfigOption("session_replication_role", "replica",
4646
4647 /* Connect to our database. */
4650 0);
4651
4652 /*
4653 * Set always-secure search path, so malicious users can't redirect user
4654 * code (e.g. pg_index.indexprs).
4655 */
4656 SetConfigOption("search_path", "", PGC_SUSET, PGC_S_OVERRIDE);
4657
4658 /* Load the subscription into persistent memory context. */
4660 "ApplyContext",
4664
4666 if (!MySubscription)
4667 {
4668 ereport(LOG,
4669 (errmsg("logical replication worker for subscription %u will not start because the subscription was removed during startup",
4671
4672 /* Ensure we remove no-longer-useful entry for worker's start time */
4675
4676 proc_exit(0);
4677 }
4678
4679 MySubscriptionValid = true;
4680 MemoryContextSwitchTo(oldctx);
4681
4682 if (!MySubscription->enabled)
4683 {
4684 ereport(LOG,
4685 (errmsg("logical replication worker for subscription \"%s\" will not start because the subscription was disabled during startup",
4686 MySubscription->name)));
4687
4689 }
4690
4691 /* Setup synchronous commit according to the user's wishes */
4692 SetConfigOption("synchronous_commit", MySubscription->synccommit,
4694
4695 /*
4696 * Keep us informed about subscription or role changes. Note that the
4697 * role's superuser privilege can be revoked.
4698 */
4699 CacheRegisterSyscacheCallback(SUBSCRIPTIONOID,
4701 (Datum) 0);
4702
4705 (Datum) 0);
4706
4707 if (am_tablesync_worker())
4708 ereport(LOG,
4709 (errmsg("logical replication table synchronization worker for subscription \"%s\", table \"%s\" has started",
4712 else
4713 ereport(LOG,
4714 (errmsg("logical replication apply worker for subscription \"%s\" has started",
4715 MySubscription->name)));
4716
4718}
static void subscription_change_cb(Datum arg, int cacheid, uint32 hashvalue)
Definition: worker.c:4086
static void apply_worker_exit(void)
Definition: worker.c:3923
MemoryContext ApplyContext
Definition: worker.c:292
static bool MySubscriptionValid
Definition: worker.c:300
void BackgroundWorkerInitializeConnectionByOid(Oid dboid, Oid useroid, uint32 flags)
Definition: bgworker.c:886
void SetConfigOption(const char *name, const char *value, GucContext context, GucSource source)
Definition: guc.c:4332
@ PGC_S_OVERRIDE
Definition: guc.h:119
@ PGC_SUSET
Definition: guc.h:74
@ PGC_BACKEND
Definition: guc.h:73
void CacheRegisterSyscacheCallback(int cacheid, SyscacheCallbackFunction func, Datum arg)
Definition: inval.c:1704
char * get_rel_name(Oid relid)
Definition: lsyscache.c:1928
MemoryContext TopMemoryContext
Definition: mcxt.c:149
#define AllocSetContextCreate
Definition: memutils.h:129
#define ALLOCSET_DEFAULT_SIZES
Definition: memutils.h:160
Subscription * GetSubscription(Oid subid, bool missing_ok)

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

Referenced by ParallelApplyWorkerMain(), and SetupApplyOrSyncWorker().

◆ invalidate_syncing_table_states()

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

Definition at line 281 of file tablesync.c.

282{
284}
@ SYNC_TABLE_STATE_NEEDS_REBUILD
Definition: tablesync.c:128
static SyncingTablesState table_states_validity
Definition: tablesync.c:133

References SYNC_TABLE_STATE_NEEDS_REBUILD, and table_states_validity.

Referenced by ParallelApplyWorkerMain(), and SetupApplyOrSyncWorker().

◆ logicalrep_pa_worker_stop()

void logicalrep_pa_worker_stop ( ParallelApplyWorkerInfo winfo)

Definition at line 630 of file launcher.c.

631{
632 int slot_no;
633 uint16 generation;
634 LogicalRepWorker *worker;
635
636 SpinLockAcquire(&winfo->shared->mutex);
637 generation = winfo->shared->logicalrep_worker_generation;
638 slot_no = winfo->shared->logicalrep_worker_slot_no;
639 SpinLockRelease(&winfo->shared->mutex);
640
641 Assert(slot_no >= 0 && slot_no < max_logical_replication_workers);
642
643 /*
644 * Detach from the error_mq_handle for the parallel apply worker before
645 * stopping it. This prevents the leader apply worker from trying to
646 * receive the message from the error queue that might already be detached
647 * by the parallel apply worker.
648 */
649 if (winfo->error_mq_handle)
650 {
652 winfo->error_mq_handle = NULL;
653 }
654
655 LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
656
657 worker = &LogicalRepCtx->workers[slot_no];
659
660 /*
661 * Only stop the worker if the generation matches and the worker is alive.
662 */
663 if (worker->generation == generation && worker->proc)
664 logicalrep_worker_stop_internal(worker, SIGINT);
665
666 LWLockRelease(LogicalRepWorkerLock);
667}
uint16_t uint16
Definition: c.h:484
int max_logical_replication_workers
Definition: launcher.c:50
static void logicalrep_worker_stop_internal(LogicalRepWorker *worker, int signo)
Definition: launcher.c:524
static LogicalRepCtxStruct * LogicalRepCtx
Definition: launcher.c:69
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1168
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1781
@ LW_SHARED
Definition: lwlock.h:115
void shm_mq_detach(shm_mq_handle *mqh)
Definition: shm_mq.c:843
#define SpinLockRelease(lock)
Definition: spin.h:61
#define SpinLockAcquire(lock)
Definition: spin.h:59
LogicalRepWorker workers[FLEXIBLE_ARRAY_MEMBER]
Definition: launcher.c:66
shm_mq_handle * error_mq_handle
ParallelApplyWorkerShared * shared

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

Referenced by pa_free_worker().

◆ logicalrep_sync_worker_count()

int logicalrep_sync_worker_count ( Oid  subid)

Definition at line 845 of file launcher.c.

846{
847 int i;
848 int res = 0;
849
850 Assert(LWLockHeldByMe(LogicalRepWorkerLock));
851
852 /* Search for attached worker for a given subscription id. */
853 for (i = 0; i < max_logical_replication_workers; i++)
854 {
856
857 if (isTablesyncWorker(w) && w->subid == subid)
858 res++;
859 }
860
861 return res;
862}
int i
Definition: isn.c:72
bool LWLockHeldByMe(LWLock *lock)
Definition: lwlock.c:1893

References Assert, i, isTablesyncWorker, LogicalRepCtx, LWLockHeldByMe(), max_logical_replication_workers, 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 704 of file launcher.c.

705{
706 /* Block concurrent access. */
707 LWLockAcquire(LogicalRepWorkerLock, LW_EXCLUSIVE);
708
709 Assert(slot >= 0 && slot < max_logical_replication_workers);
711
713 {
714 LWLockRelease(LogicalRepWorkerLock);
716 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
717 errmsg("logical replication worker slot %d is empty, cannot attach",
718 slot)));
719 }
720
722 {
723 LWLockRelease(LogicalRepWorkerLock);
725 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
726 errmsg("logical replication worker slot %d is already used by "
727 "another worker, cannot attach", slot)));
728 }
729
732
733 LWLockRelease(LogicalRepWorkerLock);
734}
void before_shmem_exit(pg_on_exit_callback function, Datum arg)
Definition: ipc.c:337
static void logicalrep_worker_onexit(int code, Datum arg)
Definition: launcher.c:815
@ LW_EXCLUSIVE
Definition: lwlock.h:114
PGPROC * MyProc
Definition: proc.c:66

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

Referenced by ParallelApplyWorkerMain(), and SetupApplyOrSyncWorker().

◆ logicalrep_worker_find()

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

Definition at line 234 of file launcher.c.

235{
236 int i;
237 LogicalRepWorker *res = NULL;
238
239 Assert(LWLockHeldByMe(LogicalRepWorkerLock));
240
241 /* Search for attached worker for a given subscription id. */
242 for (i = 0; i < max_logical_replication_workers; i++)
243 {
245
246 /* Skip parallel apply workers. */
248 continue;
249
250 if (w->in_use && w->subid == subid && w->relid == relid &&
251 (!only_running || w->proc))
252 {
253 res = w;
254 break;
255 }
256 }
257
258 return res;
259}

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 ( LogicalRepWorkerType  wtype,
Oid  dbid,
Oid  subid,
const char *  subname,
Oid  userid,
Oid  relid,
dsm_handle  subworker_dsm 
)

Definition at line 297 of file launcher.c.

300{
302 BackgroundWorkerHandle *bgw_handle;
303 uint16 generation;
304 int i;
305 int slot = 0;
306 LogicalRepWorker *worker = NULL;
307 int nsyncworkers;
308 int nparallelapplyworkers;
310 bool is_tablesync_worker = (wtype == WORKERTYPE_TABLESYNC);
311 bool is_parallel_apply_worker = (wtype == WORKERTYPE_PARALLEL_APPLY);
312
313 /*----------
314 * Sanity checks:
315 * - must be valid worker type
316 * - tablesync workers are only ones to have relid
317 * - parallel apply worker is the only kind of subworker
318 */
319 Assert(wtype != WORKERTYPE_UNKNOWN);
320 Assert(is_tablesync_worker == OidIsValid(relid));
321 Assert(is_parallel_apply_worker == (subworker_dsm != DSM_HANDLE_INVALID));
322
324 (errmsg_internal("starting logical replication worker for subscription \"%s\"",
325 subname)));
326
327 /* Report this after the initial starting message for consistency. */
328 if (max_replication_slots == 0)
330 (errcode(ERRCODE_CONFIGURATION_LIMIT_EXCEEDED),
331 errmsg("cannot start logical replication workers when \"max_replication_slots\"=0")));
332
333 /*
334 * We need to do the modification of the shared memory under lock so that
335 * we have consistent view.
336 */
337 LWLockAcquire(LogicalRepWorkerLock, LW_EXCLUSIVE);
338
339retry:
340 /* Find unused worker slot. */
341 for (i = 0; i < max_logical_replication_workers; i++)
342 {
344
345 if (!w->in_use)
346 {
347 worker = w;
348 slot = i;
349 break;
350 }
351 }
352
353 nsyncworkers = logicalrep_sync_worker_count(subid);
354
356
357 /*
358 * If we didn't find a free slot, try to do garbage collection. The
359 * reason we do this is because if some worker failed to start up and its
360 * parent has crashed while waiting, the in_use state was never cleared.
361 */
362 if (worker == NULL || nsyncworkers >= max_sync_workers_per_subscription)
363 {
364 bool did_cleanup = false;
365
366 for (i = 0; i < max_logical_replication_workers; i++)
367 {
369
370 /*
371 * If the worker was marked in use but didn't manage to attach in
372 * time, clean it up.
373 */
374 if (w->in_use && !w->proc &&
377 {
379 "logical replication worker for subscription %u took too long to start; canceled",
380 w->subid);
381
383 did_cleanup = true;
384 }
385 }
386
387 if (did_cleanup)
388 goto retry;
389 }
390
391 /*
392 * We don't allow to invoke more sync workers once we have reached the
393 * sync worker limit per subscription. So, just return silently as we
394 * might get here because of an otherwise harmless race condition.
395 */
396 if (is_tablesync_worker && nsyncworkers >= max_sync_workers_per_subscription)
397 {
398 LWLockRelease(LogicalRepWorkerLock);
399 return false;
400 }
401
402 nparallelapplyworkers = logicalrep_pa_worker_count(subid);
403
404 /*
405 * Return false if the number of parallel apply workers reached the limit
406 * per subscription.
407 */
408 if (is_parallel_apply_worker &&
409 nparallelapplyworkers >= max_parallel_apply_workers_per_subscription)
410 {
411 LWLockRelease(LogicalRepWorkerLock);
412 return false;
413 }
414
415 /*
416 * However if there are no more free worker slots, inform user about it
417 * before exiting.
418 */
419 if (worker == NULL)
420 {
421 LWLockRelease(LogicalRepWorkerLock);
423 (errcode(ERRCODE_CONFIGURATION_LIMIT_EXCEEDED),
424 errmsg("out of logical replication worker slots"),
425 errhint("You might need to increase \"%s\".", "max_logical_replication_workers")));
426 return false;
427 }
428
429 /* Prepare the worker slot. */
430 worker->type = wtype;
431 worker->launch_time = now;
432 worker->in_use = true;
433 worker->generation++;
434 worker->proc = NULL;
435 worker->dbid = dbid;
436 worker->userid = userid;
437 worker->subid = subid;
438 worker->relid = relid;
439 worker->relstate = SUBREL_STATE_UNKNOWN;
441 worker->stream_fileset = NULL;
442 worker->leader_pid = is_parallel_apply_worker ? MyProcPid : InvalidPid;
443 worker->parallel_apply = is_parallel_apply_worker;
444 worker->last_lsn = InvalidXLogRecPtr;
449
450 /* Before releasing lock, remember generation for future identification. */
451 generation = worker->generation;
452
453 LWLockRelease(LogicalRepWorkerLock);
454
455 /* Register the new dynamic worker. */
456 memset(&bgw, 0, sizeof(bgw));
460 snprintf(bgw.bgw_library_name, MAXPGPATH, "postgres");
461
462 switch (worker->type)
463 {
464 case WORKERTYPE_APPLY:
465 snprintf(bgw.bgw_function_name, BGW_MAXLEN, "ApplyWorkerMain");
467 "logical replication apply worker for subscription %u",
468 subid);
469 snprintf(bgw.bgw_type, BGW_MAXLEN, "logical replication apply worker");
470 break;
471
473 snprintf(bgw.bgw_function_name, BGW_MAXLEN, "ParallelApplyWorkerMain");
475 "logical replication parallel apply worker for subscription %u",
476 subid);
477 snprintf(bgw.bgw_type, BGW_MAXLEN, "logical replication parallel worker");
478
479 memcpy(bgw.bgw_extra, &subworker_dsm, sizeof(dsm_handle));
480 break;
481
483 snprintf(bgw.bgw_function_name, BGW_MAXLEN, "TablesyncWorkerMain");
485 "logical replication tablesync worker for subscription %u sync %u",
486 subid,
487 relid);
488 snprintf(bgw.bgw_type, BGW_MAXLEN, "logical replication tablesync worker");
489 break;
490
492 /* Should never happen. */
493 elog(ERROR, "unknown worker type");
494 }
495
498 bgw.bgw_main_arg = Int32GetDatum(slot);
499
500 if (!RegisterDynamicBackgroundWorker(&bgw, &bgw_handle))
501 {
502 /* Failed to start worker, so clean up the worker slot. */
503 LWLockAcquire(LogicalRepWorkerLock, LW_EXCLUSIVE);
504 Assert(generation == worker->generation);
506 LWLockRelease(LogicalRepWorkerLock);
507
509 (errcode(ERRCODE_CONFIGURATION_LIMIT_EXCEEDED),
510 errmsg("out of background worker slots"),
511 errhint("You might need to increase \"%s\".", "max_worker_processes")));
512 return false;
513 }
514
515 /* Now wait until it attaches. */
516 return WaitForReplicationWorkerAttach(worker, generation, bgw_handle);
517}
bool TimestampDifferenceExceeds(TimestampTz start_time, TimestampTz stop_time, int msec)
Definition: timestamp.c:1780
TimestampTz GetCurrentTimestamp(void)
Definition: timestamp.c:1644
Datum now(PG_FUNCTION_ARGS)
Definition: timestamp.c:1608
bool RegisterDynamicBackgroundWorker(BackgroundWorker *worker, BackgroundWorkerHandle **handle)
Definition: bgworker.c:1045
#define BGW_NEVER_RESTART
Definition: bgworker.h:85
@ BgWorkerStart_RecoveryFinished
Definition: bgworker.h:81
#define BGWORKER_BACKEND_DATABASE_CONNECTION
Definition: bgworker.h:60
#define BGWORKER_SHMEM_ACCESS
Definition: bgworker.h:53
#define BGW_MAXLEN
Definition: bgworker.h:86
#define OidIsValid(objectId)
Definition: c.h:729
int64 TimestampTz
Definition: timestamp.h:39
#define TIMESTAMP_NOBEGIN(j)
Definition: timestamp.h:159
uint32 dsm_handle
Definition: dsm_impl.h:55
#define DSM_HANDLE_INVALID
Definition: dsm_impl.h:58
int errmsg_internal(const char *fmt,...)
Definition: elog.c:1157
int errhint(const char *fmt,...)
Definition: elog.c:1317
#define WARNING
Definition: elog.h:36
int MyProcPid
Definition: globals.c:46
static int logicalrep_pa_worker_count(Oid subid)
Definition: launcher.c:869
int max_sync_workers_per_subscription
Definition: launcher.c:51
static bool WaitForReplicationWorkerAttach(LogicalRepWorker *worker, uint16 generation, BackgroundWorkerHandle *handle)
Definition: launcher.c:174
int logicalrep_sync_worker_count(Oid subid)
Definition: launcher.c:845
int max_parallel_apply_workers_per_subscription
Definition: launcher.c:52
static void logicalrep_worker_cleanup(LogicalRepWorker *worker)
Definition: launcher.c:783
#define InvalidPid
Definition: miscadmin.h:32
NameData subname
#define snprintf
Definition: port.h:238
static Datum Int32GetDatum(int32 X)
Definition: postgres.h:212
int max_replication_slots
Definition: slot.c:141
char bgw_function_name[BGW_MAXLEN]
Definition: bgworker.h:97
Datum bgw_main_arg
Definition: bgworker.h:98
char bgw_name[BGW_MAXLEN]
Definition: bgworker.h:91
int bgw_restart_time
Definition: bgworker.h:95
char bgw_type[BGW_MAXLEN]
Definition: bgworker.h:92
BgWorkerStartTime bgw_start_time
Definition: bgworker.h:94
char bgw_extra[BGW_EXTRALEN]
Definition: bgworker.h:99
pid_t bgw_notify_pid
Definition: bgworker.h:100
char bgw_library_name[MAXPGPATH]
Definition: bgworker.h:96
XLogRecPtr relstate_lsn
TimestampTz last_recv_time
TimestampTz launch_time
TimestampTz reply_time
FileSet * stream_fileset
XLogRecPtr reply_lsn
XLogRecPtr last_lsn
TimestampTz last_send_time
int wal_receiver_timeout
Definition: walreceiver.c:88
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28

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, MAXPGPATH, 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::type, LogicalRepWorker::userid, WaitForReplicationWorkerAttach(), wal_receiver_timeout, WARNING, LogicalRepCtxStruct::workers, WORKERTYPE_APPLY, WORKERTYPE_PARALLEL_APPLY, WORKERTYPE_TABLESYNC, and WORKERTYPE_UNKNOWN.

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

◆ logicalrep_worker_stop()

void logicalrep_worker_stop ( Oid  subid,
Oid  relid 
)

Definition at line 606 of file launcher.c.

607{
608 LogicalRepWorker *worker;
609
610 LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
611
612 worker = logicalrep_worker_find(subid, relid, false);
613
614 if (worker)
615 {
617 logicalrep_worker_stop_internal(worker, SIGTERM);
618 }
619
620 LWLockRelease(LogicalRepWorkerLock);
621}
LogicalRepWorker * logicalrep_worker_find(Oid subid, Oid relid, bool only_running)
Definition: launcher.c:234

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

Referenced by AlterSubscription_refresh(), and DropSubscription().

◆ logicalrep_worker_wakeup()

void logicalrep_worker_wakeup ( Oid  subid,
Oid  relid 
)

Definition at line 673 of file launcher.c.

674{
675 LogicalRepWorker *worker;
676
677 LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
678
679 worker = logicalrep_worker_find(subid, relid, true);
680
681 if (worker)
683
684 LWLockRelease(LogicalRepWorkerLock);
685}
void logicalrep_worker_wakeup_ptr(LogicalRepWorker *worker)
Definition: launcher.c:693

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 693 of file launcher.c.

694{
695 Assert(LWLockHeldByMe(LogicalRepWorkerLock));
696
697 SetLatch(&worker->proc->procLatch);
698}
void SetLatch(Latch *latch)
Definition: latch.c:632
Latch procLatch
Definition: proc.h:169

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

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

◆ logicalrep_workers_find()

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

Definition at line 266 of file launcher.c.

267{
268 int i;
269 List *res = NIL;
270
271 if (acquire_lock)
272 LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
273
274 Assert(LWLockHeldByMe(LogicalRepWorkerLock));
275
276 /* Search for attached worker for a given subscription id. */
277 for (i = 0; i < max_logical_replication_workers; i++)
278 {
280
281 if (w->in_use && w->subid == subid && (!only_running || w->proc))
282 res = lappend(res, w);
283 }
284
285 if (acquire_lock)
286 LWLockRelease(LogicalRepWorkerLock);
287
288 return res;
289}
List * lappend(List *list, void *datum)
Definition: list.c:339
Definition: pg_list.h:54

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

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

◆ maybe_reread_subscription()

void maybe_reread_subscription ( void  )

Definition at line 3957 of file worker.c.

3958{
3959 MemoryContext oldctx;
3961 bool started_tx = false;
3962
3963 /* When cache state is valid there is nothing to do here. */
3965 return;
3966
3967 /* This function might be called inside or outside of transaction. */
3968 if (!IsTransactionState())
3969 {
3971 started_tx = true;
3972 }
3973
3974 /* Ensure allocations in permanent context. */
3976
3978
3979 /*
3980 * Exit if the subscription was removed. This normally should not happen
3981 * as the worker gets killed during DROP SUBSCRIPTION.
3982 */
3983 if (!newsub)
3984 {
3985 ereport(LOG,
3986 (errmsg("logical replication worker for subscription \"%s\" will stop because the subscription was removed",
3987 MySubscription->name)));
3988
3989 /* Ensure we remove no-longer-useful entry for worker's start time */
3992
3993 proc_exit(0);
3994 }
3995
3996 /* Exit if the subscription was disabled. */
3997 if (!newsub->enabled)
3998 {
3999 ereport(LOG,
4000 (errmsg("logical replication worker for subscription \"%s\" will stop because the subscription was disabled",
4001 MySubscription->name)));
4002
4004 }
4005
4006 /* !slotname should never happen when enabled is true. */
4007 Assert(newsub->slotname);
4008
4009 /* two-phase cannot be altered while the worker is running */
4010 Assert(newsub->twophasestate == MySubscription->twophasestate);
4011
4012 /*
4013 * Exit if any parameter that affects the remote connection was changed.
4014 * The launcher will start a new worker but note that the parallel apply
4015 * worker won't restart if the streaming option's value is changed from
4016 * 'parallel' to any other value or the server decides not to stream the
4017 * in-progress transaction.
4018 */
4019 if (strcmp(newsub->conninfo, MySubscription->conninfo) != 0 ||
4020 strcmp(newsub->name, MySubscription->name) != 0 ||
4021 strcmp(newsub->slotname, MySubscription->slotname) != 0 ||
4022 newsub->binary != MySubscription->binary ||
4023 newsub->stream != MySubscription->stream ||
4024 newsub->passwordrequired != MySubscription->passwordrequired ||
4025 strcmp(newsub->origin, MySubscription->origin) != 0 ||
4026 newsub->owner != MySubscription->owner ||
4027 !equal(newsub->publications, MySubscription->publications))
4028 {
4030 ereport(LOG,
4031 (errmsg("logical replication parallel apply worker for subscription \"%s\" will stop because of a parameter change",
4032 MySubscription->name)));
4033 else
4034 ereport(LOG,
4035 (errmsg("logical replication worker for subscription \"%s\" will restart because of a parameter change",
4036 MySubscription->name)));
4037
4039 }
4040
4041 /*
4042 * Exit if the subscription owner's superuser privileges have been
4043 * revoked.
4044 */
4045 if (!newsub->ownersuperuser && MySubscription->ownersuperuser)
4046 {
4048 ereport(LOG,
4049 errmsg("logical replication parallel apply worker for subscription \"%s\" will stop because the subscription owner's superuser privileges have been revoked",
4051 else
4052 ereport(LOG,
4053 errmsg("logical replication worker for subscription \"%s\" will restart because the subscription owner's superuser privileges have been revoked",
4055
4057 }
4058
4059 /* Check for other changes that should never happen too. */
4060 if (newsub->dbid != MySubscription->dbid)
4061 {
4062 elog(ERROR, "subscription %u changed unexpectedly",
4064 }
4065
4066 /* Clean old subscription info and switch to new one. */
4069
4070 MemoryContextSwitchTo(oldctx);
4071
4072 /* Change synchronous commit according to the user's wishes */
4073 SetConfigOption("synchronous_commit", MySubscription->synccommit,
4075
4076 if (started_tx)
4078
4079 MySubscriptionValid = true;
4080}
bool equal(const void *a, const void *b)
Definition: equalfuncs.c:223
void FreeSubscription(Subscription *sub)
static color newsub(struct colormap *cm, color co)
Definition: regc_color.c:389
bool IsTransactionState(void)
Definition: xact.c:386

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

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

◆ pa_allocate_worker()

void pa_allocate_worker ( TransactionId  xid)

Definition at line 470 of file applyparallelworker.c.

471{
472 bool found;
473 ParallelApplyWorkerInfo *winfo = NULL;
475
476 if (!pa_can_start())
477 return;
478
480 if (!winfo)
481 return;
482
483 /* First time through, initialize parallel apply worker state hashtable. */
485 {
486 HASHCTL ctl;
487
488 MemSet(&ctl, 0, sizeof(ctl));
489 ctl.keysize = sizeof(TransactionId);
490 ctl.entrysize = sizeof(ParallelApplyWorkerEntry);
491 ctl.hcxt = ApplyContext;
492
493 ParallelApplyTxnHash = hash_create("logical replication parallel apply workers hash",
494 16, &ctl,
496 }
497
498 /* Create an entry for the requested transaction. */
499 entry = hash_search(ParallelApplyTxnHash, &xid, HASH_ENTER, &found);
500 if (found)
501 elog(ERROR, "hash table corrupted");
502
503 /* Update the transaction information in shared memory. */
504 SpinLockAcquire(&winfo->shared->mutex);
506 winfo->shared->xid = xid;
507 SpinLockRelease(&winfo->shared->mutex);
508
509 winfo->in_use = true;
510 winfo->serialize_changes = false;
511 entry->winfo = winfo;
512}
struct ParallelApplyWorkerEntry ParallelApplyWorkerEntry
static bool pa_can_start(void)
static HTAB * ParallelApplyTxnHash
static ParallelApplyWorkerInfo * pa_launch_parallel_worker(void)
#define MemSet(start, val, len)
Definition: c.h:974
uint32 TransactionId
Definition: c.h:606
void * hash_search(HTAB *hashp, const void *keyPtr, HASHACTION action, bool *foundPtr)
Definition: dynahash.c:955
HTAB * hash_create(const char *tabname, long nelem, const HASHCTL *info, int flags)
Definition: dynahash.c:352
@ HASH_ENTER
Definition: hsearch.h:114
#define HASH_CONTEXT
Definition: hsearch.h:102
#define HASH_ELEM
Definition: hsearch.h:95
#define HASH_BLOBS
Definition: hsearch.h:97
tree ctl
Definition: radixtree.h:1838
ParallelApplyWorkerInfo * winfo
ParallelTransState xact_state

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

Referenced by apply_handle_stream_start().

◆ pa_decr_and_wait_stream_block()

void pa_decr_and_wait_stream_block ( void  )

Definition at line 1591 of file applyparallelworker.c.

1592{
1594
1595 /*
1596 * It is only possible to not have any pending stream chunks when we are
1597 * applying spooled messages.
1598 */
1600 {
1602 return;
1603
1604 elog(ERROR, "invalid pending streaming chunk 0");
1605 }
1606
1608 {
1611 }
1612}
void pa_unlock_stream(TransactionId xid, LOCKMODE lockmode)
void pa_lock_stream(TransactionId xid, LOCKMODE lockmode)
static bool pa_has_spooled_message_pending()
ParallelApplyWorkerShared * MyParallelShared
static uint32 pg_atomic_sub_fetch_u32(volatile pg_atomic_uint32 *ptr, int32 sub_)
Definition: atomics.h:439
static uint32 pg_atomic_read_u32(volatile pg_atomic_uint32 *ptr)
Definition: atomics.h:239
#define AccessShareLock
Definition: lockdefs.h:36
pg_atomic_uint32 pending_stream_count

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.

623{
624 ListCell *lc;
625
626 foreach(lc, ParallelApplyWorkerPool)
627 {
629
630 if (winfo->error_mq_handle)
631 {
633 winfo->error_mq_handle = NULL;
634 }
635 }
636}
static List * ParallelApplyWorkerPool
#define lfirst(lc)
Definition: pg_list.h:172

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.

519{
520 bool found;
522
523 if (!TransactionIdIsValid(xid))
524 return NULL;
525
527 return NULL;
528
529 /* Return the cached parallel apply worker if valid. */
531 return stream_apply_worker;
532
533 /* Find an entry for the requested transaction. */
534 entry = hash_search(ParallelApplyTxnHash, &xid, HASH_FIND, &found);
535 if (found)
536 {
537 /* The worker must not have exited. */
538 Assert(entry->winfo->in_use);
539 return entry->winfo;
540 }
541
542 return NULL;
543}
static ParallelApplyWorkerInfo * stream_apply_worker
@ HASH_FIND
Definition: hsearch.h:113

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 
)

◆ pa_lock_transaction()

void pa_lock_transaction ( TransactionId  xid,
LOCKMODE  lockmode 
)

◆ pa_reset_subtrans()

void pa_reset_subtrans ( void  )

Definition at line 1402 of file applyparallelworker.c.

1403{
1404 /*
1405 * We don't need to free this explicitly as the allocated memory will be
1406 * freed at the transaction end.
1407 */
1408 subxactlist = NIL;
1409}
static List * subxactlist

References NIL, and subxactlist.

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

◆ pa_send_data()

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

Definition at line 1146 of file applyparallelworker.c.

1147{
1148 int rc;
1149 shm_mq_result result;
1150 TimestampTz startTime = 0;
1151
1153 Assert(!winfo->serialize_changes);
1154
1155 /*
1156 * We don't try to send data to parallel worker for 'immediate' mode. This
1157 * is primarily used for testing purposes.
1158 */
1160 return false;
1161
1162/*
1163 * This timeout is a bit arbitrary but testing revealed that it is sufficient
1164 * to send the message unless the parallel apply worker is waiting on some
1165 * lock or there is a serious resource crunch. See the comments atop this file
1166 * to know why we are using a non-blocking way to send the message.
1167 */
1168#define SHM_SEND_RETRY_INTERVAL_MS 1000
1169#define SHM_SEND_TIMEOUT_MS (10000 - SHM_SEND_RETRY_INTERVAL_MS)
1170
1171 for (;;)
1172 {
1173 result = shm_mq_send(winfo->mq_handle, nbytes, data, true, true);
1174
1175 if (result == SHM_MQ_SUCCESS)
1176 return true;
1177 else if (result == SHM_MQ_DETACHED)
1178 ereport(ERROR,
1179 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1180 errmsg("could not send data to shared-memory queue")));
1181
1182 Assert(result == SHM_MQ_WOULD_BLOCK);
1183
1184 /* Wait before retrying. */
1185 rc = WaitLatch(MyLatch,
1188 WAIT_EVENT_LOGICAL_APPLY_SEND_DATA);
1189
1190 if (rc & WL_LATCH_SET)
1191 {
1194 }
1195
1196 if (startTime == 0)
1197 startTime = GetCurrentTimestamp();
1198 else if (TimestampDifferenceExceeds(startTime, GetCurrentTimestamp(),
1200 return false;
1201 }
1202}
#define SHM_SEND_TIMEOUT_MS
#define SHM_SEND_RETRY_INTERVAL_MS
#define unlikely(x)
Definition: c.h:330
struct Latch * MyLatch
Definition: globals.c:62
void ResetLatch(Latch *latch)
Definition: latch.c:724
int WaitLatch(Latch *latch, int wakeEvents, long timeout, uint32 wait_event_info)
Definition: latch.c:517
#define WL_TIMEOUT
Definition: latch.h:130
#define WL_EXIT_ON_PM_DEATH
Definition: latch.h:132
#define WL_LATCH_SET
Definition: latch.h:127
const void * data
int debug_logical_replication_streaming
@ DEBUG_LOGICAL_REP_STREAMING_IMMEDIATE
Definition: reorderbuffer.h:34
shm_mq_result shm_mq_send(shm_mq_handle *mqh, Size nbytes, const void *data, bool nowait, bool force_flush)
Definition: shm_mq.c:329
shm_mq_result
Definition: shm_mq.h:37
@ SHM_MQ_SUCCESS
Definition: shm_mq.h:38
@ SHM_MQ_WOULD_BLOCK
Definition: shm_mq.h:39
@ SHM_MQ_DETACHED
Definition: shm_mq.h:40
shm_mq_handle * mq_handle

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

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

◆ pa_set_fileset_state()

◆ pa_set_stream_apply_worker()

void pa_set_stream_apply_worker ( ParallelApplyWorkerInfo winfo)

Definition at line 1334 of file applyparallelworker.c.

1335{
1336 stream_apply_worker = winfo;
1337}

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 
)

◆ pa_start_subtrans()

void pa_start_subtrans ( TransactionId  current_xid,
TransactionId  top_xid 
)

Definition at line 1362 of file applyparallelworker.c.

1363{
1364 if (current_xid != top_xid &&
1365 !list_member_xid(subxactlist, current_xid))
1366 {
1367 MemoryContext oldctx;
1368 char spname[NAMEDATALEN];
1369
1370 pa_savepoint_name(MySubscription->oid, current_xid,
1371 spname, sizeof(spname));
1372
1373 elog(DEBUG1, "defining savepoint %s in logical replication parallel apply worker", spname);
1374
1375 /* We must be in transaction block to define the SAVEPOINT. */
1376 if (!IsTransactionBlock())
1377 {
1378 if (!IsTransactionState())
1380
1383 }
1384
1385 DefineSavepoint(spname);
1386
1387 /*
1388 * CommitTransactionCommand is needed to start a subtransaction after
1389 * issuing a SAVEPOINT inside a transaction block (see
1390 * StartSubTransaction()).
1391 */
1393
1395 subxactlist = lappend_xid(subxactlist, current_xid);
1396 MemoryContextSwitchTo(oldctx);
1397 }
1398}
static void pa_savepoint_name(Oid suboid, TransactionId xid, char *spname, Size szsp)
List * lappend_xid(List *list, TransactionId datum)
Definition: list.c:393
bool list_member_xid(const List *list, TransactionId datum)
Definition: list.c:742
#define NAMEDATALEN
void DefineSavepoint(const char *name)
Definition: xact.c:4365
bool IsTransactionBlock(void)
Definition: xact.c:4963
void BeginTransactionBlock(void)
Definition: xact.c:3916

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

Referenced by handle_streamed_transaction().

◆ pa_stream_abort()

void pa_stream_abort ( LogicalRepStreamAbortData abort_data)

Definition at line 1416 of file applyparallelworker.c.

1417{
1418 TransactionId xid = abort_data->xid;
1419 TransactionId subxid = abort_data->subxid;
1420
1421 /*
1422 * Update origin state so we can restart streaming from correct position
1423 * in case of crash.
1424 */
1427
1428 /*
1429 * If the two XIDs are the same, it's in fact abort of toplevel xact, so
1430 * just free the subxactlist.
1431 */
1432 if (subxid == xid)
1433 {
1435
1436 /*
1437 * Release the lock as we might be processing an empty streaming
1438 * transaction in which case the lock won't be released during
1439 * transaction rollback.
1440 *
1441 * Note that it's ok to release the transaction lock before aborting
1442 * the transaction because even if the parallel apply worker dies due
1443 * to crash or some other reason, such a transaction would still be
1444 * considered aborted.
1445 */
1447
1449
1450 if (IsTransactionBlock())
1451 {
1452 EndTransactionBlock(false);
1454 }
1455
1457
1459 }
1460 else
1461 {
1462 /* OK, so it's a subxact. Rollback to the savepoint. */
1463 int i;
1464 char spname[NAMEDATALEN];
1465
1466 pa_savepoint_name(MySubscription->oid, subxid, spname, sizeof(spname));
1467
1468 elog(DEBUG1, "rolling back to savepoint %s in logical replication parallel apply worker", spname);
1469
1470 /*
1471 * Search the subxactlist, determine the offset tracked for the
1472 * subxact, and truncate the list.
1473 *
1474 * Note that for an empty sub-transaction we won't find the subxid
1475 * here.
1476 */
1477 for (i = list_length(subxactlist) - 1; i >= 0; i--)
1478 {
1480
1481 if (xid_tmp == subxid)
1482 {
1483 RollbackToSavepoint(spname);
1486 break;
1487 }
1488 }
1489 }
1490}
void pa_set_xact_state(ParallelApplyWorkerShared *wshared, ParallelTransState xact_state)
void pa_reset_subtrans(void)
void pa_unlock_transaction(TransactionId xid, LOCKMODE lockmode)
@ STATE_IDLE
List * list_truncate(List *list, int new_size)
Definition: list.c:631
#define AccessExclusiveLock
Definition: lockdefs.h:43
TimestampTz replorigin_session_origin_timestamp
Definition: origin.c:161
XLogRecPtr replorigin_session_origin_lsn
Definition: origin.c:160
static int list_length(const List *l)
Definition: pg_list.h:152
static ListCell * list_nth_cell(const List *list, int n)
Definition: pg_list.h:277
#define lfirst_xid(lc)
Definition: pg_list.h:175
void RollbackToSavepoint(const char *name)
Definition: xact.c:4559
bool EndTransactionBlock(bool chain)
Definition: xact.c:4036
void AbortCurrentTransaction(void)
Definition: xact.c:3443

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

Referenced by apply_handle_stream_abort().

◆ pa_switch_to_partial_serialize()

void pa_switch_to_partial_serialize ( ParallelApplyWorkerInfo winfo,
bool  stream_locked 
)

Definition at line 1211 of file applyparallelworker.c.

1213{
1214 ereport(LOG,
1215 (errmsg("logical replication apply worker will serialize the remaining changes of remote transaction %u to a file",
1216 winfo->shared->xid)));
1217
1218 /*
1219 * The parallel apply worker could be stuck for some reason (say waiting
1220 * on some lock by other backend), so stop trying to send data directly to
1221 * it and start serializing data to the file instead.
1222 */
1223 winfo->serialize_changes = true;
1224
1225 /* Initialize the stream fileset. */
1226 stream_start_internal(winfo->shared->xid, true);
1227
1228 /*
1229 * Acquires the stream lock if not already to make sure that the parallel
1230 * apply worker will wait for the leader to release the stream lock until
1231 * the end of the transaction.
1232 */
1233 if (!stream_locked)
1235
1237}
void pa_set_fileset_state(ParallelApplyWorkerShared *wshared, PartialFileSetState fileset_state)
void stream_start_internal(TransactionId xid, bool first_segment)
Definition: worker.c:1450

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 
)

◆ pa_unlock_transaction()

◆ pa_xact_finish()

void pa_xact_finish ( ParallelApplyWorkerInfo winfo,
XLogRecPtr  remote_lsn 
)

Definition at line 1618 of file applyparallelworker.c.

1619{
1621
1622 /*
1623 * Unlock the shared object lock so that parallel apply worker can
1624 * continue to receive and apply changes.
1625 */
1627
1628 /*
1629 * Wait for that worker to finish. This is necessary to maintain commit
1630 * order which avoids failures due to transaction dependencies and
1631 * deadlocks.
1632 */
1634
1635 if (!XLogRecPtrIsInvalid(remote_lsn))
1636 store_flush_position(remote_lsn, winfo->shared->last_commit_end);
1637
1638 pa_free_worker(winfo);
1639}
static void pa_free_worker(ParallelApplyWorkerInfo *winfo)
static void pa_wait_for_xact_finish(ParallelApplyWorkerInfo *winfo)
void store_flush_position(XLogRecPtr remote_lsn, XLogRecPtr local_lsn)
Definition: worker.c:3528

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 667 of file tablesync.c.

668{
669 switch (MyLogicalRepWorker->type)
670 {
672
673 /*
674 * Skip for parallel apply workers because they only operate on
675 * tables that are in a READY state. See pa_can_start() and
676 * should_apply_changes_for_rel().
677 */
678 break;
679
682 break;
683
684 case WORKERTYPE_APPLY:
686 break;
687
689 /* Should never happen. */
690 elog(ERROR, "Unknown worker type");
691 }
692}
static void process_syncing_tables_for_apply(XLogRecPtr current_lsn)
Definition: tablesync.c:418
static void process_syncing_tables_for_sync(XLogRecPtr current_lsn)
Definition: tablesync.c:295

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

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

◆ ReplicationOriginNameForLogicalRep()

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

Definition at line 426 of file worker.c.

428{
429 if (OidIsValid(relid))
430 {
431 /* Replication origin name for tablesync workers. */
432 snprintf(originname, szoriginname, "pg_%u_%u", suboid, relid);
433 }
434 else
435 {
436 /* Replication origin name for non-tablesync workers. */
437 snprintf(originname, szoriginname, "pg_%u", suboid);
438 }
439}

References OidIsValid, and snprintf.

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

◆ set_apply_error_context_origin()

void set_apply_error_context_origin ( char *  originname)

Definition at line 5148 of file worker.c.

5149{
5151 originname);
5152}
char * MemoryContextStrdup(MemoryContext context, const char *string)
Definition: mcxt.c:1683

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

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

◆ set_stream_options()

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

Definition at line 4433 of file worker.c.

4436{
4437 int server_version;
4438
4439 options->logical = true;
4440 options->startpoint = *origin_startpos;
4441 options->slotname = slotname;
4442
4444 options->proto.logical.proto_version =
4449
4450 options->proto.logical.publication_names = MySubscription->publications;
4451 options->proto.logical.binary = MySubscription->binary;
4452
4453 /*
4454 * Assign the appropriate option value for streaming option according to
4455 * the 'streaming' mode and the publisher's ability to support that mode.
4456 */
4457 if (server_version >= 160000 &&
4458 MySubscription->stream == LOGICALREP_STREAM_PARALLEL)
4459 {
4460 options->proto.logical.streaming_str = "parallel";
4462 }
4463 else if (server_version >= 140000 &&
4464 MySubscription->stream != LOGICALREP_STREAM_OFF)
4465 {
4466 options->proto.logical.streaming_str = "on";
4468 }
4469 else
4470 {
4471 options->proto.logical.streaming_str = NULL;
4473 }
4474
4475 options->proto.logical.twophase = false;
4476 options->proto.logical.origin = pstrdup(MySubscription->origin);
4477}
WalReceiverConn * LogRepWorkerWalRcvConn
Definition: worker.c:297
#define LOGICALREP_PROTO_STREAM_PARALLEL_VERSION_NUM
Definition: logicalproto.h:44
#define LOGICALREP_PROTO_STREAM_VERSION_NUM
Definition: logicalproto.h:42
#define LOGICALREP_PROTO_TWOPHASE_VERSION_NUM
Definition: logicalproto.h:43
#define LOGICALREP_PROTO_VERSION_NUM
Definition: logicalproto.h:41
char * pstrdup(const char *in)
Definition: mcxt.c:1696
static int server_version
Definition: pg_dumpall.c:110
#define walrcv_server_version(conn)
Definition: walreceiver.h:447

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

Referenced by run_apply_worker(), and run_tablesync_worker().

◆ SetupApplyOrSyncWorker()

void SetupApplyOrSyncWorker ( int  worker_slot)

Definition at line 4733 of file worker.c.

4734{
4735 /* Attach to slot */
4736 logicalrep_worker_attach(worker_slot);
4737
4739
4740 /* Setup signal handling */
4742 pqsignal(SIGTERM, die);
4744
4745 /*
4746 * We don't currently need any ResourceOwner in a walreceiver process, but
4747 * if we did, we could call CreateAuxProcessResourceOwner here.
4748 */
4749
4750 /* Initialise stats to a sanish value */
4753
4754 /* Load the libpq-specific functions */
4755 load_file("libpqwalreceiver", false);
4756
4758
4759 /*
4760 * Register a callback to reset the origin state before aborting any
4761 * pending transaction during shutdown (see ShutdownPostgres()). This will
4762 * avoid origin advancement for an in-complete transaction which could
4763 * otherwise lead to its loss as such a transaction won't be sent by the
4764 * server again.
4765 *
4766 * Note that even a LOG or DEBUG statement placed after setting the origin
4767 * state may process a shutdown signal before committing the current apply
4768 * operation. So, it is important to register such a callback here.
4769 */
4771
4772 /* Connect to the origin and start the replication. */
4773 elog(DEBUG1, "connecting to publisher using connection string \"%s\"",
4775
4776 /*
4777 * Setup callback for syscache so that we know when something changes in
4778 * the subscription relation state.
4779 */
4780 CacheRegisterSyscacheCallback(SUBSCRIPTIONRELMAP,
4782 (Datum) 0);
4783}
void InitializeLogRepWorker(void)
Definition: worker.c:4639
void BackgroundWorkerUnblockSignals(void)
Definition: bgworker.c:926
void load_file(const char *filename, bool restricted)
Definition: dfmgr.c:134
void SignalHandlerForConfigReload(SIGNAL_ARGS)
Definition: interrupt.c:61
void logicalrep_worker_attach(int slot)
Definition: launcher.c:704
#define die(msg)
pqsigfunc pqsignal(int signo, pqsigfunc func)
void invalidate_syncing_table_states(Datum arg, int cacheid, uint32 hashvalue)
Definition: tablesync.c:281
#define SIGHUP
Definition: win32_port.h:168

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

Referenced by ApplyWorkerMain(), and TablesyncWorkerMain().

◆ start_apply()

void start_apply ( XLogRecPtr  origin_startpos)

Definition at line 4502 of file worker.c.

4503{
4504 PG_TRY();
4505 {
4506 LogicalRepApplyLoop(origin_startpos);
4507 }
4508 PG_CATCH();
4509 {
4512 else
4513 {
4514 /*
4515 * Report the worker failed while applying changes. Abort the
4516 * current transaction so that the stats message is sent in an
4517 * idle state.
4518 */
4521
4522 PG_RE_THROW();
4523 }
4524 }
4525 PG_END_TRY();
4526}
static void LogicalRepApplyLoop(XLogRecPtr last_received)
Definition: worker.c:3570
void DisableSubscriptionAndExit(void)
Definition: worker.c:4807
#define PG_RE_THROW()
Definition: elog.h:412
#define PG_TRY(...)
Definition: elog.h:371
#define PG_END_TRY(...)
Definition: elog.h:396
#define PG_CATCH(...)
Definition: elog.h:381

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

Referenced by run_apply_worker(), and run_tablesync_worker().

◆ store_flush_position()

void store_flush_position ( XLogRecPtr  remote_lsn,
XLogRecPtr  local_lsn 
)

Definition at line 3528 of file worker.c.

3529{
3530 FlushPosition *flushpos;
3531
3532 /*
3533 * Skip for parallel apply workers, because the lsn_mapping is maintained
3534 * by the leader apply worker.
3535 */
3537 return;
3538
3539 /* Need to do this in permanent context */
3541
3542 /* Track commit lsn */
3543 flushpos = (FlushPosition *) palloc(sizeof(FlushPosition));
3544 flushpos->local_end = local_lsn;
3545 flushpos->remote_end = remote_lsn;
3546
3547 dlist_push_tail(&lsn_mapping, &flushpos->node);
3549}
static dlist_head lsn_mapping
Definition: worker.c:205
static void dlist_push_tail(dlist_head *head, dlist_node *node)
Definition: ilist.h:364
dlist_node node
Definition: worker.c:200
XLogRecPtr remote_end
Definition: worker.c:202
XLogRecPtr local_end
Definition: worker.c:201

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 4300 of file worker.c.

4301{
4302 char path[MAXPGPATH];
4303
4304 /* Delete the changes file. */
4305 changes_filename(path, subid, xid);
4307
4308 /* Delete the subxact file, if it exists. */
4309 subxact_filename(path, subid, xid);
4311}
static void subxact_filename(char *path, Oid subid, TransactionId xid)
Definition: worker.c:4279
void BufFileDeleteFileSet(FileSet *fileset, const char *name, bool missing_ok)
Definition: buffile.c:364

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 1450 of file worker.c.

1451{
1453
1454 /*
1455 * Initialize the worker's stream_fileset if we haven't yet. This will be
1456 * used for the entire duration of the worker so create it in a permanent
1457 * context. We create this on the very first streaming message from any
1458 * transaction and then use it for this and other streaming transactions.
1459 * Now, we could create a fileset at the start of the worker as well but
1460 * then we won't be sure that it will ever be used.
1461 */
1463 {
1464 MemoryContext oldctx;
1465
1467
1470
1471 MemoryContextSwitchTo(oldctx);
1472 }
1473
1474 /* Open the spool file for this transaction. */
1475 stream_open_file(MyLogicalRepWorker->subid, xid, first_segment);
1476
1477 /* If this is not the first segment, open existing subxact file. */
1478 if (!first_segment)
1480
1482}
static void stream_open_file(Oid subid, TransactionId xid, bool first_segment)
Definition: worker.c:4324
static void subxact_info_read(Oid subid, TransactionId xid)
Definition: worker.c:4150
void FileSetInit(FileSet *fileset)
Definition: fileset.c:52

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 1624 of file worker.c.

1625{
1626 /*
1627 * Serialize information about subxacts for the toplevel transaction, then
1628 * close the stream messages spool file.
1629 */
1632
1633 /* We must be in a valid transaction state */
1635
1636 /* Commit the per-stream transaction */
1638
1639 /* Reset per-stream context */
1641}
static void subxact_info_write(Oid subid, TransactionId xid)
Definition: worker.c:4101
static MemoryContext LogicalStreamingContext
Definition: worker.c:295

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 1764 of file tablesync.c.

1765{
1766 Relation rel;
1767 HeapTuple tup;
1768 bool nulls[Natts_pg_subscription];
1769 bool replaces[Natts_pg_subscription];
1770 Datum values[Natts_pg_subscription];
1771
1772 Assert(new_state == LOGICALREP_TWOPHASE_STATE_DISABLED ||
1773 new_state == LOGICALREP_TWOPHASE_STATE_PENDING ||
1774 new_state == LOGICALREP_TWOPHASE_STATE_ENABLED);
1775
1776 rel = table_open(SubscriptionRelationId, RowExclusiveLock);
1777 tup = SearchSysCacheCopy1(SUBSCRIPTIONOID, ObjectIdGetDatum(suboid));
1778 if (!HeapTupleIsValid(tup))
1779 elog(ERROR,
1780 "cache lookup failed for subscription oid %u",
1781 suboid);
1782
1783 /* Form a new tuple. */
1784 memset(values, 0, sizeof(values));
1785 memset(nulls, false, sizeof(nulls));
1786 memset(replaces, false, sizeof(replaces));
1787
1788 /* And update/set two_phase state */
1789 values[Anum_pg_subscription_subtwophasestate - 1] = CharGetDatum(new_state);
1790 replaces[Anum_pg_subscription_subtwophasestate - 1] = true;
1791
1792 tup = heap_modify_tuple(tup, RelationGetDescr(rel),
1793 values, nulls, replaces);
1794 CatalogTupleUpdate(rel, &tup->t_self, tup);
1795
1796 heap_freetuple(tup);
1798}
static Datum values[MAXATTR]
Definition: bootstrap.c:151
HeapTuple heap_modify_tuple(HeapTuple tuple, TupleDesc tupleDesc, const Datum *replValues, const bool *replIsnull, const bool *doReplace)
Definition: heaptuple.c:1210
void heap_freetuple(HeapTuple htup)
Definition: heaptuple.c:1435
#define HeapTupleIsValid(tuple)
Definition: htup.h:78
void CatalogTupleUpdate(Relation heapRel, ItemPointer otid, HeapTuple tup)
Definition: indexing.c:313
#define RowExclusiveLock
Definition: lockdefs.h:38
static Datum ObjectIdGetDatum(Oid X)
Definition: postgres.h:252
static Datum CharGetDatum(char X)
Definition: postgres.h:122
#define RelationGetDescr(relation)
Definition: rel.h:531
ItemPointerData t_self
Definition: htup.h:65
#define SearchSysCacheCopy1(cacheId, key1)
Definition: syscache.h:91
void table_close(Relation relation, LOCKMODE lockmode)
Definition: table.c:126
Relation table_open(Oid relationId, LOCKMODE lockmode)
Definition: table.c:40

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

Referenced by CreateSubscription(), and run_apply_worker().

Variable Documentation

◆ apply_error_context_stack

PGDLLIMPORT ErrorContextCallback* apply_error_context_stack
extern

Definition at line 289 of file worker.c.

Referenced by HandleParallelApplyMessage(), and LogicalRepApplyLoop().

◆ ApplyContext

◆ ApplyMessageContext

◆ in_remote_transaction

◆ InitializingApplyWorker

PGDLLIMPORT bool InitializingApplyWorker
extern

Definition at line 319 of file worker.c.

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

◆ LogRepWorkerWalRcvConn

◆ MyLogicalRepWorker

◆ MyParallelShared

◆ MySubscription