PostgreSQL Source Code  git master
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 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 328 of file worker_internal.h.

◆ isTablesyncWorker

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

Definition at line 330 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.

30 {
LogicalRepWorkerType
@ WORKERTYPE_UNKNOWN
@ WORKERTYPE_APPLY

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

1715 {
1716  bool started_tx = false;
1717  bool has_subrels = false;
1718 
1719  /* We need up-to-date sync state info for subscription tables here. */
1720  has_subrels = FetchTableStates(&started_tx);
1721 
1722  if (started_tx)
1723  {
1725  pgstat_report_stat(true);
1726  }
1727 
1728  /*
1729  * Return false when there are no tables in subscription or not all tables
1730  * are in ready state; true otherwise.
1731  */
1732  return has_subrels && (table_states_not_ready == NIL);
1733 }
#define NIL
Definition: pg_list.h:68
long pgstat_report_stat(bool force)
Definition: pgstat.c:579
static List * table_states_not_ready
Definition: tablesync.c:127
static bool FetchTableStates(bool *started_tx)
Definition: tablesync.c:1565
void CommitTransactionCommand(void)
Definition: xact.c:3093

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

3273 {
3275  LogicalRepMsgType saved_command;
3276 
3277  /*
3278  * Set the current command being applied. Since this function can be
3279  * called recursively when applying spooled changes, save the current
3280  * command.
3281  */
3282  saved_command = apply_error_callback_arg.command;
3284 
3285  switch (action)
3286  {
3287  case LOGICAL_REP_MSG_BEGIN:
3288  apply_handle_begin(s);
3289  break;
3290 
3293  break;
3294 
3297  break;
3298 
3301  break;
3302 
3305  break;
3306 
3309  break;
3310 
3313  break;
3314 
3315  case LOGICAL_REP_MSG_TYPE:
3316  apply_handle_type(s);
3317  break;
3318 
3321  break;
3322 
3324 
3325  /*
3326  * Logical replication does not use generic logical messages yet.
3327  * Although, it could be used by other applications that use this
3328  * output plugin.
3329  */
3330  break;
3331 
3334  break;
3335 
3338  break;
3339 
3342  break;
3343 
3346  break;
3347 
3350  break;
3351 
3354  break;
3355 
3358  break;
3359 
3362  break;
3363 
3366  break;
3367 
3368  default:
3369  ereport(ERROR,
3370  (errcode(ERRCODE_PROTOCOL_VIOLATION),
3371  errmsg("invalid logical replication message type \"??? (%d)\"", action)));
3372  }
3373 
3374  /* Reset the current command */
3375  apply_error_callback_arg.command = saved_command;
3376 }
static void apply_handle_stream_prepare(StringInfo s)
Definition: worker.c:1271
static void apply_handle_type(StringInfo s)
Definition: worker.c:2324
static void apply_handle_truncate(StringInfo s)
Definition: worker.c:3144
static void apply_handle_update(StringInfo s)
Definition: worker.c:2520
static void apply_handle_stream_commit(StringInfo s)
Definition: worker.c:2131
static void apply_handle_commit_prepared(StringInfo s)
Definition: worker.c:1169
ApplyErrorCallbackArg apply_error_callback_arg
Definition: worker.c:278
static void apply_handle_delete(StringInfo s)
Definition: worker.c:2704
static void apply_handle_begin(StringInfo s)
Definition: worker.c:991
static void apply_handle_commit(StringInfo s)
Definition: worker.c:1016
static void apply_handle_stream_abort(StringInfo s)
Definition: worker.c:1812
static void apply_handle_relation(StringInfo s)
Definition: worker.c:2301
static void apply_handle_prepare(StringInfo s)
Definition: worker.c:1108
static void apply_handle_rollback_prepared(StringInfo s)
Definition: worker.c:1218
static void apply_handle_stream_stop(StringInfo s)
Definition: worker.c:1626
static void apply_handle_origin(StringInfo s)
Definition: worker.c:1408
static void apply_handle_begin_prepare(StringInfo s)
Definition: worker.c:1042
static void apply_handle_stream_start(StringInfo s)
Definition: worker.c:1467
static void apply_handle_insert(StringInfo s)
Definition: worker.c:2371
int errcode(int sqlerrcode)
Definition: elog.c:859
int errmsg(const char *fmt,...)
Definition: elog.c:1072
#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:221

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

4894 {
4896 
4898  return;
4899 
4900  Assert(errarg->origin_name);
4901 
4902  if (errarg->rel == NULL)
4903  {
4904  if (!TransactionIdIsValid(errarg->remote_xid))
4905  errcontext("processing remote data for replication origin \"%s\" during message type \"%s\"",
4906  errarg->origin_name,
4907  logicalrep_message_type(errarg->command));
4908  else if (XLogRecPtrIsInvalid(errarg->finish_lsn))
4909  errcontext("processing remote data for replication origin \"%s\" during message type \"%s\" in transaction %u",
4910  errarg->origin_name,
4912  errarg->remote_xid);
4913  else
4914  errcontext("processing remote data for replication origin \"%s\" during message type \"%s\" in transaction %u, finished at %X/%X",
4915  errarg->origin_name,
4917  errarg->remote_xid,
4918  LSN_FORMAT_ARGS(errarg->finish_lsn));
4919  }
4920  else
4921  {
4922  if (errarg->remote_attnum < 0)
4923  {
4924  if (XLogRecPtrIsInvalid(errarg->finish_lsn))
4925  errcontext("processing remote data for replication origin \"%s\" during message type \"%s\" for replication target relation \"%s.%s\" in transaction %u",
4926  errarg->origin_name,
4928  errarg->rel->remoterel.nspname,
4929  errarg->rel->remoterel.relname,
4930  errarg->remote_xid);
4931  else
4932  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",
4933  errarg->origin_name,
4935  errarg->rel->remoterel.nspname,
4936  errarg->rel->remoterel.relname,
4937  errarg->remote_xid,
4938  LSN_FORMAT_ARGS(errarg->finish_lsn));
4939  }
4940  else
4941  {
4942  if (XLogRecPtrIsInvalid(errarg->finish_lsn))
4943  errcontext("processing remote data for replication origin \"%s\" during message type \"%s\" for replication target relation \"%s.%s\" column \"%s\" in transaction %u",
4944  errarg->origin_name,
4946  errarg->rel->remoterel.nspname,
4947  errarg->rel->remoterel.relname,
4948  errarg->rel->remoterel.attnames[errarg->remote_attnum],
4949  errarg->remote_xid);
4950  else
4951  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",
4952  errarg->origin_name,
4954  errarg->rel->remoterel.nspname,
4955  errarg->rel->remoterel.relname,
4956  errarg->rel->remoterel.attnames[errarg->remote_attnum],
4957  errarg->remote_xid,
4958  LSN_FORMAT_ARGS(errarg->finish_lsn));
4959  }
4960  }
4961 }
#define errcontext
Definition: elog.h:196
const char * logicalrep_message_type(LogicalRepMsgType action)
Definition: proto.c:1217
TransactionId remote_xid
Definition: worker.c:226
XLogRecPtr finish_lsn
Definition: worker.c:227
LogicalRepRelMapEntry * rel
Definition: worker.c:222
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, 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 2001 of file worker.c.

