PostgreSQL Source Code  git master
worker_internal.h File Reference
#include <signal.h>
#include "access/xlogdefs.h"
#include "catalog/pg_subscription.h"
#include "datatype/timestamp.h"
#include "miscadmin.h"
#include "replication/logicalrelation.h"
#include "storage/buffile.h"
#include "storage/fileset.h"
#include "storage/lock.h"
#include "storage/shm_mq.h"
#include "storage/shm_toc.h"
#include "storage/spin.h"
Include dependency graph for worker_internal.h:
This graph shows which files directly or indirectly include this file:

Go to the source code of this file.

Data Structures

struct  LogicalRepWorker
 
struct  ParallelApplyWorkerShared
 
struct  ParallelApplyWorkerInfo
 

Macros

#define isParallelApplyWorker(worker)   ((worker)->leader_pid != InvalidPid)
 

Typedefs

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

Enumerations

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

Functions

void logicalrep_worker_attach (int slot)
 
LogicalRepWorkerlogicalrep_worker_find (Oid subid, Oid relid, bool only_running)
 
Listlogicalrep_workers_find (Oid subid, bool only_running)
 
bool logicalrep_worker_launch (Oid dbid, Oid subid, const char *subname, Oid userid, Oid relid, dsm_handle subworker_dsm)
 
void logicalrep_worker_stop (Oid subid, Oid relid)
 
void logicalrep_pa_worker_stop (int slot_no, uint16 generation)
 
void logicalrep_worker_wakeup (Oid subid, Oid relid)
 
void logicalrep_worker_wakeup_ptr (LogicalRepWorker *worker)
 
int logicalrep_sync_worker_count (Oid subid)
 
void ReplicationOriginNameForLogicalRep (Oid suboid, Oid relid, char *originname, Size szoriginname)
 
char * LogicalRepSyncTableStart (XLogRecPtr *origin_startpos)
 
bool AllTablesyncsReady (void)
 
void UpdateTwoPhaseState (Oid suboid, char new_state)
 
void process_syncing_tables (XLogRecPtr current_lsn)
 
void invalidate_syncing_table_states (Datum arg, int cacheid, uint32 hashvalue)
 
void stream_start_internal (TransactionId xid, bool first_segment)
 
void stream_stop_internal (TransactionId xid)
 
void apply_spooled_messages (FileSet *stream_fileset, TransactionId xid, XLogRecPtr lsn)
 
void apply_dispatch (StringInfo s)
 
void maybe_reread_subscription (void)
 
void stream_cleanup_files (Oid subid, TransactionId xid)
 
void InitializeApplyWorker (void)
 
void store_flush_position (XLogRecPtr remote_lsn, XLogRecPtr local_lsn)
 
void apply_error_callback (void *arg)
 
void set_apply_error_context_origin (char *originname)
 
void pa_allocate_worker (TransactionId xid)
 
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 in_xact)
 
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
 

Macro Definition Documentation

◆ isParallelApplyWorker

#define isParallelApplyWorker (   worker)    ((worker)->leader_pid != InvalidPid)

Definition at line 306 of file worker_internal.h.

Typedef Documentation

◆ LogicalRepWorker

◆ ParallelApplyWorkerInfo

◆ ParallelApplyWorkerShared

◆ ParallelTransState

◆ PartialFileSetState

Enumeration Type Documentation

◆ ParallelTransState

Enumerator
PARALLEL_TRANS_UNKNOWN 
PARALLEL_TRANS_STARTED 
PARALLEL_TRANS_FINISHED 

Definition at line 93 of file worker_internal.h.