2003 {
2004  int nchanges;
2005  char path[MAXPGPATH];
2006  char *buffer = NULL;
2007  MemoryContext oldcxt;
2008  ResourceOwner oldowner;
2009  int fileno;
2010  off_t offset;
2011 
2012  if (!am_parallel_apply_worker())
2014 
2015  /* Make sure we have an open transaction */
2017 
2018  /*
2019  * Allocate file handle and memory required to process all the messages in
2020  * TopTransactionContext to avoid them getting reset after each message is
2021  * processed.
2022  */
2024 
2025  /* Open the spool file for the committed/prepared transaction */
2027  elog(DEBUG1, "replaying changes from file \"%s\"", path);
2028 
2029  /*
2030  * Make sure the file is owned by the toplevel transaction so that the
2031  * file will not be accidentally closed when aborting a subtransaction.
2032  */
2033  oldowner = CurrentResourceOwner;
2035 
2036  stream_fd = BufFileOpenFileSet(stream_fileset, path, O_RDONLY, false);
2037 
2038  CurrentResourceOwner = oldowner;
2039 
2040  buffer = palloc(BLCKSZ);
2041 
2042  MemoryContextSwitchTo(oldcxt);
2043 
2044  remote_final_lsn = lsn;
2045 
2046  /*
2047  * Make sure the handle apply_dispatch methods are aware we're in a remote
2048  * transaction.
2049  */
2050  in_remote_transaction = true;
2052 
2054 
2055  /*
2056  * Read the entries one by one and pass them through the same logic as in
2057  * apply_dispatch.
2058  */
2059  nchanges = 0;
2060  while (true)
2061  {
2063  size_t nbytes;
2064  int len;
2065 
2067 
2068  /* read length of the on-disk record */
2069  nbytes = BufFileReadMaybeEOF(stream_fd, &len, sizeof(len), true);
2070 
2071  /* have we reached end of the file? */
2072  if (nbytes == 0)
2073  break;
2074 
2075  /* do we have a correct length? */
2076  if (len <= 0)
2077  elog(ERROR, "incorrect length %d in streaming transaction's changes file \"%s\"",
2078  len, path);
2079 
2080  /* make sure we have sufficiently large buffer */
2081  buffer = repalloc(buffer, len);
2082 
2083  /* and finally read the data into the buffer */
2084  BufFileReadExact(stream_fd, buffer, len);
2085 
2086  BufFileTell(stream_fd, &fileno, &offset);
2087 
2088  /* init a stringinfo using the buffer and call apply_dispatch */
2089  initReadOnlyStringInfo(&s2, buffer, len);
2090 
2091  /* Ensure we are reading the data into our memory context. */
2093 
2094  apply_dispatch(&s2);
2095 
2097 
2098  MemoryContextSwitchTo(oldcxt);
2099 
2100  nchanges++;
2101 
2102  /*
2103  * It is possible the file has been closed because we have processed
2104  * the transaction end message like stream_commit in which case that
2105  * must be the last message.
2106  */
2107  if (!stream_fd)
2108  {
2109  ensure_last_message(stream_fileset, xid, fileno, offset);
2110  break;
2111  }
2112 
2113  if (nchanges % 1000 == 0)
2114  elog(DEBUG1, "replayed %d changes from file \"%s\"",
2115  nchanges, path);
2116  }
2117 
2118  if (stream_fd)
2120 
2121  elog(DEBUG1, "replayed %d (all) changes from file \"%s\"",
2122  nchanges, path);
2123 
2124  return;
2125 }
static void begin_replication_step(void)
Definition: worker.c:508
static void end_replication_step(void)
Definition: worker.c:531
MemoryContext ApplyMessageContext
Definition: worker.c:290
static void changes_filename(char *path, Oid subid, TransactionId xid)
Definition: worker.c:4191
static BufFile * stream_fd
Definition: worker.c:339
bool in_remote_transaction
Definition: worker.c:303
void apply_dispatch(StringInfo s)
Definition: worker.c:3272
static void ensure_last_message(FileSet *stream_fileset, TransactionId xid, int fileno, off_t offset)
Definition: worker.c:1969
static XLogRecPtr remote_final_lsn
Definition: worker.c:304
static void maybe_start_skipping_changes(XLogRecPtr finish_lsn)
Definition: worker.c:4763
static void stream_close_file(void)
Definition: worker.c:4274
void pgstat_report_activity(BackendState state, const char *cmd_str)
@ STATE_RUNNING
void BufFileReadExact(BufFile *file, void *ptr, size_t size)
Definition: buffile.c:654
BufFile * BufFileOpenFileSet(FileSet *fileset, const char *name, int mode, bool missing_ok)
Definition: buffile.c:291
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:224
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:1540
void * palloc(Size size)
Definition: mcxt.c:1316
#define CHECK_FOR_INTERRUPTS()
Definition: miscadmin.h:122
#define MAXPGPATH
const void size_t len
char * s2
MemoryContextSwitchTo(old_ctx)
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 4705 of file worker.c.