94 {
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 116 of file worker_internal.h.

117 {
118  FS_EMPTY,
121  FS_READY
PartialFileSetState
@ FS_EMPTY
@ FS_SERIALIZE_DONE
@ FS_READY
@ FS_SERIALIZE_IN_PROGRESS

Function Documentation

◆ AllTablesyncsReady()

bool AllTablesyncsReady ( void  )

Definition at line 1558 of file tablesync.c.

1559 {
1560  bool started_tx = false;
1561  bool has_subrels = false;
1562 
1563  /* We need up-to-date sync state info for subscription tables here. */
1564  has_subrels = FetchTableStates(&started_tx);
1565 
1566  if (started_tx)
1567  {
1569  pgstat_report_stat(true);
1570  }
1571 
1572  /*
1573  * Return false when there are no tables in subscription or not all tables
1574  * are in ready state; true otherwise.
1575  */
1576  return has_subrels && (table_states_not_ready == NIL);
1577 }
#define NIL
Definition: pg_list.h:68
long pgstat_report_stat(bool force)
Definition: pgstat.c:565
static List * table_states_not_ready
Definition: tablesync.c:124
static bool FetchTableStates(bool *started_tx)
Definition: tablesync.c:1497
void CommitTransactionCommand(void)
Definition: xact.c:3041

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

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

◆ am_leader_apply_worker()

static bool am_leader_apply_worker ( void  )
inlinestatic

Definition at line 315 of file worker_internal.h.

316 {
317  return (!am_tablesync_worker() &&
319 }
#define isParallelApplyWorker(worker)
PGDLLIMPORT LogicalRepWorker * MyLogicalRepWorker
Definition: launcher.c:61
static bool am_tablesync_worker(void)

References am_tablesync_worker(), isParallelApplyWorker, and MyLogicalRepWorker.

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

◆ am_parallel_apply_worker()

◆ am_tablesync_worker()

◆ apply_dispatch()

void apply_dispatch ( StringInfo  s)

Definition at line 3244 of file worker.c.

3245 {
3247  LogicalRepMsgType saved_command;
3248 
3249  /*
3250  * Set the current command being applied. Since this function can be
3251  * called recursively when applying spooled changes, save the current
3252  * command.
3253  */
3254  saved_command = apply_error_callback_arg.command;
3256 
3257  switch (action)
3258  {
3259  case LOGICAL_REP_MSG_BEGIN:
3260  apply_handle_begin(s);
3261  break;
3262 
3265  break;
3266 
3269  break;
3270 
3273  break;
3274 
3277  break;
3278 
3281  break;
3282 
3285  break;
3286 
3287  case LOGICAL_REP_MSG_TYPE:
3288  apply_handle_type(s);
3289  break;
3290 
3293  break;
3294 
3296 
3297  /*
3298  * Logical replication does not use generic logical messages yet.
3299  * Although, it could be used by other applications that use this
3300  * output plugin.
3301  */
3302  break;
3303 
3306  break;
3307 
3310  break;
3311 
3314  break;
3315 
3318  break;
3319 
3322  break;
3323 
3326  break;
3327 
3330  break;
3331 
3334  break;
3335 
3338  break;
3339 
3340  default:
3341  ereport(ERROR,
3342  (errcode(ERRCODE_PROTOCOL_VIOLATION),
3343  errmsg("invalid logical replication message type \"%c\"", action)));
3344  }
3345 
3346  /* Reset the current command */
3347  apply_error_callback_arg.command = saved_command;
3348 }
static void apply_handle_stream_prepare(StringInfo s)
Definition: worker.c:1286
static void apply_handle_type(StringInfo s)
Definition: worker.c:2341
static void apply_handle_truncate(StringInfo s)
Definition: worker.c:3122
static void apply_handle_update(StringInfo s)
Definition: worker.c:2542
static void apply_handle_stream_commit(StringInfo s)
Definition: worker.c:2148
static void apply_handle_commit_prepared(StringInfo s)
Definition: worker.c:1184
ApplyErrorCallbackArg apply_error_callback_arg
Definition: worker.c:294
static void apply_handle_delete(StringInfo s)
Definition: worker.c:2711
static void apply_handle_begin(StringInfo s)
Definition: worker.c:1006
static void apply_handle_commit(StringInfo s)
Definition: worker.c:1031
static void apply_handle_stream_abort(StringInfo s)
Definition: worker.c:1827
static void apply_handle_relation(StringInfo s)
Definition: worker.c:2318
static void apply_handle_prepare(StringInfo s)
Definition: worker.c:1123
static void apply_handle_rollback_prepared(StringInfo s)
Definition: worker.c:1233
static void apply_handle_stream_stop(StringInfo s)
Definition: worker.c:1641
static void apply_handle_origin(StringInfo s)
Definition: worker.c:1423
static void apply_handle_begin_prepare(StringInfo s)
Definition: worker.c:1057
static void apply_handle_stream_start(StringInfo s)
Definition: worker.c:1482
static void apply_handle_insert(StringInfo s)
Definition: worker.c:2406
int errcode(int sqlerrcode)
Definition: elog.c:858
int errmsg(const char *fmt,...)
Definition: elog.c:1069
#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:402
LogicalRepMsgType command
Definition: worker.c:237

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

4863 {
4865 
4867  return;
4868 
4869  Assert(errarg->origin_name);
4870 
4871  if (errarg->rel == NULL)
4872  {
4873  if (!TransactionIdIsValid(errarg->remote_xid))
4874  errcontext("processing remote data for replication origin \"%s\" during message type \"%s\"",
4875  errarg->origin_name,
4876  logicalrep_message_type(errarg->command));
4877  else if (XLogRecPtrIsInvalid(errarg->finish_lsn))
4878  errcontext("processing remote data for replication origin \"%s\" during message type \"%s\" in transaction %u",
4879  errarg->origin_name,
4881  errarg->remote_xid);
4882  else
4883  errcontext("processing remote data for replication origin \"%s\" during message type \"%s\" in transaction %u, finished at %X/%X",
4884  errarg->origin_name,
4886  errarg->remote_xid,
4887  LSN_FORMAT_ARGS(errarg->finish_lsn));
4888  }
4889  else
4890  {
4891  if (errarg->remote_attnum < 0)
4892  {
4893  if (XLogRecPtrIsInvalid(errarg->finish_lsn))
4894  errcontext("processing remote data for replication origin \"%s\" during message type \"%s\" for replication target relation \"%s.%s\" in transaction %u",
4895  errarg->origin_name,
4897  errarg->rel->remoterel.nspname,
4898  errarg->rel->remoterel.relname,
4899  errarg->remote_xid);
4900  else
4901  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",
4902  errarg->origin_name,
4904  errarg->rel->remoterel.nspname,
4905  errarg->rel->remoterel.relname,
4906  errarg->remote_xid,
4907  LSN_FORMAT_ARGS(errarg->finish_lsn));
4908  }
4909  else
4910  {
4911  if (XLogRecPtrIsInvalid(errarg->finish_lsn))
4912  errcontext("processing remote data for replication origin \"%s\" during message type \"%s\" for replication target relation \"%s.%s\" column \"%s\" in transaction %u",
4913  errarg->origin_name,
4915  errarg->rel->remoterel.nspname,
4916  errarg->rel->remoterel.relname,
4917  errarg->rel->remoterel.attnames[errarg->remote_attnum],
4918  errarg->remote_xid);
4919  else
4920  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",
4921  errarg->origin_name,
4923  errarg->rel->remoterel.nspname,
4924  errarg->rel->remoterel.relname,
4925  errarg->rel->remoterel.attnames[errarg->remote_attnum],
4926  errarg->remote_xid,
4927  LSN_FORMAT_ARGS(errarg->finish_lsn));
4928  }
4929  }
4930 }
#define errcontext
Definition: elog.h:196
Assert(fmt[strlen(fmt) - 1] !='\n')
char * logicalrep_message_type(LogicalRepMsgType action)
Definition: proto.c:1223
TransactionId remote_xid
Definition: worker.c:242
XLogRecPtr finish_lsn
Definition: worker.c:243
LogicalRepRelMapEntry * rel
Definition: worker.c:238
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 2016 of file worker.c.

2018 {
2020  int nchanges;
2021  char path[MAXPGPATH];
2022  char *buffer = NULL;
2023  MemoryContext oldcxt;
2024  ResourceOwner oldowner;
2025  int fileno;
2026  off_t offset;
2027 
2028  if (!am_parallel_apply_worker())
2030 
2031  /* Make sure we have an open transaction */
2033 
2034  /*
2035  * Allocate file handle and memory required to process all the messages in
2036  * TopTransactionContext to avoid them getting reset after each message is
2037  * processed.
2038  */
2040 
2041  /* Open the spool file for the committed/prepared transaction */
2043  elog(DEBUG1, "replaying changes from file \"%s\"", path);
2044 
2045  /*
2046  * Make sure the file is owned by the toplevel transaction so that the
2047  * file will not be accidentally closed when aborting a subtransaction.
2048  */
2049  oldowner = CurrentResourceOwner;
2051 
2052  stream_fd = BufFileOpenFileSet(stream_fileset, path, O_RDONLY, false);
2053 
2054  CurrentResourceOwner = oldowner;
2055 
2056  buffer = palloc(BLCKSZ);
2057  initStringInfo(&s2);
2058 
2059  MemoryContextSwitchTo(oldcxt);
2060 
2061  remote_final_lsn = lsn;
2062 
2063  /*
2064  * Make sure the handle apply_dispatch methods are aware we're in a remote
2065  * transaction.
2066  */
2067  in_remote_transaction = true;
2069 
2071 
2072  /*
2073  * Read the entries one by one and pass them through the same logic as in
2074  * apply_dispatch.
2075  */
2076  nchanges = 0;
2077  while (true)
2078  {
2079  size_t nbytes;
2080  int len;
2081 
2083 
2084  /* read length of the on-disk record */
2085  nbytes = BufFileReadMaybeEOF(stream_fd, &len, sizeof(len), true);
2086 
2087  /* have we reached end of the file? */
2088  if (nbytes == 0)
2089  break;
2090 
2091  /* do we have a correct length? */
2092  if (len <= 0)
2093  elog(ERROR, "incorrect length %d in streaming transaction's changes file \"%s\"",
2094  len, path);
2095 
2096  /* make sure we have sufficiently large buffer */
2097  buffer = repalloc(buffer, len);
2098 
2099  /* and finally read the data into the buffer */
2100  BufFileReadExact(stream_fd, buffer, len);
2101 
2102  BufFileTell(stream_fd, &fileno, &offset);
2103 
2104  /* copy the buffer to the stringinfo and call apply_dispatch */
2105  resetStringInfo(&s2);
2106  appendBinaryStringInfo(&s2, buffer, len);
2107 
2108  /* Ensure we are reading the data into our memory context. */
2110 
2111  apply_dispatch(&s2);
2112 
2114 
2115  MemoryContextSwitchTo(oldcxt);
2116 
2117  nchanges++;
2118 
2119  /*
2120  * It is possible the file has been closed because we have processed
2121  * the transaction end message like stream_commit in which case that
2122  * must be the last message.
2123  */
2124  if (!stream_fd)
2125  {
2126  ensure_last_message(stream_fileset, xid, fileno, offset);
2127  break;
2128  }
2129 
2130  if (nchanges % 1000 == 0)
2131  elog(DEBUG1, "replayed %d changes from file \"%s\"",
2132  nchanges, path);
2133  }
2134 
2135  if (stream_fd)
2137 
2138  elog(DEBUG1, "replayed %d (all) changes from file \"%s\"",
2139  nchanges, path);
2140 
2141  return;
2142 }
static void begin_replication_step(void)
Definition: worker.c:525
static void end_replication_step(void)
Definition: worker.c:548
MemoryContext ApplyMessageContext
Definition: worker.c:306
static void changes_filename(char *path, Oid subid, TransactionId xid)
Definition: worker.c:4143
static BufFile * stream_fd
Definition: worker.c:352
bool in_remote_transaction
Definition: worker.c:319
void apply_dispatch(StringInfo s)
Definition: worker.c:3244
static void ensure_last_message(FileSet *stream_fileset, TransactionId xid, int fileno, off_t offset)
Definition: worker.c:1984
static XLogRecPtr remote_final_lsn
Definition: worker.c:320
static void maybe_start_skipping_changes(XLogRecPtr finish_lsn)
Definition: worker.c:4732
static void stream_close_file(void)
Definition: worker.c:4226
void pgstat_report_activity(BackendState state, const char *cmd_str)
@ STATE_RUNNING
void BufFileReadExact(BufFile *file, void *ptr, size_t size)
Definition: buffile.c:651
BufFile * BufFileOpenFileSet(FileSet *fileset, const char *name, int mode, bool missing_ok)
Definition: buffile.c:286
void BufFileTell(BufFile *file, int *fileno, off_t *offset)
Definition: buffile.c:830
size_t BufFileReadMaybeEOF(BufFile *file, void *ptr, size_t size, bool eofOK)
Definition: buffile.c:661
#define DEBUG1
Definition: elog.h:30
LogicalRepWorker * MyLogicalRepWorker
Definition: launcher.c:61
void MemoryContextReset(MemoryContext context)
Definition: mcxt.c:314
MemoryContext TopTransactionContext
Definition: mcxt.c:146
void * repalloc(void *pointer, Size size)
Definition: mcxt.c:1456
void * palloc(Size size)
Definition: mcxt.c:1210
#define CHECK_FOR_INTERRUPTS()
Definition: miscadmin.h:121
static MemoryContext MemoryContextSwitchTo(MemoryContext context)
Definition: palloc.h:138
#define MAXPGPATH
const void size_t len
char * s2
ResourceOwner TopTransactionResourceOwner
Definition: resowner.c:148
ResourceOwner CurrentResourceOwner
Definition: resowner.c:146
void resetStringInfo(StringInfo str)
Definition: stringinfo.c:75
void appendBinaryStringInfo(StringInfo str, const void *data, int datalen)
Definition: stringinfo.c:227
void initStringInfo(StringInfo str)
Definition: stringinfo.c:59
static bool am_parallel_apply_worker(void)

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

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

◆ InitializeApplyWorker()

void InitializeApplyWorker ( void  )

Definition at line 4403 of file worker.c.

4404 {
4405  MemoryContext oldctx;
4406 
4407  /* Run as replica session replication role. */
4408  SetConfigOption("session_replication_role", "replica",
4410 
4411  /* Connect to our database. */
4414  0);
4415 
4416  /*
4417  * Set always-secure search path, so malicious users can't redirect user
4418  * code (e.g. pg_index.indexprs).
4419  */
4420  SetConfigOption("search_path", "", PGC_SUSET, PGC_S_OVERRIDE);
4421 
4422  /* Load the subscription into persistent memory context. */
4424  "ApplyContext",
4428 
4430  if (!MySubscription)
4431  {
4432  ereport(LOG,
4433  /* translator: %s is the name of logical replication worker */
4434  (errmsg("%s for subscription %u will not start because the subscription was removed during startup",
4436 
4437  /* Ensure we remove no-longer-useful entry for worker's start time */
4440  proc_exit(0);
4441  }
4442 
4443  MySubscriptionValid = true;
4444  MemoryContextSwitchTo(oldctx);
4445 
4446  if (!MySubscription->enabled)
4447  {
4448  ereport(LOG,
4449  /* translator: first %s is the name of logical replication worker */
4450  (errmsg("%s for subscription \"%s\" will not start because the subscription was disabled during startup",
4452 
4454  }
4455 
4456  /* Setup synchronous commit according to the user's wishes */
4457  SetConfigOption("synchronous_commit", MySubscription->synccommit,
4459 
4460  /* Keep us informed about subscription changes. */
4463  (Datum) 0);
4464 
4465  if (am_tablesync_worker())
4466  ereport(LOG,
4467  (errmsg("logical replication table synchronization worker for subscription \"%s\", table \"%s\" has started",
4470  else
4471  ereport(LOG,
4472  /* translator: first %s is the name of logical replication worker */
4473  (errmsg("%s for subscription \"%s\" has started",
4475 
4477 }
static void subscription_change_cb(Datum arg, int cacheid, uint32 hashvalue)
Definition: worker.c:3943
static void apply_worker_exit(void)
Definition: worker.c:3800
MemoryContext ApplyContext
Definition: worker.c:307
static bool MySubscriptionValid
Definition: worker.c:315
Subscription * MySubscription
Definition: worker.c:314
static const char * get_worker_name(void)
Definition: worker.c:435
#define LOG
Definition: elog.h:31
void SetConfigOption(const char *name, const char *value, GucContext context, GucSource source)
Definition: guc.c:4157
@ 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:1519
void proc_exit(int code)
Definition: ipc.c:104
void ApplyLauncherForgetWorkerStartTime(Oid subid)
Definition: launcher.c:1031
char * get_rel_name(Oid relid)
Definition: lsyscache.c:1910
MemoryContext TopMemoryContext
Definition: mcxt.c:141
#define AllocSetContextCreate
Definition: memutils.h:129
#define ALLOCSET_DEFAULT_SIZES
Definition: memutils.h:153
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:5635
@ SUBSCRIPTIONOID
Definition: syscache.h:99
void StartTransactionCommand(void)
Definition: xact.c:2944

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

Referenced by ApplyWorkerMain(), and ParallelApplyWorkerMain().

◆ invalidate_syncing_table_states()

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

Definition at line 271 of file tablesync.c.

272 {
273  table_states_valid = false;
274 }
static bool table_states_valid
Definition: tablesync.c:123

References table_states_valid.

Referenced by ApplyWorkerMain(), and ParallelApplyWorkerMain().

◆ logicalrep_pa_worker_stop()

void logicalrep_pa_worker_stop ( int  slot_no,
uint16  generation 
)

Definition at line 619 of file launcher.c.

620 {
621  LogicalRepWorker *worker;
622 
623  Assert(slot_no >= 0 && slot_no < max_logical_replication_workers);
624 
625  LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
626 
627  worker = &LogicalRepCtx->workers[slot_no];
628  Assert(isParallelApplyWorker(worker));
629 
630  /*
631  * Only stop the worker if the generation matches and the worker is alive.
632  */
633  if (worker->generation == generation && worker->proc)
634  logicalrep_worker_stop_internal(worker, SIGINT);
635 
636  LWLockRelease(LogicalRepWorkerLock);
637 }
int max_logical_replication_workers
Definition: launcher.c:57
static void logicalrep_worker_stop_internal(LogicalRepWorker *worker, int signo)
Definition: launcher.c:512
static LogicalRepCtxStruct * LogicalRepCtx
Definition: launcher.c:76
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1195
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1803
@ LW_SHARED
Definition: lwlock.h:116
LogicalRepWorker workers[FLEXIBLE_ARRAY_MEMBER]
Definition: launcher.c:73

References Assert(), LogicalRepWorker::generation, isParallelApplyWorker, logicalrep_worker_stop_internal(), LogicalRepCtx, LW_SHARED, LWLockAcquire(), LWLockRelease(), max_logical_replication_workers, LogicalRepWorker::proc, and LogicalRepCtxStruct::workers.

Referenced by pa_free_worker().

◆ logicalrep_sync_worker_count()

int logicalrep_sync_worker_count ( Oid  subid)

Definition at line 811 of file launcher.c.

812 {
813  int i;
814  int res = 0;
815 
816  Assert(LWLockHeldByMe(LogicalRepWorkerLock));
817 
818  /* Search for attached worker for a given subscription id. */
819  for (i = 0; i < max_logical_replication_workers; i++)
820  {
822 
823  if (w->subid == subid && OidIsValid(w->relid))
824  res++;
825  }
826 
827  return res;
828 }
int i
Definition: isn.c:73
bool LWLockHeldByMe(LWLock *lock)
Definition: lwlock.c:1919

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

Referenced by logicalrep_worker_launch(), and process_syncing_tables_for_apply().

◆ logicalrep_worker_attach()

void logicalrep_worker_attach ( int  slot)

Definition at line 674 of file launcher.c.

675 {
676  /* Block concurrent access. */
677  LWLockAcquire(LogicalRepWorkerLock, LW_EXCLUSIVE);
678 
679  Assert(slot >= 0 && slot < max_logical_replication_workers);
681 
683  {
684  LWLockRelease(LogicalRepWorkerLock);
685  ereport(ERROR,
686  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
687  errmsg("logical replication worker slot %d is empty, cannot attach",
688  slot)));
689  }
690 
692  {
693  LWLockRelease(LogicalRepWorkerLock);
694  ereport(ERROR,
695  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
696  errmsg("logical replication worker slot %d is already used by "
697  "another worker, cannot attach", slot)));
698  }
699 
702 
703  LWLockRelease(LogicalRepWorkerLock);
704 }
void before_shmem_exit(pg_on_exit_callback function, Datum arg)
Definition: ipc.c:333
static void logicalrep_worker_onexit(int code, Datum arg)
Definition: launcher.c:784
@ LW_EXCLUSIVE
Definition: lwlock.h:115
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 ApplyWorkerMain(), and ParallelApplyWorkerMain().

◆ logicalrep_worker_find()

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

Definition at line 249 of file launcher.c.

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

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

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

◆ logicalrep_worker_launch()

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

Definition at line 306 of file launcher.c.

308 {
309  BackgroundWorker bgw;
310  BackgroundWorkerHandle *bgw_handle;
311  uint16 generation;
312  int i;
313  int slot = 0;
314  LogicalRepWorker *worker = NULL;
315  int nsyncworkers;
316  int nparallelapplyworkers;
318  bool is_parallel_apply_worker = (subworker_dsm != DSM_HANDLE_INVALID);
319 
320  /* Sanity check - tablesync worker cannot be a subworker */
321  Assert(!(is_parallel_apply_worker && OidIsValid(relid)));
322 
323  ereport(DEBUG1,
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)
329  ereport(ERROR,
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 
339 retry:
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  {
378  elog(WARNING,
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 (OidIsValid(relid) && 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 max_logical_replication_workers.")));
426  return false;
427  }
428 
429  /* Prepare the worker slot. */
430  worker->launch_time = now;
431  worker->in_use = true;
432  worker->generation++;
433  worker->proc = NULL;
434  worker->dbid = dbid;
435  worker->userid = userid;
436  worker->subid = subid;
437  worker->relid = relid;
438  worker->relstate = SUBREL_STATE_UNKNOWN;
440  worker->stream_fileset = NULL;
441  worker->leader_pid = is_parallel_apply_worker ? MyProcPid : InvalidPid;
442  worker->parallel_apply = is_parallel_apply_worker;
443  worker->last_lsn = InvalidXLogRecPtr;
446  worker->reply_lsn = InvalidXLogRecPtr;
447  TIMESTAMP_NOBEGIN(worker->reply_time);
448 
449  /* Before releasing lock, remember generation for future identification. */
450  generation = worker->generation;
451 
452  LWLockRelease(LogicalRepWorkerLock);
453 
454  /* Register the new dynamic worker. */
455  memset(&bgw, 0, sizeof(bgw));
459  snprintf(bgw.bgw_library_name, BGW_MAXLEN, "postgres");
460 
461  if (is_parallel_apply_worker)
462  snprintf(bgw.bgw_function_name, BGW_MAXLEN, "ParallelApplyWorkerMain");
463  else
464  snprintf(bgw.bgw_function_name, BGW_MAXLEN, "ApplyWorkerMain");
465 
466  if (OidIsValid(relid))
468  "logical replication worker for subscription %u sync %u", subid, relid);
469  else if (is_parallel_apply_worker)
471  "logical replication parallel apply worker for subscription %u", subid);
472  else
474  "logical replication apply worker for subscription %u", subid);
475 
476  if (is_parallel_apply_worker)
477  snprintf(bgw.bgw_type, BGW_MAXLEN, "logical replication parallel worker");
478  else
479  snprintf(bgw.bgw_type, BGW_MAXLEN, "logical replication worker");
480 
483  bgw.bgw_main_arg = Int32GetDatum(slot);
484 
485  if (is_parallel_apply_worker)
486  memcpy(bgw.bgw_extra, &subworker_dsm, sizeof(dsm_handle));
487 
488  if (!RegisterDynamicBackgroundWorker(&bgw, &bgw_handle))
489  {
490  /* Failed to start worker, so clean up the worker slot. */
491  LWLockAcquire(LogicalRepWorkerLock, LW_EXCLUSIVE);
492  Assert(generation == worker->generation);
494  LWLockRelease(LogicalRepWorkerLock);
495 
497  (errcode(ERRCODE_CONFIGURATION_LIMIT_EXCEEDED),
498  errmsg("out of background worker slots"),
499  errhint("You might need to increase max_worker_processes.")));
500  return false;
501  }
502 
503  /* Now wait until it attaches. */
504  return WaitForReplicationWorkerAttach(worker, generation, bgw_handle);
505 }
bool TimestampDifferenceExceeds(TimestampTz start_time, TimestampTz stop_time, int msec)
Definition: timestamp.c:1730
TimestampTz GetCurrentTimestamp(void)
Definition: timestamp.c:1585
Datum now(PG_FUNCTION_ARGS)
Definition: timestamp.c:1549
bool RegisterDynamicBackgroundWorker(BackgroundWorker *worker, BackgroundWorkerHandle **handle)
Definition: bgworker.c:959
#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
unsigned short uint16
Definition: c.h:489
int64 TimestampTz
Definition: timestamp.h:39
#define TIMESTAMP_NOBEGIN(j)
Definition: timestamp.h:158
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:1156
int errhint(const char *fmt,...)
Definition: elog.c:1316
#define WARNING
Definition: elog.h:36
int MyProcPid
Definition: globals.c:44
static int logicalrep_pa_worker_count(Oid subid)
Definition: launcher.c:835
int max_sync_workers_per_subscription
Definition: launcher.c:58
static bool WaitForReplicationWorkerAttach(LogicalRepWorker *worker, uint16 generation, BackgroundWorkerHandle *handle)
Definition: launcher.c:189
int logicalrep_sync_worker_count(Oid subid)
Definition: launcher.c:811
int max_parallel_apply_workers_per_subscription
Definition: launcher.c:59
static void logicalrep_worker_cleanup(LogicalRepWorker *worker)
Definition: launcher.c:753
#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:101
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[BGW_MAXLEN]
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:91
#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, MyProcPid, now(), OidIsValid, LogicalRepWorker::parallel_apply, LogicalRepWorker::proc, RegisterDynamicBackgroundWorker(), LogicalRepWorker::relid, LogicalRepWorker::relstate, LogicalRepWorker::relstate_lsn, LogicalRepWorker::reply_lsn, LogicalRepWorker::reply_time, snprintf, LogicalRepWorker::stream_fileset, LogicalRepWorker::subid, subname, TIMESTAMP_NOBEGIN, TimestampDifferenceExceeds(), LogicalRepWorker::userid, WaitForReplicationWorkerAttach(), wal_receiver_timeout, WARNING, and LogicalRepCtxStruct::workers.

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

◆ logicalrep_worker_stop()

void logicalrep_worker_stop ( Oid  subid,
Oid  relid 
)

Definition at line 594 of file launcher.c.

595 {
596  LogicalRepWorker *worker;
597 
598  LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
599 
600  worker = logicalrep_worker_find(subid, relid, false);
601 
602  if (worker)
603  {
604  Assert(!isParallelApplyWorker(worker));
605  logicalrep_worker_stop_internal(worker, SIGTERM);
606  }
607 
608  LWLockRelease(LogicalRepWorkerLock);
609 }
LogicalRepWorker * logicalrep_worker_find(Oid subid, Oid relid, bool only_running)
Definition: launcher.c:249

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

644 {
645  LogicalRepWorker *worker;
646 
647  LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
648 
649  worker = logicalrep_worker_find(subid, relid, true);
650 
651  if (worker)
653 
654  LWLockRelease(LogicalRepWorkerLock);
655 }
void logicalrep_worker_wakeup_ptr(LogicalRepWorker *worker)
Definition: launcher.c:663

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

664 {
665  Assert(LWLockHeldByMe(LogicalRepWorkerLock));
666 
667  SetLatch(&worker->proc->procLatch);
668 }
void SetLatch(Latch *latch)
Definition: latch.c:607
Latch procLatch
Definition: proc.h:170

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

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

◆ logicalrep_workers_find()

List* logicalrep_workers_find ( Oid  subid,
bool  only_running 
)

Definition at line 281 of file launcher.c.

282 {
283  int i;
284  List *res = NIL;
285 
286  Assert(LWLockHeldByMe(LogicalRepWorkerLock));
287 
288  /* Search for attached worker for a given subscription id. */
289  for (i = 0; i < max_logical_replication_workers; i++)
290  {
292 
293  if (w->in_use && w->subid == subid && (!only_running || w->proc))
294  res = lappend(res, w);
295  }
296 
297  return res;
298 }
List * lappend(List *list, void *datum)
Definition: list.c:338
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().

◆ LogicalRepSyncTableStart()

char* LogicalRepSyncTableStart ( XLogRecPtr origin_startpos)

Definition at line 1229 of file tablesync.c.

1230 {
1231  char *slotname;
1232  char *err;
1233  char relstate;
1234  XLogRecPtr relstate_lsn;
1235  Relation rel;
1236  AclResult aclresult;
1238  char originname[NAMEDATALEN];
1239  RepOriginId originid;
1240 
1241  /* Check the state of the table synchronization. */
1245  &relstate_lsn);
1247 
1249  MyLogicalRepWorker->relstate = relstate;
1250  MyLogicalRepWorker->relstate_lsn = relstate_lsn;
1252 
1253  /*
1254  * If synchronization is already done or no longer necessary, exit now
1255  * that we've updated shared memory state.
1256  */
1257  switch (relstate)
1258  {
1259  case SUBREL_STATE_SYNCDONE:
1260  case SUBREL_STATE_READY:
1261  case SUBREL_STATE_UNKNOWN:
1262  finish_sync_worker(); /* doesn't return */
1263  }
1264 
1265  /* Calculate the name of the tablesync slot. */
1266  slotname = (char *) palloc(NAMEDATALEN);
1269  slotname,
1270  NAMEDATALEN);
1271 
1272  /*
1273  * Here we use the slot name instead of the subscription name as the
1274  * application_name, so that it is different from the leader apply worker,
1275  * so that synchronous replication can distinguish them.
1276  */
1278  walrcv_connect(MySubscription->conninfo, true, slotname, &err);
1279  if (LogRepWorkerWalRcvConn == NULL)
1280  ereport(ERROR,
1281  (errcode(ERRCODE_CONNECTION_FAILURE),
1282  errmsg("could not connect to the publisher: %s", err)));
1283 
1284  Assert(MyLogicalRepWorker->relstate == SUBREL_STATE_INIT ||
1285  MyLogicalRepWorker->relstate == SUBREL_STATE_DATASYNC ||
1286  MyLogicalRepWorker->relstate == SUBREL_STATE_FINISHEDCOPY);
1287 
1288  /* Assign the origin tracking record name. */
1291  originname,
1292  sizeof(originname));
1293 
1294  if (MyLogicalRepWorker->relstate == SUBREL_STATE_DATASYNC)
1295  {
1296  /*
1297  * We have previously errored out before finishing the copy so the
1298  * replication slot might exist. We want to remove the slot if it
1299  * already exists and proceed.
1300  *
1301  * XXX We could also instead try to drop the slot, last time we failed
1302  * but for that, we might need to clean up the copy state as it might
1303  * be in the middle of fetching the rows. Also, if there is a network
1304  * breakdown then it wouldn't have succeeded so trying it next time
1305  * seems like a better bet.
1306  */
1308  }
1309  else if (MyLogicalRepWorker->relstate == SUBREL_STATE_FINISHEDCOPY)
1310  {
1311  /*
1312  * The COPY phase was previously done, but tablesync then crashed
1313  * before it was able to finish normally.
1314  */
1316 
1317  /*
1318  * The origin tracking name must already exist. It was created first
1319  * time this tablesync was launched.
1320  */
1321  originid = replorigin_by_name(originname, false);
1322  replorigin_session_setup(originid, 0);
1323  replorigin_session_origin = originid;
1324  *origin_startpos = replorigin_session_get_progress(false);
1325 
1327 
1328  goto copy_table_done;
1329  }
1330 
1332  MyLogicalRepWorker->relstate = SUBREL_STATE_DATASYNC;
1335 
1336  /* Update the state and make it visible to others. */
1343  pgstat_report_stat(true);
1344 
1346 
1347  /*
1348  * Use a standard write lock here. It might be better to disallow access
1349  * to the table while it's being synchronized. But we don't want to block
1350  * the main apply process from working and it has to open the relation in
1351  * RowExclusiveLock when remapping remote relation id to local one.
1352  */
1354 
1355  /*
1356  * Check that our table sync worker has permission to insert into the
1357  * target table.
1358  */
1359  aclresult = pg_class_aclcheck(RelationGetRelid(rel), GetUserId(),
1360  ACL_INSERT);
1361  if (aclresult != ACLCHECK_OK)
1362  aclcheck_error(aclresult,
1363  get_relkind_objtype(rel->rd_rel->relkind),
1365 
1366  /*
1367  * COPY FROM does not honor RLS policies. That is not a problem for
1368  * subscriptions owned by roles with BYPASSRLS privilege (or superuser,
1369  * who has it implicitly), but other roles should not be able to
1370  * circumvent RLS. Disallow logical replication into RLS enabled
1371  * relations for such roles.
1372  */
1374  ereport(ERROR,
1375  (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
1376  errmsg("user \"%s\" cannot replicate into relation with row-level security enabled: \"%s\"",
1377  GetUserNameFromId(GetUserId(), true),
1378  RelationGetRelationName(rel))));
1379 
1380  /*
1381  * Start a transaction in the remote node in REPEATABLE READ mode. This
1382  * ensures that both the replication slot we create (see below) and the
1383  * COPY are consistent with each other.
1384  */
1386  "BEGIN READ ONLY ISOLATION LEVEL REPEATABLE READ",
1387  0, NULL);
1388  if (res->status != WALRCV_OK_COMMAND)
1389  ereport(ERROR,
1390  (errcode(ERRCODE_CONNECTION_FAILURE),
1391  errmsg("table copy could not start transaction on publisher: %s",
1392  res->err)));
1394 
1395  /*
1396  * Create a new permanent logical decoding slot. This slot will be used
1397  * for the catchup phase after COPY is done, so tell it to use the
1398  * snapshot to make the final data consistent.
1399  */
1401  slotname, false /* permanent */ , false /* two_phase */ ,
1402  CRS_USE_SNAPSHOT, origin_startpos);
1403 
1404  /*
1405  * Setup replication origin tracking. The purpose of doing this before the
1406  * copy is to avoid doing the copy again due to any error in setting up
1407  * origin tracking.
1408  */
1409  originid = replorigin_by_name(originname, true);
1410  if (!OidIsValid(originid))
1411  {
1412  /*
1413  * Origin tracking does not exist, so create it now.
1414  *
1415  * Then advance to the LSN got from walrcv_create_slot. This is WAL
1416  * logged for the purpose of recovery. Locks are to prevent the
1417  * replication origin from vanishing while advancing.
1418  */
1419  originid = replorigin_create(originname);
1420 
1421  LockRelationOid(ReplicationOriginRelationId, RowExclusiveLock);
1422  replorigin_advance(originid, *origin_startpos, InvalidXLogRecPtr,
1423  true /* go backward */ , true /* WAL log */ );
1424  UnlockRelationOid(ReplicationOriginRelationId, RowExclusiveLock);
1425 
1426  replorigin_session_setup(originid, 0);
1427  replorigin_session_origin = originid;
1428  }
1429  else
1430  {
1431  ereport(ERROR,
1433  errmsg("replication origin \"%s\" already exists",
1434  originname)));
1435  }
1436 
1437  /* Now do the initial data copy */
1439  copy_table(rel);
1441 
1442  res = walrcv_exec(LogRepWorkerWalRcvConn, "COMMIT", 0, NULL);
1443  if (res->status != WALRCV_OK_COMMAND)
1444  ereport(ERROR,
1445  (errcode(ERRCODE_CONNECTION_FAILURE),
1446  errmsg("table copy could not finish transaction on publisher: %s",
1447  res->err)));
1449 
1450  table_close(rel, NoLock);
1451 
1452  /* Make the copy visible. */
1454 
1455  /*
1456  * Update the persisted state to indicate the COPY phase is done; make it
1457  * visible to others.
1458  */
1461  SUBREL_STATE_FINISHEDCOPY,
1463 
1465 
1466 copy_table_done:
1467 
1468  elog(DEBUG1,
1469  "LogicalRepSyncTableStart: '%s' origin_startpos lsn %X/%X",
1470  originname, LSN_FORMAT_ARGS(*origin_startpos));
1471 
1472  /*
1473  * We are done with the initial data synchronization, update the state.
1474  */
1476  MyLogicalRepWorker->relstate = SUBREL_STATE_SYNCWAIT;
1477  MyLogicalRepWorker->relstate_lsn = *origin_startpos;
1479 
1480  /*
1481  * Finally, wait until the leader apply worker tells us to catch up and
1482  * then return to let LogicalRepApplyLoop do it.
1483  */
1484  wait_for_worker_state_change(SUBREL_STATE_CATCHUP);
1485  return slotname;
1486 }
AclResult
Definition: acl.h:182
@ ACLCHECK_OK
Definition: acl.h:183
void aclcheck_error(AclResult aclerr, ObjectType objtype, const char *objectname)
Definition: aclchk.c:2679
AclResult pg_class_aclcheck(Oid table_oid, Oid roleid, AclMode mode)
Definition: aclchk.c:3931
void ReplicationOriginNameForLogicalRep(Oid suboid, Oid relid, char *originname, Size szoriginname)
Definition: worker.c:454
WalReceiverConn * LogRepWorkerWalRcvConn
Definition: worker.c:312
void UnlockRelationOid(Oid relid, LOCKMODE lockmode)
Definition: lmgr.c:228
void LockRelationOid(Oid relid, LOCKMODE lockmode)
Definition: lmgr.c:109
#define NoLock
Definition: lockdefs.h:34
#define RowExclusiveLock
Definition: lockdefs.h:38
char * GetUserNameFromId(Oid roleid, bool noerr)
Definition: miscinit.c:972
Oid GetUserId(void)
Definition: miscinit.c:502
ObjectType get_relkind_objtype(char relkind)
RepOriginId replorigin_by_name(const char *roname, bool missing_ok)
Definition: origin.c:221
RepOriginId replorigin_create(const char *roname)
Definition: origin.c:252
RepOriginId replorigin_session_origin
Definition: origin.c:156
void replorigin_advance(RepOriginId node, XLogRecPtr remote_commit, XLogRecPtr local_commit, bool go_backward, bool wal_log)
Definition: origin.c:884
void replorigin_session_setup(RepOriginId node, int acquired_by)
Definition: origin.c:1091
XLogRecPtr replorigin_session_get_progress(bool flush)
Definition: origin.c:1230
#define ACL_INSERT
Definition: parsenodes.h:83
#define NAMEDATALEN
char GetSubscriptionRelState(Oid subid, Oid relid, XLogRecPtr *sublsn)
void UpdateSubscriptionRelState(Oid subid, Oid relid, char state, XLogRecPtr sublsn)
#define InvalidOid
Definition: postgres_ext.h:36
#define RelationGetRelid(relation)
Definition: rel.h:501
#define RelationGetRelationName(relation)
Definition: rel.h:535
int check_enable_rls(Oid relid, Oid checkAsUser, bool noError)
Definition: rls.c:52
@ RLS_ENABLED
Definition: rls.h:45
Snapshot GetTransactionSnapshot(void)
Definition: snapmgr.c:251
void PushActiveSnapshot(Snapshot snapshot)
Definition: snapmgr.c:683
void PopActiveSnapshot(void)
Definition: snapmgr.c:778
#define SpinLockRelease(lock)
Definition: spin.h:64
#define SpinLockAcquire(lock)
Definition: spin.h:62
#define ERRCODE_DUPLICATE_OBJECT
Definition: streamutil.c:32
Form_pg_class rd_rel
Definition: rel.h:110
void ReplicationSlotDropAtPubNode(WalReceiverConn *wrconn, char *slotname, bool missing_ok)
void table_close(Relation relation, LOCKMODE lockmode)
Definition: table.c:126
Relation table_open(Oid relationId, LOCKMODE lockmode)
Definition: table.c:40
static bool wait_for_worker_state_change(char expected_state)
Definition: tablesync.c:222
void ReplicationSlotNameForTablesync(Oid suboid, Oid relid, char *syncslotname, Size szslot)
Definition: tablesync.c:1213
static void copy_table(Relation rel)
Definition: tablesync.c:1083
#define walrcv_create_slot(conn, slotname, temporary, two_phase, snapshot_action, lsn)
Definition: walreceiver.h:430
#define walrcv_connect(conninfo, logical, appname, err)
Definition: walreceiver.h:408
@ WALRCV_OK_COMMAND
Definition: walreceiver.h:205
static void walrcv_clear_result(WalRcvExecResult *walres)
Definition: walreceiver.h:440
#define walrcv_exec(conn, exec, nRetTypes, retTypes)
Definition: walreceiver.h:434
@ CRS_USE_SNAPSHOT
Definition: walsender.h:24
void CommandCounterIncrement(void)
Definition: xact.c:1078
uint16 RepOriginId
Definition: xlogdefs.h:65
uint64 XLogRecPtr
Definition: xlogdefs.h:21

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

Referenced by start_table_sync().

◆ maybe_reread_subscription()

void maybe_reread_subscription ( void  )

Definition at line 3831 of file worker.c.

3832 {
3833  MemoryContext oldctx;
3835  bool started_tx = false;
3836 
3837  /* When cache state is valid there is nothing to do here. */
3838  if (MySubscriptionValid)
3839  return;
3840 
3841  /* This function might be called inside or outside of transaction. */
3842  if (!IsTransactionState())
3843  {
3845  started_tx = true;
3846  }
3847 
3848  /* Ensure allocations in permanent context. */
3850 
3852 
3853  /*
3854  * Exit if the subscription was removed. This normally should not happen
3855  * as the worker gets killed during DROP SUBSCRIPTION.
3856  */
3857  if (!newsub)
3858  {
3859  ereport(LOG,
3860  /* translator: first %s is the name of logical replication worker */
3861  (errmsg("%s for subscription \"%s\" will stop because the subscription was removed",
3863 
3864  /* Ensure we remove no-longer-useful entry for worker's start time */
3867  proc_exit(0);
3868  }
3869 
3870  /* Exit if the subscription was disabled. */
3871  if (!newsub->enabled)
3872  {
3873  ereport(LOG,
3874  /* translator: first %s is the name of logical replication worker */
3875  (errmsg("%s for subscription \"%s\" will stop because the subscription was disabled",
3877 
3879  }
3880 
3881  /* !slotname should never happen when enabled is true. */
3882  Assert(newsub->slotname);
3883 
3884  /* two-phase should not be altered */
3885  Assert(newsub->twophasestate == MySubscription->twophasestate);
3886 
3887  /*
3888  * Exit if any parameter that affects the remote connection was changed.
3889  * The launcher will start a new worker but note that the parallel apply
3890  * worker won't restart if the streaming option's value is changed from
3891  * 'parallel' to any other value or the server decides not to stream the
3892  * in-progress transaction.
3893  */
3894  if (strcmp(newsub->conninfo, MySubscription->conninfo) != 0 ||
3895  strcmp(newsub->name, MySubscription->name) != 0 ||
3896  strcmp(newsub->slotname, MySubscription->slotname) != 0 ||
3897  newsub->binary != MySubscription->binary ||
3898  newsub->stream != MySubscription->stream ||
3899  strcmp(newsub->origin, MySubscription->origin) != 0 ||
3900  newsub->owner != MySubscription->owner ||
3901  !equal(newsub->publications, MySubscription->publications))
3902  {
3904  ereport(LOG,
3905  (errmsg("logical replication parallel apply worker for subscription \"%s\" will stop because of a parameter change",
3906  MySubscription->name)));
3907  else
3908  ereport(LOG,
3909  /* translator: first %s is the name of logical replication worker */
3910  (errmsg("%s for subscription \"%s\" will restart because of a parameter change",
3912 
3914  }
3915 
3916  /* Check for other changes that should never happen too. */
3917  if (newsub->dbid != MySubscription->dbid)
3918  {
3919  elog(ERROR, "subscription %u changed unexpectedly",
3921  }
3922 
3923  /* Clean old subscription info and switch to new one. */
3926 
3927  MemoryContextSwitchTo(oldctx);
3928 
3929  /* Change synchronous commit according to the user's wishes */
3930  SetConfigOption("synchronous_commit", MySubscription->synccommit,
3932 
3933  if (started_tx)
3935 
3936  MySubscriptionValid = true;
3937 }
bool equal(const void *a, const void *b)
Definition: equalfuncs.c:225
void FreeSubscription(Subscription *sub)
static color newsub(struct colormap *cm, color co)
Definition: regc_color.c:389
bool IsTransactionState(void)
Definition: xact.c:378

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

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

◆ pa_allocate_worker()

void pa_allocate_worker ( TransactionId  xid)

Definition at line 469 of file applyparallelworker.c.

470 {
471  bool found;
472  ParallelApplyWorkerInfo *winfo = NULL;
474 
475  if (!pa_can_start())
476  return;
477 
478  winfo = pa_launch_parallel_worker();
479  if (!winfo)
480  return;
481 
482  /* First time through, initialize parallel apply worker state hashtable. */
484  {
485  HASHCTL ctl;
486 
487  MemSet(&ctl, 0, sizeof(ctl));
488  ctl.keysize = sizeof(TransactionId);
489  ctl.entrysize = sizeof(ParallelApplyWorkerEntry);
490  ctl.hcxt = ApplyContext;
491 
492  ParallelApplyTxnHash = hash_create("logical replication parallel apply workers hash",
493  16, &ctl,
495  }
496 
497  /* Create an entry for the requested transaction. */
498  entry = hash_search(ParallelApplyTxnHash, &xid, HASH_ENTER, &found);
499  if (found)
500  elog(ERROR, "hash table corrupted");
501 
502  /* Update the transaction information in shared memory. */
503  SpinLockAcquire(&winfo->shared->mutex);
505  winfo->shared->xid = xid;
506  SpinLockRelease(&winfo->shared->mutex);
507 
508  winfo->in_use = true;
509  winfo->serialize_changes = false;
510  entry->winfo = winfo;
511  entry->xid = xid;
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:1004
uint32 TransactionId
Definition: c.h:636
void * hash_search(HTAB *hashp, const void *keyPtr, HASHACTION action, bool *foundPtr)
Definition: dynahash.c:953
HTAB * hash_create(const char *tabname, long nelem, const HASHCTL *info, int flags)
Definition: dynahash.c:350
@ 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
Size keysize
Definition: hsearch.h:75
Size entrysize
Definition: hsearch.h:76
MemoryContext hcxt
Definition: hsearch.h:86
ParallelApplyWorkerInfo * winfo
ParallelApplyWorkerShared * shared
ParallelTransState xact_state

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

Referenced by apply_handle_stream_start().

◆ pa_decr_and_wait_stream_block()

void pa_decr_and_wait_stream_block ( void  )

Definition at line 1588 of file applyparallelworker.c.

1589 {
1591 
1592  /*
1593  * It is only possible to not have any pending stream chunks when we are
1594  * applying spooled messages.
1595  */
1597  {
1599  return;
1600 
1601  elog(ERROR, "invalid pending streaming chunk 0");
1602  }
1603 
1605  {
1608  }
1609 }
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:396
static uint32 pg_atomic_read_u32(volatile pg_atomic_uint32 *ptr)
Definition: atomics.h:236
#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 631 of file applyparallelworker.c.

632 {
633  ListCell *lc;
634 
635  foreach(lc, ParallelApplyWorkerPool)
636  {
638 
640  winfo->error_mq_handle = NULL;
641  }
642 }
static List * ParallelApplyWorkerPool
#define lfirst(lc)
Definition: pg_list.h:172
void shm_mq_detach(shm_mq_handle *mqh)
Definition: shm_mq.c:844
shm_mq_handle * error_mq_handle

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 1399 of file applyparallelworker.c.

1400 {
1401  /*
1402  * We don't need to free this explicitly as the allocated memory will be
1403  * freed at the transaction end.
1404  */
1405  subxactlist = NIL;
1406 }
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 1143 of file applyparallelworker.c.

1144 {
1145  int rc;
1146  shm_mq_result result;
1147  TimestampTz startTime = 0;
1148 
1150  Assert(!winfo->serialize_changes);
1151 
1152  /*
1153  * We don't try to send data to parallel worker for 'immediate' mode. This
1154  * is primarily used for testing purposes.
1155  */
1157  return false;
1158 
1159 /*
1160  * This timeout is a bit arbitrary but testing revealed that it is sufficient
1161  * to send the message unless the parallel apply worker is waiting on some
1162  * lock or there is a serious resource crunch. See the comments atop this file
1163  * to know why we are using a non-blocking way to send the message.
1164  */
1165 #define SHM_SEND_RETRY_INTERVAL_MS 1000
1166 #define SHM_SEND_TIMEOUT_MS (10000 - SHM_SEND_RETRY_INTERVAL_MS)
1167 
1168  for (;;)
1169  {
1170  result = shm_mq_send(winfo->mq_handle, nbytes, data, true, true);
1171 
1172  if (result == SHM_MQ_SUCCESS)
1173  return true;
1174  else if (result == SHM_MQ_DETACHED)
1175  ereport(ERROR,
1176  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1177  errmsg("could not send data to shared-memory queue")));
1178 
1179  Assert(result == SHM_MQ_WOULD_BLOCK);
1180 
1181  /* Wait before retrying. */
1182  rc = WaitLatch(MyLatch,
1186 
1187  if (rc & WL_LATCH_SET)
1188  {
1191  }
1192 
1193  if (startTime == 0)
1194  startTime = GetCurrentTimestamp();
1195  else if (TimestampDifferenceExceeds(startTime, GetCurrentTimestamp(),
1197  return false;
1198  }
1199 }
#define SHM_SEND_TIMEOUT_MS
#define SHM_SEND_RETRY_INTERVAL_MS
#define unlikely(x)
Definition: c.h:295
struct Latch * MyLatch
Definition: globals.c:58
void ResetLatch(Latch *latch)
Definition: latch.c:699
int WaitLatch(Latch *latch, int wakeEvents, long timeout, uint32 wait_event_info)
Definition: latch.c:492
#define WL_TIMEOUT
Definition: latch.h:128
#define WL_EXIT_ON_PM_DEATH
Definition: latch.h:130
#define WL_LATCH_SET
Definition: latch.h:125
const void * data
int logical_replication_mode
@ LOGICAL_REP_MODE_IMMEDIATE
Definition: reorderbuffer.h:28
shm_mq_result shm_mq_send(shm_mq_handle *mqh, Size nbytes, const void *data, bool nowait, bool force_flush)
Definition: shm_mq.c:330
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
@ WAIT_EVENT_LOGICAL_PARALLEL_APPLY_STATE_CHANGE
Definition: wait_event.h:109

References Assert(), CHECK_FOR_INTERRUPTS, data, ereport, errcode(), errmsg(), ERROR, GetCurrentTimestamp(), IsTransactionState(), LOGICAL_REP_MODE_IMMEDIATE, logical_replication_mode, ParallelApplyWorkerInfo::mq_handle, MyLatch, ResetLatch(), ParallelApplyWorkerInfo::serialize_changes, SHM_MQ_DETACHED, shm_mq_send(), SHM_MQ_SUCCESS, SHM_MQ_WOULD_BLOCK, SHM_SEND_RETRY_INTERVAL_MS, SHM_SEND_TIMEOUT_MS, TimestampDifferenceExceeds(), unlikely, WAIT_EVENT_LOGICAL_PARALLEL_APPLY_STATE_CHANGE, 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 1331 of file applyparallelworker.c.

1332 {
1333  stream_apply_worker = winfo;
1334 }

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

◆ pa_start_subtrans()

void pa_start_subtrans ( TransactionId  current_xid,
TransactionId  top_xid 
)

Definition at line 1359 of file applyparallelworker.c.

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

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 1413 of file applyparallelworker.c.

1414 {
1415  TransactionId xid = abort_data->xid;
1416  TransactionId subxid = abort_data->subxid;
1417 
1418  /*
1419  * Update origin state so we can restart streaming from correct position
1420  * in case of crash.
1421  */
1424 
1425  /*
1426  * If the two XIDs are the same, it's in fact abort of toplevel xact, so
1427  * just free the subxactlist.
1428  */
1429  if (subxid == xid)
1430  {
1432 
1433  /*
1434  * Release the lock as we might be processing an empty streaming
1435  * transaction in which case the lock won't be released during
1436  * transaction rollback.
1437  *
1438  * Note that it's ok to release the transaction lock before aborting
1439  * the transaction because even if the parallel apply worker dies due
1440  * to crash or some other reason, such a transaction would still be
1441  * considered aborted.
1442  */
1444 
1446 
1447  if (IsTransactionBlock())
1448  {
1449  EndTransactionBlock(false);
1451  }
1452 
1454 
1456  }
1457  else
1458  {
1459  /* OK, so it's a subxact. Rollback to the savepoint. */
1460  int i;
1461  char spname[NAMEDATALEN];
1462 
1463  pa_savepoint_name(MySubscription->oid, subxid, spname, sizeof(spname));
1464 
1465  elog(DEBUG1, "rolling back to savepoint %s in logical replication parallel apply worker", spname);
1466 
1467  /*
1468  * Search the subxactlist, determine the offset tracked for the
1469  * subxact, and truncate the list.
1470  *
1471  * Note that for an empty sub-transaction we won't find the subxid
1472  * here.
1473  */
1474  for (i = list_length(subxactlist) - 1; i >= 0; i--)
1475  {
1477 
1478  if (xid_tmp == subxid)
1479  {
1480  RollbackToSavepoint(spname);
1483  break;
1484  }
1485  }
1486  }
1487 }
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:630
#define AccessExclusiveLock
Definition: lockdefs.h:43
TimestampTz replorigin_session_origin_timestamp
Definition: origin.c:158
XLogRecPtr replorigin_session_origin_lsn
Definition: origin.c:157
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:4420
bool EndTransactionBlock(bool chain)
Definition: xact.c:3897
void AbortCurrentTransaction(void)
Definition: xact.c:3312

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 1208 of file applyparallelworker.c.

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

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 1615 of file applyparallelworker.c.

1616 {
1618 
1619  /*
1620  * Unlock the shared object lock so that parallel apply worker can
1621  * continue to receive and apply changes.
1622  */
1624 
1625  /*
1626  * Wait for that worker to finish. This is necessary to maintain commit
1627  * order which avoids failures due to transaction dependencies and
1628  * deadlocks.
1629  */
1630  pa_wait_for_xact_finish(winfo);
1631 
1632  if (!XLogRecPtrIsInvalid(remote_lsn))
1633  store_flush_position(remote_lsn, winfo->shared->last_commit_end);
1634 
1635  pa_free_worker(winfo);
1636 }
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:3408

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

647 {
648  /*
649  * Skip for parallel apply workers because they only operate on tables
650  * that are in a READY state. See pa_can_start() and
651  * should_apply_changes_for_rel().
652  */
654  return;
655 
656  if (am_tablesync_worker())
657  process_syncing_tables_for_sync(current_lsn);
658  else
660 }
static void process_syncing_tables_for_apply(XLogRecPtr current_lsn)
Definition: tablesync.c:408
static void process_syncing_tables_for_sync(XLogRecPtr current_lsn)
Definition: tablesync.c:285

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

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

◆ ReplicationOriginNameForLogicalRep()

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

Definition at line 454 of file worker.c.

456 {
457  if (OidIsValid(relid))
458  {
459  /* Replication origin name for tablesync workers. */
460  snprintf(originname, szoriginname, "pg_%u_%u", suboid, relid);
461  }
462  else
463  {
464  /* Replication origin name for non-tablesync workers. */
465  snprintf(originname, szoriginname, "pg_%u", suboid);
466  }
467 }

References OidIsValid, and snprintf.

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

◆ set_apply_error_context_origin()

void set_apply_error_context_origin ( char *  originname)

Definition at line 5004 of file worker.c.

5005 {
5007  originname);
5008 }
char * MemoryContextStrdup(MemoryContext context, const char *string)
Definition: mcxt.c:1611

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

Referenced by ApplyWorkerMain(), and ParallelApplyWorkerMain().

◆ store_flush_position()

void store_flush_position ( XLogRecPtr  remote_lsn,
XLogRecPtr  local_lsn 
)

Definition at line 3408 of file worker.c.

3409 {
3410  FlushPosition *flushpos;
3411 
3412  /*
3413  * Skip for parallel apply workers, because the lsn_mapping is maintained
3414  * by the leader apply worker.
3415  */
3417  return;
3418 
3419  /* Need to do this in permanent context */
3421 
3422  /* Track commit lsn */
3423  flushpos = (FlushPosition *) palloc(sizeof(FlushPosition));
3424  flushpos->local_end = local_lsn;
3425  flushpos->remote_end = remote_lsn;
3426 
3427  dlist_push_tail(&lsn_mapping, &flushpos->node);
3429 }
static dlist_head lsn_mapping
Definition: worker.c:220
static void dlist_push_tail(dlist_head *head, dlist_node *node)
Definition: ilist.h:364
dlist_node node
Definition: worker.c:215
XLogRecPtr remote_end
Definition: worker.c:217
XLogRecPtr local_end
Definition: worker.c:216

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

4158 {
4159  char path[MAXPGPATH];
4160 
4161  /* Delete the changes file. */
4162  changes_filename(path, subid, xid);
4164 
4165  /* Delete the subxact file, if it exists. */
4166  subxact_filename(path, subid, xid);
4168 }
static void subxact_filename(char *path, Oid subid, TransactionId xid)
Definition: worker.c:4136
void BufFileDeleteFileSet(FileSet *fileset, const char *name, bool missing_ok)
Definition: buffile.c:359

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

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

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

1619 {
1620  /*
1621  * Serialize information about subxacts for the toplevel transaction, then
1622  * close the stream messages spool file.
1623  */
1626 
1627  /* We must be in a valid transaction state */
1629 
1630  /* Commit the per-stream transaction */
1632 
1633  /* Reset per-stream context */
1635 }
static void subxact_info_write(Oid subid, TransactionId xid)
Definition: worker.c:3958
static MemoryContext LogicalStreamingContext
Definition: worker.c:310

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

1584 {
1585  Relation rel;
1586  HeapTuple tup;
1587  bool nulls[Natts_pg_subscription];
1588  bool replaces[Natts_pg_subscription];
1589  Datum values[Natts_pg_subscription];
1590 
1592  new_state == LOGICALREP_TWOPHASE_STATE_PENDING ||
1593  new_state == LOGICALREP_TWOPHASE_STATE_ENABLED);
1594 
1595  rel = table_open(SubscriptionRelationId, RowExclusiveLock);
1597  if (!HeapTupleIsValid(tup))
1598  elog(ERROR,
1599  "cache lookup failed for subscription oid %u",
1600  suboid);
1601 
1602  /* Form a new tuple. */
1603  memset(values, 0, sizeof(values));
1604  memset(nulls, false, sizeof(nulls));
1605  memset(replaces, false, sizeof(replaces));
1606 
1607  /* And update/set two_phase state */
1608  values[Anum_pg_subscription_subtwophasestate - 1] = CharGetDatum(new_state);
1609  replaces[Anum_pg_subscription_subtwophasestate - 1] = true;
1610 
1611  tup = heap_modify_tuple(tup, RelationGetDescr(rel),
1612  values, nulls, replaces);
1613  CatalogTupleUpdate(rel, &tup->t_self, tup);
1614 
1615  heap_freetuple(tup);
1617 }
static Datum values[MAXATTR]
Definition: bootstrap.c:156
HeapTuple heap_modify_tuple(HeapTuple tuple, TupleDesc tupleDesc, Datum *replValues, bool *replIsnull, bool *doReplace)
Definition: heaptuple.c:1113
void heap_freetuple(HeapTuple htup)
Definition: heaptuple.c:1338
#define HeapTupleIsValid(tuple)
Definition: htup.h:78
void CatalogTupleUpdate(Relation heapRel, ItemPointer otid, HeapTuple tup)
Definition: indexing.c:301
#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:527
ItemPointerData t_self
Definition: htup.h:65
#define SearchSysCacheCopy1(cacheId, key1)
Definition: syscache.h:179

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

Referenced by ApplyWorkerMain(), and CreateSubscription().

Variable Documentation

◆ apply_error_context_stack

PGDLLIMPORT ErrorContextCallback* apply_error_context_stack
extern

Definition at line 304 of file worker.c.

Referenced by HandleParallelApplyMessage(), and LogicalRepApplyLoop().

◆ ApplyContext

◆ ApplyMessageContext

◆ in_remote_transaction

◆ LogRepWorkerWalRcvConn

◆ MyLogicalRepWorker

◆ MyParallelShared

◆ MySubscription