4706 {
4707  /*
4708  * Emit the error message, and recover from the error state to an idle
4709  * state
4710  */
4711  HOLD_INTERRUPTS();
4712 
4713  EmitErrorReport();
4715  FlushErrorState();
4716 
4718 
4719  /* Report the worker failed during either table synchronization or apply */
4721  !am_tablesync_worker());
4722 
4723  /* Disable the subscription */
4727 
4728  /* Ensure we remove no-longer-useful entry for worker's start time */
4729  if (am_leader_apply_worker())
4731 
4732  /* Notify the subscription has been disabled and exit */
4733  ereport(LOG,
4734  errmsg("subscription \"%s\" has been disabled because of an error",
4735  MySubscription->name));
4736 
4737  proc_exit(0);
4738 }
Subscription * MySubscription
Definition: worker.c:298
void EmitErrorReport(void)
Definition: elog.c:1672
void FlushErrorState(void)
Definition: elog.c:1836
#define LOG
Definition: elog.h:31
void proc_exit(int code)
Definition: ipc.c:104
void ApplyLauncherForgetWorkerStartTime(Oid subid)
Definition: launcher.c:1075
#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:2995
void AbortOutOfAnyTransaction(void)
Definition: xact.c:4824

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

4562 {
4563  MemoryContext oldctx;
4564 
4565  /* Run as replica session replication role. */
4566  SetConfigOption("session_replication_role", "replica",
4568 
4569  /* Connect to our database. */
4572  0);
4573 
4574  /*
4575  * Set always-secure search path, so malicious users can't redirect user
4576  * code (e.g. pg_index.indexprs).
4577  */
4578  SetConfigOption("search_path", "", PGC_SUSET, PGC_S_OVERRIDE);
4579 
4580  /* Load the subscription into persistent memory context. */
4582  "ApplyContext",
4586 
4588  if (!MySubscription)
4589  {
4590  ereport(LOG,
4591  (errmsg("logical replication worker for subscription %u will not start because the subscription was removed during startup",
4593 
4594  /* Ensure we remove no-longer-useful entry for worker's start time */
4595  if (am_leader_apply_worker())
4597 
4598  proc_exit(0);
4599  }
4600 
4601  MySubscriptionValid = true;
4602  MemoryContextSwitchTo(oldctx);
4603 
4604  if (!MySubscription->enabled)
4605  {
4606  ereport(LOG,
4607  (errmsg("logical replication worker for subscription \"%s\" will not start because the subscription was disabled during startup",
4608  MySubscription->name)));
4609 
4611  }
4612 
4613  /* Setup synchronous commit according to the user's wishes */
4614  SetConfigOption("synchronous_commit", MySubscription->synccommit,
4616 
4617  /*
4618  * Keep us informed about subscription or role changes. Note that the
4619  * role's superuser privilege can be revoked.
4620  */
4621  CacheRegisterSyscacheCallback(SUBSCRIPTIONOID,
4623  (Datum) 0);
4624 
4627  (Datum) 0);
4628 
4629  if (am_tablesync_worker())
4630  ereport(LOG,
4631  (errmsg("logical replication table synchronization worker for subscription \"%s\", table \"%s\" has started",
4634  else
4635  ereport(LOG,
4636  (errmsg("logical replication apply worker for subscription \"%s\" has started",
4637  MySubscription->name)));
4638 
4640 }
static void subscription_change_cb(Datum arg, int cacheid, uint32 hashvalue)
Definition: worker.c:3991
static void apply_worker_exit(void)
Definition: worker.c:3831
MemoryContext ApplyContext
Definition: worker.c:291
static bool MySubscriptionValid
Definition: worker.c:299
void SetConfigOption(const char *name, const char *value, GucContext context, GucSource source)
Definition: guc.c:4275
@ 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:1516
char * get_rel_name(Oid relid)
Definition: lsyscache.c:1906
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)
uintptr_t Datum
Definition: postgres.h:64
void BackgroundWorkerInitializeConnectionByOid(Oid dboid, Oid useroid, uint32 flags)
Definition: postmaster.c:4189

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

275 {
276  table_states_valid = false;
277 }
static bool table_states_valid
Definition: tablesync.c:126

References table_states_valid.

Referenced by ParallelApplyWorkerMain(), and SetupApplyOrSyncWorker().

◆ logicalrep_pa_worker_stop()

void logicalrep_pa_worker_stop ( ParallelApplyWorkerInfo winfo)

Definition at line 633 of file launcher.c.

634 {
635  int slot_no;
636  uint16 generation;
637  LogicalRepWorker *worker;
638 
639  SpinLockAcquire(&winfo->shared->mutex);
640  generation = winfo->shared->logicalrep_worker_generation;
641  slot_no = winfo->shared->logicalrep_worker_slot_no;
642  SpinLockRelease(&winfo->shared->mutex);
643 
644  Assert(slot_no >= 0 && slot_no < max_logical_replication_workers);
645 
646  /*
647  * Detach from the error_mq_handle for the parallel apply worker before
648  * stopping it. This prevents the leader apply worker from trying to
649  * receive the message from the error queue that might already be detached
650  * by the parallel apply worker.
651  */
652  if (winfo->error_mq_handle)
653  {
655  winfo->error_mq_handle = NULL;
656  }
657 
658  LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
659 
660  worker = &LogicalRepCtx->workers[slot_no];
661  Assert(isParallelApplyWorker(worker));
662 
663  /*
664  * Only stop the worker if the generation matches and the worker is alive.
665  */
666  if (worker->generation == generation && worker->proc)
667  logicalrep_worker_stop_internal(worker, SIGINT);
668 
669  LWLockRelease(LogicalRepWorkerLock);
670 }
unsigned short uint16
Definition: c.h:505
int max_logical_replication_workers
Definition: launcher.c:50
static void logicalrep_worker_stop_internal(LogicalRepWorker *worker, int signo)
Definition: launcher.c:527
static LogicalRepCtxStruct * LogicalRepCtx
Definition: launcher.c:69
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1170
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1783
@ 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:64
#define SpinLockAcquire(lock)
Definition: spin.h:62
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 848 of file launcher.c.

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

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

708 {
709  /* Block concurrent access. */
710  LWLockAcquire(LogicalRepWorkerLock, LW_EXCLUSIVE);
711 
712  Assert(slot >= 0 && slot < max_logical_replication_workers);
714 
716  {
717  LWLockRelease(LogicalRepWorkerLock);
718  ereport(ERROR,
719  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
720  errmsg("logical replication worker slot %d is empty, cannot attach",
721  slot)));
722  }
723 
725  {
726  LWLockRelease(LogicalRepWorkerLock);
727  ereport(ERROR,
728  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
729  errmsg("logical replication worker slot %d is already used by "
730  "another worker, cannot attach", slot)));
731  }
732 
735 
736  LWLockRelease(LogicalRepWorkerLock);
737 }
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:818
@ 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 243 of file launcher.c.

244 {
245  int i;
246  LogicalRepWorker *res = NULL;
247 
248  Assert(LWLockHeldByMe(LogicalRepWorkerLock));
249 
250  /* Search for attached worker for a given subscription id. */
251  for (i = 0; i < max_logical_replication_workers; i++)
252  {
254 
255  /* Skip parallel apply workers. */
256  if (isParallelApplyWorker(w))
257  continue;
258 
259  if (w->in_use && w->subid == subid && w->relid == relid &&
260  (!only_running || w->proc))
261  {
262  res = w;
263  break;
264  }
265  }
266 
267  return res;
268 }

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

303 {
304  BackgroundWorker bgw;
305  BackgroundWorkerHandle *bgw_handle;
306  uint16 generation;
307  int i;
308  int slot = 0;
309  LogicalRepWorker *worker = NULL;
310  int nsyncworkers;
311  int nparallelapplyworkers;
313  bool is_tablesync_worker = (wtype == WORKERTYPE_TABLESYNC);
314  bool is_parallel_apply_worker = (wtype == WORKERTYPE_PARALLEL_APPLY);
315 
316  /*----------
317  * Sanity checks:
318  * - must be valid worker type
319  * - tablesync workers are only ones to have relid
320  * - parallel apply worker is the only kind of subworker
321  */
322  Assert(wtype != WORKERTYPE_UNKNOWN);
323  Assert(is_tablesync_worker == OidIsValid(relid));
324  Assert(is_parallel_apply_worker == (subworker_dsm != DSM_HANDLE_INVALID));
325 
326  ereport(DEBUG1,
327  (errmsg_internal("starting logical replication worker for subscription \"%s\"",
328  subname)));
329 
330  /* Report this after the initial starting message for consistency. */
331  if (max_replication_slots == 0)
332  ereport(ERROR,
333  (errcode(ERRCODE_CONFIGURATION_LIMIT_EXCEEDED),
334  errmsg("cannot start logical replication workers when max_replication_slots = 0")));
335 
336  /*
337  * We need to do the modification of the shared memory under lock so that
338  * we have consistent view.
339  */
340  LWLockAcquire(LogicalRepWorkerLock, LW_EXCLUSIVE);
341 
342 retry:
343  /* Find unused worker slot. */
344  for (i = 0; i < max_logical_replication_workers; i++)
345  {
347 
348  if (!w->in_use)
349  {
350  worker = w;
351  slot = i;
352  break;
353  }
354  }
355 
356  nsyncworkers = logicalrep_sync_worker_count(subid);
357 
359 
360  /*
361  * If we didn't find a free slot, try to do garbage collection. The
362  * reason we do this is because if some worker failed to start up and its
363  * parent has crashed while waiting, the in_use state was never cleared.
364  */
365  if (worker == NULL || nsyncworkers >= max_sync_workers_per_subscription)
366  {
367  bool did_cleanup = false;
368 
369  for (i = 0; i < max_logical_replication_workers; i++)
370  {
372 
373  /*
374  * If the worker was marked in use but didn't manage to attach in
375  * time, clean it up.
376  */
377  if (w->in_use && !w->proc &&
380  {
381  elog(WARNING,
382  "logical replication worker for subscription %u took too long to start; canceled",
383  w->subid);
384 
386  did_cleanup = true;
387  }
388  }
389 
390  if (did_cleanup)
391  goto retry;
392  }
393 
394  /*
395  * We don't allow to invoke more sync workers once we have reached the
396  * sync worker limit per subscription. So, just return silently as we
397  * might get here because of an otherwise harmless race condition.
398  */
399  if (is_tablesync_worker && nsyncworkers >= max_sync_workers_per_subscription)
400  {
401  LWLockRelease(LogicalRepWorkerLock);
402  return false;
403  }
404 
405  nparallelapplyworkers = logicalrep_pa_worker_count(subid);
406 
407  /*
408  * Return false if the number of parallel apply workers reached the limit
409  * per subscription.
410  */
411  if (is_parallel_apply_worker &&
412  nparallelapplyworkers >= max_parallel_apply_workers_per_subscription)
413  {
414  LWLockRelease(LogicalRepWorkerLock);
415  return false;
416  }
417 
418  /*
419  * However if there are no more free worker slots, inform user about it
420  * before exiting.
421  */
422  if (worker == NULL)
423  {
424  LWLockRelease(LogicalRepWorkerLock);
426  (errcode(ERRCODE_CONFIGURATION_LIMIT_EXCEEDED),
427  errmsg("out of logical replication worker slots"),
428  errhint("You might need to increase %s.", "max_logical_replication_workers")));
429  return false;
430  }
431 
432  /* Prepare the worker slot. */
433  worker->type = wtype;
434  worker->launch_time = now;
435  worker->in_use = true;
436  worker->generation++;
437  worker->proc = NULL;
438  worker->dbid = dbid;
439  worker->userid = userid;
440  worker->subid = subid;
441  worker->relid = relid;
442  worker->relstate = SUBREL_STATE_UNKNOWN;
444  worker->stream_fileset = NULL;
445  worker->leader_pid = is_parallel_apply_worker ? MyProcPid : InvalidPid;
446  worker->parallel_apply = is_parallel_apply_worker;
447  worker->last_lsn = InvalidXLogRecPtr;
450  worker->reply_lsn = InvalidXLogRecPtr;
451  TIMESTAMP_NOBEGIN(worker->reply_time);
452 
453  /* Before releasing lock, remember generation for future identification. */
454  generation = worker->generation;
455 
456  LWLockRelease(LogicalRepWorkerLock);
457 
458  /* Register the new dynamic worker. */
459  memset(&bgw, 0, sizeof(bgw));
463  snprintf(bgw.bgw_library_name, MAXPGPATH, "postgres");
464 
465  switch (worker->type)
466  {
467  case WORKERTYPE_APPLY:
468  snprintf(bgw.bgw_function_name, BGW_MAXLEN, "ApplyWorkerMain");
470  "logical replication apply worker for subscription %u",
471  subid);
472  snprintf(bgw.bgw_type, BGW_MAXLEN, "logical replication apply worker");
473  break;
474 
476  snprintf(bgw.bgw_function_name, BGW_MAXLEN, "ParallelApplyWorkerMain");
478  "logical replication parallel apply worker for subscription %u",
479  subid);
480  snprintf(bgw.bgw_type, BGW_MAXLEN, "logical replication parallel worker");
481 
482  memcpy(bgw.bgw_extra, &subworker_dsm, sizeof(dsm_handle));
483  break;
484 
486  snprintf(bgw.bgw_function_name, BGW_MAXLEN, "TablesyncWorkerMain");
488  "logical replication tablesync worker for subscription %u sync %u",
489  subid,
490  relid);
491  snprintf(bgw.bgw_type, BGW_MAXLEN, "logical replication tablesync worker");
492  break;
493 
494  case WORKERTYPE_UNKNOWN:
495  /* Should never happen. */
496  elog(ERROR, "unknown worker type");
497  }
498 
501  bgw.bgw_main_arg = Int32GetDatum(slot);
502 
503  if (!RegisterDynamicBackgroundWorker(&bgw, &bgw_handle))
504  {
505  /* Failed to start worker, so clean up the worker slot. */
506  LWLockAcquire(LogicalRepWorkerLock, LW_EXCLUSIVE);
507  Assert(generation == worker->generation);
509  LWLockRelease(LogicalRepWorkerLock);
510 
512  (errcode(ERRCODE_CONFIGURATION_LIMIT_EXCEEDED),
513  errmsg("out of background worker slots"),
514  errhint("You might need to increase %s.", "max_worker_processes")));
515  return false;
516  }
517 
518  /* Now wait until it attaches. */
519  return WaitForReplicationWorkerAttach(worker, generation, bgw_handle);
520 }
bool TimestampDifferenceExceeds(TimestampTz start_time, TimestampTz stop_time, int msec)
Definition: timestamp.c:1790
TimestampTz GetCurrentTimestamp(void)
Definition: timestamp.c:1654
Datum now(PG_FUNCTION_ARGS)
Definition: timestamp.c:1618
bool RegisterDynamicBackgroundWorker(BackgroundWorker *worker, BackgroundWorkerHandle **handle)
Definition: bgworker.c:970
#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:775
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:1159
int errhint(const char *fmt,...)
Definition: elog.c:1319
#define WARNING
Definition: elog.h:36
int MyProcPid
Definition: globals.c:45
static int logicalrep_pa_worker_count(Oid subid)
Definition: launcher.c:872
int max_sync_workers_per_subscription
Definition: launcher.c:51
static bool WaitForReplicationWorkerAttach(LogicalRepWorker *worker, uint16 generation, BackgroundWorkerHandle *handle)
Definition: launcher.c:183
int logicalrep_sync_worker_count(Oid subid)
Definition: launcher.c:848
int max_parallel_apply_workers_per_subscription
Definition: launcher.c:52
static void logicalrep_worker_cleanup(LogicalRepWorker *worker)
Definition: launcher.c:786
#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 609 of file launcher.c.

610 {
611  LogicalRepWorker *worker;
612 
613  LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
614 
615  worker = logicalrep_worker_find(subid, relid, false);
616 
617  if (worker)
618  {
619  Assert(!isParallelApplyWorker(worker));
620  logicalrep_worker_stop_internal(worker, SIGTERM);
621  }
622 
623  LWLockRelease(LogicalRepWorkerLock);
624 }
LogicalRepWorker * logicalrep_worker_find(Oid subid, Oid relid, bool only_running)
Definition: launcher.c:243

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

677 {
678  LogicalRepWorker *worker;
679 
680  LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
681 
682  worker = logicalrep_worker_find(subid, relid, true);
683 
684  if (worker)
686 
687  LWLockRelease(LogicalRepWorkerLock);
688 }
void logicalrep_worker_wakeup_ptr(LogicalRepWorker *worker)
Definition: launcher.c:696

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

697 {
698  Assert(LWLockHeldByMe(LogicalRepWorkerLock));
699 
700  SetLatch(&worker->proc->procLatch);
701 }
void SetLatch(Latch *latch)
Definition: latch.c:632
Latch procLatch
Definition: proc.h:165

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

276 {
277  int i;
278  List *res = NIL;
279 
280  Assert(LWLockHeldByMe(LogicalRepWorkerLock));
281 
282  /* Search for attached worker for a given subscription id. */
283  for (i = 0; i < max_logical_replication_workers; i++)
284  {
286 
287  if (w->in_use && w->subid == subid && (!only_running || w->proc))
288  res = lappend(res, w);
289  }
290 
291  return res;
292 }
List * lappend(List *list, void *datum)
Definition: list.c:339
Definition: pg_list.h:54

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

◆ maybe_reread_subscription()

void maybe_reread_subscription ( void  )

Definition at line 3862 of file worker.c.

3863 {
3864  MemoryContext oldctx;
3866  bool started_tx = false;
3867 
3868  /* When cache state is valid there is nothing to do here. */
3869  if (MySubscriptionValid)
3870  return;
3871 
3872  /* This function might be called inside or outside of transaction. */
3873  if (!IsTransactionState())
3874  {
3876  started_tx = true;
3877  }
3878 
3879  /* Ensure allocations in permanent context. */
3881 
3883 
3884  /*
3885  * Exit if the subscription was removed. This normally should not happen
3886  * as the worker gets killed during DROP SUBSCRIPTION.
3887  */
3888  if (!newsub)
3889  {
3890  ereport(LOG,
3891  (errmsg("logical replication worker for subscription \"%s\" will stop because the subscription was removed",
3892  MySubscription->name)));
3893 
3894  /* Ensure we remove no-longer-useful entry for worker's start time */
3895  if (am_leader_apply_worker())
3897 
3898  proc_exit(0);
3899  }
3900 
3901  /* Exit if the subscription was disabled. */
3902  if (!newsub->enabled)
3903  {
3904  ereport(LOG,
3905  (errmsg("logical replication worker for subscription \"%s\" will stop because the subscription was disabled",
3906  MySubscription->name)));
3907 
3909  }
3910 
3911  /* !slotname should never happen when enabled is true. */
3912  Assert(newsub->slotname);
3913 
3914  /* two-phase should not be altered */
3915  Assert(newsub->twophasestate == MySubscription->twophasestate);
3916 
3917  /*
3918  * Exit if any parameter that affects the remote connection was changed.
3919  * The launcher will start a new worker but note that the parallel apply
3920  * worker won't restart if the streaming option's value is changed from
3921  * 'parallel' to any other value or the server decides not to stream the
3922  * in-progress transaction.
3923  */
3924  if (strcmp(newsub->conninfo, MySubscription->conninfo) != 0 ||
3925  strcmp(newsub->name, MySubscription->name) != 0 ||
3926  strcmp(newsub->slotname, MySubscription->slotname) != 0 ||
3927  newsub->binary != MySubscription->binary ||
3928  newsub->stream != MySubscription->stream ||
3929  newsub->passwordrequired != MySubscription->passwordrequired ||
3930  strcmp(newsub->origin, MySubscription->origin) != 0 ||
3931  newsub->owner != MySubscription->owner ||
3932  !equal(newsub->publications, MySubscription->publications))
3933  {
3935  ereport(LOG,
3936  (errmsg("logical replication parallel apply worker for subscription \"%s\" will stop because of a parameter change",
3937  MySubscription->name)));
3938  else
3939  ereport(LOG,
3940  (errmsg("logical replication worker for subscription \"%s\" will restart because of a parameter change",
3941  MySubscription->name)));
3942 
3944  }
3945 
3946  /*
3947  * Exit if the subscription owner's superuser privileges have been
3948  * revoked.
3949  */
3950  if (!newsub->ownersuperuser && MySubscription->ownersuperuser)
3951  {
3953  ereport(LOG,
3954  errmsg("logical replication parallel apply worker for subscription \"%s\" will stop because the subscription owner's superuser privileges have been revoked",
3955  MySubscription->name));
3956  else
3957  ereport(LOG,
3958  errmsg("logical replication worker for subscription \"%s\" will restart because the subscription owner's superuser privileges have been revoked",
3959  MySubscription->name));
3960 
3962  }
3963 
3964  /* Check for other changes that should never happen too. */
3965  if (newsub->dbid != MySubscription->dbid)
3966  {
3967  elog(ERROR, "subscription %u changed unexpectedly",
3969  }
3970 
3971  /* Clean old subscription info and switch to new one. */
3974 
3975  MemoryContextSwitchTo(oldctx);
3976 
3977  /* Change synchronous commit according to the user's wishes */
3978  SetConfigOption("synchronous_commit", MySubscription->synccommit,
3980 
3981  if (started_tx)
3983 
3984  MySubscriptionValid = true;
3985 }
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:384

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 
479  winfo = pa_launch_parallel_worker();
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 ParallelApplyWorkerInfo * pa_launch_parallel_worker(void)
static HTAB * ParallelApplyTxnHash
#define MemSet(start, val, len)
Definition: c.h:1020
uint32 TransactionId
Definition: c.h:652
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:1847
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:434
static uint32 pg_atomic_read_u32(volatile pg_atomic_uint32 *ptr)
Definition: atomics.h:234
#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:311
struct Latch * MyLatch
Definition: globals.c:60
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:29
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:4335
bool IsTransactionBlock(void)
Definition: xact.c:4928
void BeginTransactionBlock(void)
Definition: xact.c:3886

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:157
XLogRecPtr replorigin_session_origin_lsn
Definition: origin.c:156
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:4529
bool EndTransactionBlock(bool chain)
Definition: xact.c:4006
void AbortCurrentTransaction(void)
Definition: xact.c:3391

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:1429

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  */
1633  pa_wait_for_xact_finish(winfo);
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:3436

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

661 {
662  switch (MyLogicalRepWorker->type)
663  {
665 
666  /*
667  * Skip for parallel apply workers because they only operate on
668  * tables that are in a READY state. See pa_can_start() and
669  * should_apply_changes_for_rel().
670  */
671  break;
672 
674  process_syncing_tables_for_sync(current_lsn);
675  break;
676 
677  case WORKERTYPE_APPLY:
679  break;
680 
681  case WORKERTYPE_UNKNOWN:
682  /* Should never happen. */
683  elog(ERROR, "Unknown worker type");
684  }
685 }
static void process_syncing_tables_for_apply(XLogRecPtr current_lsn)
Definition: tablesync.c:411
static void process_syncing_tables_for_sync(XLogRecPtr current_lsn)
Definition: tablesync.c:288

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

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

◆ ReplicationOriginNameForLogicalRep()

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

Definition at line 428 of file worker.c.

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

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

5036 {
5038  originname);
5039 }
char * MemoryContextStrdup(MemoryContext context, const char *string)
Definition: mcxt.c:1682

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

4341 {
4342  int server_version;
4343 
4344  options->logical = true;
4345  options->startpoint = *origin_startpos;
4346  options->slotname = slotname;
4347 
4349  options->proto.logical.proto_version =
4354 
4355  options->proto.logical.publication_names = MySubscription->publications;
4356  options->proto.logical.binary = MySubscription->binary;
4357 
4358  /*
4359  * Assign the appropriate option value for streaming option according to
4360  * the 'streaming' mode and the publisher's ability to support that mode.
4361  */
4362  if (server_version >= 160000 &&
4364  {
4365  options->proto.logical.streaming_str = "parallel";
4367  }
4368  else if (server_version >= 140000 &&
4370  {
4371  options->proto.logical.streaming_str = "on";
4373  }
4374  else
4375  {
4376  options->proto.logical.streaming_str = NULL;
4378  }
4379 
4380  options->proto.logical.twophase = false;
4381  options->proto.logical.origin = pstrdup(MySubscription->origin);
4382 }
WalReceiverConn * LogRepWorkerWalRcvConn
Definition: worker.c:296
#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:1695
static int server_version
Definition: pg_dumpall.c:110
#define LOGICALREP_STREAM_OFF
#define LOGICALREP_STREAM_PARALLEL
#define walrcv_server_version(conn)
Definition: walreceiver.h:444

References Subscription::binary, LOGICALREP_PROTO_STREAM_PARALLEL_VERSION_NUM, LOGICALREP_PROTO_STREAM_VERSION_NUM, LOGICALREP_PROTO_TWOPHASE_VERSION_NUM, LOGICALREP_PROTO_VERSION_NUM, LOGICALREP_STREAM_OFF, LOGICALREP_STREAM_PARALLEL, 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 4644 of file worker.c.

4645 {
4646  /* Attach to slot */
4647  logicalrep_worker_attach(worker_slot);
4648 
4650 
4651  /* Setup signal handling */
4653  pqsignal(SIGTERM, die);
4655 
4656  /*
4657  * We don't currently need any ResourceOwner in a walreceiver process, but
4658  * if we did, we could call CreateAuxProcessResourceOwner here.
4659  */
4660 
4661  /* Initialise stats to a sanish value */
4664 
4665  /* Load the libpq-specific functions */
4666  load_file("libpqwalreceiver", false);
4667 
4669 
4670  /* Connect to the origin and start the replication. */
4671  elog(DEBUG1, "connecting to publisher using connection string \"%s\"",
4673 
4674  /*
4675  * Setup callback for syscache so that we know when something changes in
4676  * the subscription relation state.
4677  */
4678  CacheRegisterSyscacheCallback(SUBSCRIPTIONRELMAP,
4680  (Datum) 0);
4681 }
void InitializeLogRepWorker(void)
Definition: worker.c:4561
void load_file(const char *filename, bool restricted)
Definition: dfmgr.c:144
void SignalHandlerForConfigReload(SIGNAL_ARGS)
Definition: interrupt.c:61
void logicalrep_worker_attach(int slot)
Definition: launcher.c:707
#define die(msg)
pqsigfunc pqsignal(int signo, pqsigfunc func)
void BackgroundWorkerUnblockSignals(void)
Definition: postmaster.c:4229
void invalidate_syncing_table_states(Datum arg, int cacheid, uint32 hashvalue)
Definition: tablesync.c:274
#define SIGHUP
Definition: win32_port.h:168

References am_leader_apply_worker(), am_tablesync_worker(), Assert, BackgroundWorkerUnblockSignals(), 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(), LogicalRepWorker::reply_time, SIGHUP, and SignalHandlerForConfigReload().

Referenced by ApplyWorkerMain(), and TablesyncWorkerMain().

◆ start_apply()

void start_apply ( XLogRecPtr  origin_startpos)

Definition at line 4425 of file worker.c.

4426 {
4427  PG_TRY();
4428  {
4429  LogicalRepApplyLoop(origin_startpos);
4430  }
4431  PG_CATCH();
4432  {
4435  else
4436  {
4437  /*
4438  * Report the worker failed while applying changes. Abort the
4439  * current transaction so that the stats message is sent in an
4440  * idle state.
4441  */
4444 
4445  PG_RE_THROW();
4446  }
4447  }
4448  PG_END_TRY();
4449 }
static void LogicalRepApplyLoop(XLogRecPtr last_received)
Definition: worker.c:3478
void DisableSubscriptionAndExit(void)
Definition: worker.c:4705
#define PG_RE_THROW()
Definition: elog.h:411
#define PG_TRY(...)
Definition: elog.h:370
#define PG_END_TRY(...)
Definition: elog.h:395
#define PG_CATCH(...)
Definition: elog.h:380

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

3437 {
3438  FlushPosition *flushpos;
3439 
3440  /*
3441  * Skip for parallel apply workers, because the lsn_mapping is maintained
3442  * by the leader apply worker.
3443  */
3445  return;
3446 
3447  /* Need to do this in permanent context */
3449 
3450  /* Track commit lsn */
3451  flushpos = (FlushPosition *) palloc(sizeof(FlushPosition));
3452  flushpos->local_end = local_lsn;
3453  flushpos->remote_end = remote_lsn;
3454 
3455  dlist_push_tail(&lsn_mapping, &flushpos->node);
3457 }
static dlist_head lsn_mapping
Definition: worker.c:204
static void dlist_push_tail(dlist_head *head, dlist_node *node)
Definition: ilist.h:364
dlist_node node
Definition: worker.c:199
XLogRecPtr remote_end
Definition: worker.c:201
XLogRecPtr local_end
Definition: worker.c:200

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

4206 {
4207  char path[MAXPGPATH];
4208 
4209  /* Delete the changes file. */
4210  changes_filename(path, subid, xid);
4212 
4213  /* Delete the subxact file, if it exists. */
4214  subxact_filename(path, subid, xid);
4216 }
static void subxact_filename(char *path, Oid subid, TransactionId xid)
Definition: worker.c:4184
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 1429 of file worker.c.

1430 {
1432 
1433  /*
1434  * Initialize the worker's stream_fileset if we haven't yet. This will be
1435  * used for the entire duration of the worker so create it in a permanent
1436  * context. We create this on the very first streaming message from any
1437  * transaction and then use it for this and other streaming transactions.
1438  * Now, we could create a fileset at the start of the worker as well but
1439  * then we won't be sure that it will ever be used.
1440  */
1442  {
1443  MemoryContext oldctx;
1444 
1446 
1449 
1450  MemoryContextSwitchTo(oldctx);
1451  }
1452 
1453  /* Open the spool file for this transaction. */
1454  stream_open_file(MyLogicalRepWorker->subid, xid, first_segment);
1455 
1456  /* If this is not the first segment, open existing subxact file. */
1457  if (!first_segment)
1459 
1461 }
static void stream_open_file(Oid subid, TransactionId xid, bool first_segment)
Definition: worker.c:4229
static void subxact_info_read(Oid subid, TransactionId xid)
Definition: worker.c:4055
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 1603 of file worker.c.

1604 {
1605  /*
1606  * Serialize information about subxacts for the toplevel transaction, then
1607  * close the stream messages spool file.
1608  */
1611 
1612  /* We must be in a valid transaction state */
1614 
1615  /* Commit the per-stream transaction */
1617 
1618  /* Reset per-stream context */
1620 }
static void subxact_info_write(Oid subid, TransactionId xid)
Definition: worker.c:4006
static MemoryContext LogicalStreamingContext
Definition: worker.c:294

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

1740 {
1741  Relation rel;
1742  HeapTuple tup;
1743  bool nulls[Natts_pg_subscription];
1744  bool replaces[Natts_pg_subscription];
1745  Datum values[Natts_pg_subscription];
1746 
1748  new_state == LOGICALREP_TWOPHASE_STATE_PENDING ||
1749  new_state == LOGICALREP_TWOPHASE_STATE_ENABLED);
1750 
1751  rel = table_open(SubscriptionRelationId, RowExclusiveLock);
1752  tup = SearchSysCacheCopy1(SUBSCRIPTIONOID, ObjectIdGetDatum(suboid));
1753  if (!HeapTupleIsValid(tup))
1754  elog(ERROR,
1755  "cache lookup failed for subscription oid %u",
1756  suboid);
1757 
1758  /* Form a new tuple. */
1759  memset(values, 0, sizeof(values));
1760  memset(nulls, false, sizeof(nulls));
1761  memset(replaces, false, sizeof(replaces));
1762 
1763  /* And update/set two_phase state */
1764  values[Anum_pg_subscription_subtwophasestate - 1] = CharGetDatum(new_state);
1765  replaces[Anum_pg_subscription_subtwophasestate - 1] = true;
1766 
1767  tup = heap_modify_tuple(tup, RelationGetDescr(rel),
1768  values, nulls, replaces);
1769  CatalogTupleUpdate(rel, &tup->t_self, tup);
1770 
1771  heap_freetuple(tup);
1773 }
static Datum values[MAXATTR]
Definition: bootstrap.c:152
HeapTuple heap_modify_tuple(HeapTuple tuple, TupleDesc tupleDesc, const Datum *replValues, const bool *replIsnull, const bool *doReplace)
Definition: heaptuple.c:1209
void heap_freetuple(HeapTuple htup)
Definition: heaptuple.c:1434
#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
#define LOGICALREP_TWOPHASE_STATE_DISABLED
#define LOGICALREP_TWOPHASE_STATE_PENDING
#define LOGICALREP_TWOPHASE_STATE_ENABLED
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:86
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, LOGICALREP_TWOPHASE_STATE_DISABLED, LOGICALREP_TWOPHASE_STATE_ENABLED, LOGICALREP_TWOPHASE_STATE_PENDING, 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 288 of file worker.c.

Referenced by HandleParallelApplyMessage(), and LogicalRepApplyLoop().

◆ ApplyContext

◆ ApplyMessageContext

◆ in_remote_transaction

◆ InitializingApplyWorker

PGDLLIMPORT bool InitializingApplyWorker
extern

Definition at line 318 of file worker.c.

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

◆ LogRepWorkerWalRcvConn

◆ MyLogicalRepWorker

◆ MyParallelShared

◆ MySubscription