PostgreSQL Source Code  git master
applyparallelworker.c File Reference
#include "postgres.h"
#include "libpq/pqformat.h"
#include "libpq/pqmq.h"
#include "pgstat.h"
#include "postmaster/interrupt.h"
#include "replication/logicallauncher.h"
#include "replication/logicalworker.h"
#include "replication/origin.h"
#include "replication/worker_internal.h"
#include "storage/ipc.h"
#include "storage/lmgr.h"
#include "tcop/tcopprot.h"
#include "utils/inval.h"
#include "utils/memutils.h"
#include "utils/syscache.h"
Include dependency graph for applyparallelworker.c:

Go to the source code of this file.

Data Structures

struct  ParallelApplyWorkerEntry
 

Macros

#define PG_LOGICAL_APPLY_SHM_MAGIC   0x787ca067
 
#define PARALLEL_APPLY_KEY_SHARED   1
 
#define PARALLEL_APPLY_KEY_MQ   2
 
#define PARALLEL_APPLY_KEY_ERROR_QUEUE   3
 
#define DSM_QUEUE_SIZE   (16 * 1024 * 1024)
 
#define DSM_ERROR_QUEUE_SIZE   (16 * 1024)
 
#define SIZE_STATS_MESSAGE   (2 * sizeof(XLogRecPtr) + sizeof(TimestampTz))
 
#define PARALLEL_APPLY_LOCK_STREAM   0
 
#define PARALLEL_APPLY_LOCK_XACT   1
 
#define SHM_SEND_RETRY_INTERVAL_MS   1000
 
#define SHM_SEND_TIMEOUT_MS   (10000 - SHM_SEND_RETRY_INTERVAL_MS)
 

Typedefs

typedef struct ParallelApplyWorkerEntry ParallelApplyWorkerEntry
 

Functions

static void pa_free_worker_info (ParallelApplyWorkerInfo *winfo)
 
static ParallelTransState pa_get_xact_state (ParallelApplyWorkerShared *wshared)
 
static PartialFileSetState pa_get_fileset_state (void)
 
static bool pa_can_start (void)
 
static bool pa_setup_dsm (ParallelApplyWorkerInfo *winfo)
 
static ParallelApplyWorkerInfopa_launch_parallel_worker (void)
 
void pa_allocate_worker (TransactionId xid)
 
ParallelApplyWorkerInfopa_find_worker (TransactionId xid)
 
static void pa_free_worker (ParallelApplyWorkerInfo *winfo)
 
void pa_detach_all_error_mq (void)
 
static bool pa_has_spooled_message_pending ()
 
static bool pa_process_spooled_messages_if_required (void)
 
static void ProcessParallelApplyInterrupts (void)
 
static void LogicalParallelApplyLoop (shm_mq_handle *mqh)
 
static void pa_shutdown (int code, Datum arg)
 
void ParallelApplyWorkerMain (Datum main_arg)
 
void HandleParallelApplyMessageInterrupt (void)
 
static void HandleParallelApplyMessage (StringInfo msg)
 
void HandleParallelApplyMessages (void)
 
bool pa_send_data (ParallelApplyWorkerInfo *winfo, Size nbytes, const void *data)
 
void pa_switch_to_partial_serialize (ParallelApplyWorkerInfo *winfo, bool stream_locked)
 
static void pa_wait_for_xact_state (ParallelApplyWorkerInfo *winfo, ParallelTransState xact_state)
 
static void pa_wait_for_xact_finish (ParallelApplyWorkerInfo *winfo)
 
void pa_set_xact_state (ParallelApplyWorkerShared *wshared, ParallelTransState xact_state)
 
void pa_set_stream_apply_worker (ParallelApplyWorkerInfo *winfo)
 
static void pa_savepoint_name (Oid suboid, TransactionId xid, char *spname, Size szsp)
 
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)
 

Variables

static HTABParallelApplyTxnHash = NULL
 
static ListParallelApplyWorkerPool = NIL
 
ParallelApplyWorkerSharedMyParallelShared = NULL
 
volatile sig_atomic_t ParallelApplyMessagePending = false
 
static ParallelApplyWorkerInfostream_apply_worker = NULL
 
static Listsubxactlist = NIL
 

Macro Definition Documentation

◆ DSM_ERROR_QUEUE_SIZE

#define DSM_ERROR_QUEUE_SIZE   (16 * 1024)

Definition at line 195 of file applyparallelworker.c.

◆ DSM_QUEUE_SIZE

#define DSM_QUEUE_SIZE   (16 * 1024 * 1024)

Definition at line 187 of file applyparallelworker.c.

◆ PARALLEL_APPLY_KEY_ERROR_QUEUE

#define PARALLEL_APPLY_KEY_ERROR_QUEUE   3

Definition at line 184 of file applyparallelworker.c.

◆ PARALLEL_APPLY_KEY_MQ

#define PARALLEL_APPLY_KEY_MQ   2

Definition at line 183 of file applyparallelworker.c.

◆ PARALLEL_APPLY_KEY_SHARED

#define PARALLEL_APPLY_KEY_SHARED   1

Definition at line 182 of file applyparallelworker.c.

◆ PARALLEL_APPLY_LOCK_STREAM

#define PARALLEL_APPLY_LOCK_STREAM   0

Definition at line 209 of file applyparallelworker.c.

◆ PARALLEL_APPLY_LOCK_XACT

#define PARALLEL_APPLY_LOCK_XACT   1

Definition at line 210 of file applyparallelworker.c.

◆ PG_LOGICAL_APPLY_SHM_MAGIC

#define PG_LOGICAL_APPLY_SHM_MAGIC   0x787ca067

Definition at line 175 of file applyparallelworker.c.

◆ SHM_SEND_RETRY_INTERVAL_MS

#define SHM_SEND_RETRY_INTERVAL_MS   1000

◆ SHM_SEND_TIMEOUT_MS

#define SHM_SEND_TIMEOUT_MS   (10000 - SHM_SEND_RETRY_INTERVAL_MS)

◆ SIZE_STATS_MESSAGE

#define SIZE_STATS_MESSAGE   (2 * sizeof(XLogRecPtr) + sizeof(TimestampTz))

Definition at line 203 of file applyparallelworker.c.

Typedef Documentation

◆ ParallelApplyWorkerEntry

Function Documentation

◆ HandleParallelApplyMessage()

static void HandleParallelApplyMessage ( StringInfo  msg)
static

Definition at line 1005 of file applyparallelworker.c.

1006 {
1007  char msgtype;
1008 
1009  msgtype = pq_getmsgbyte(msg);
1010 
1011  switch (msgtype)
1012  {
1013  case 'E': /* ErrorResponse */
1014  {
1015  ErrorData edata;
1016 
1017  /* Parse ErrorResponse. */
1018  pq_parse_errornotice(msg, &edata);
1019 
1020  /*
1021  * If desired, add a context line to show that this is a
1022  * message propagated from a parallel apply worker. Otherwise,
1023  * it can sometimes be confusing to understand what actually
1024  * happened.
1025  */
1026  if (edata.context)
1027  edata.context = psprintf("%s\n%s", edata.context,
1028  _("logical replication parallel apply worker"));
1029  else
1030  edata.context = pstrdup(_("logical replication parallel apply worker"));
1031 
1032  /*
1033  * Context beyond that should use the error context callbacks
1034  * that were in effect in LogicalRepApplyLoop().
1035  */
1037 
1038  /*
1039  * The actual error must have been reported by the parallel
1040  * apply worker.
1041  */
1042  ereport(ERROR,
1043  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1044  errmsg("logical replication parallel apply worker exited due to error"),
1045  errcontext("%s", edata.context)));
1046  }
1047 
1048  /*
1049  * Don't need to do anything about NoticeResponse and
1050  * NotifyResponse as the logical replication worker doesn't need
1051  * to send messages to the client.
1052  */
1053  case 'N':
1054  case 'A':
1055  break;
1056 
1057  default:
1058  elog(ERROR, "unrecognized message type received from logical replication parallel apply worker: %c (message length %d bytes)",
1059  msgtype, msg->len);
1060  }
1061 }
ErrorContextCallback * apply_error_context_stack
Definition: worker.c:306
ErrorContextCallback * error_context_stack
Definition: elog.c:95
int errcode(int sqlerrcode)
Definition: elog.c:858
int errmsg(const char *fmt,...)
Definition: elog.c:1069
#define _(x)
Definition: elog.c:91
#define errcontext
Definition: elog.h:196
#define ERROR
Definition: elog.h:39
#define ereport(elevel,...)
Definition: elog.h:149
char * pstrdup(const char *in)
Definition: mcxt.c:1644
int pq_getmsgbyte(StringInfo msg)
Definition: pqformat.c:402
void pq_parse_errornotice(StringInfo msg, ErrorData *edata)
Definition: pqmq.c:216
char * psprintf(const char *fmt,...)
Definition: psprintf.c:46
char * context
Definition: elog.h:443

References _, apply_error_context_stack, ErrorData::context, elog(), ereport, errcode(), errcontext, errmsg(), ERROR, error_context_stack, StringInfoData::len, pq_getmsgbyte(), pq_parse_errornotice(), psprintf(), and pstrdup().

Referenced by HandleParallelApplyMessages().

◆ HandleParallelApplyMessageInterrupt()

void HandleParallelApplyMessageInterrupt ( void  )

Definition at line 993 of file applyparallelworker.c.

994 {
995  InterruptPending = true;
997  SetLatch(MyLatch);
998 }
volatile sig_atomic_t ParallelApplyMessagePending
volatile sig_atomic_t InterruptPending
Definition: globals.c:30
struct Latch * MyLatch
Definition: globals.c:58
void SetLatch(Latch *latch)
Definition: latch.c:605

References InterruptPending, MyLatch, ParallelApplyMessagePending, and SetLatch().

Referenced by procsignal_sigusr1_handler().

◆ HandleParallelApplyMessages()

void HandleParallelApplyMessages ( void  )

Definition at line 1067 of file applyparallelworker.c.

1068 {
1069  ListCell *lc;
1070  MemoryContext oldcontext;
1071 
1072  static MemoryContext hpam_context = NULL;
1073 
1074  /*
1075  * This is invoked from ProcessInterrupts(), and since some of the
1076  * functions it calls contain CHECK_FOR_INTERRUPTS(), there is a potential
1077  * for recursive calls if more signals are received while this runs. It's
1078  * unclear that recursive entry would be safe, and it doesn't seem useful
1079  * even if it is safe, so let's block interrupts until done.
1080  */
1081  HOLD_INTERRUPTS();
1082 
1083  /*
1084  * Moreover, CurrentMemoryContext might be pointing almost anywhere. We
1085  * don't want to risk leaking data into long-lived contexts, so let's do
1086  * our work here in a private context that we can reset on each use.
1087  */
1088  if (!hpam_context) /* first time through? */
1089  hpam_context = AllocSetContextCreate(TopMemoryContext,
1090  "HandleParallelApplyMessages",
1092  else
1093  MemoryContextReset(hpam_context);
1094 
1095  oldcontext = MemoryContextSwitchTo(hpam_context);
1096 
1098 
1099  foreach(lc, ParallelApplyWorkerPool)
1100  {
1102  Size nbytes;
1103  void *data;
1105 
1106  /*
1107  * The leader will detach from the error queue and set it to NULL
1108  * before preparing to stop all parallel apply workers, so we don't
1109  * need to handle error messages anymore. See
1110  * logicalrep_worker_detach.
1111  */
1112  if (!winfo->error_mq_handle)
1113  continue;
1114 
1115  res = shm_mq_receive(winfo->error_mq_handle, &nbytes, &data, true);
1116 
1117  if (res == SHM_MQ_WOULD_BLOCK)
1118  continue;
1119  else if (res == SHM_MQ_SUCCESS)
1120  {
1121  StringInfoData msg;
1122 
1123  initStringInfo(&msg);
1124  appendBinaryStringInfo(&msg, data, nbytes);
1126  pfree(msg.data);
1127  }
1128  else
1129  ereport(ERROR,
1130  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1131  errmsg("lost connection to the logical replication parallel apply worker")));
1132  }
1133 
1134  MemoryContextSwitchTo(oldcontext);
1135 
1136  /* Might as well clear the context on our way out */
1137  MemoryContextReset(hpam_context);
1138 
1140 }
static List * ParallelApplyWorkerPool
static void HandleParallelApplyMessage(StringInfo msg)
size_t Size
Definition: c.h:594
void MemoryContextReset(MemoryContext context)
Definition: mcxt.c:330
void pfree(void *pointer)
Definition: mcxt.c:1456
MemoryContext TopMemoryContext
Definition: mcxt.c:141
#define AllocSetContextCreate
Definition: memutils.h:129
#define ALLOCSET_DEFAULT_SIZES
Definition: memutils.h:153
#define RESUME_INTERRUPTS()
Definition: miscadmin.h:134
#define HOLD_INTERRUPTS()
Definition: miscadmin.h:132
static MemoryContext MemoryContextSwitchTo(MemoryContext context)
Definition: palloc.h:138
const void * data
#define lfirst(lc)
Definition: pg_list.h:172
shm_mq_result shm_mq_receive(shm_mq_handle *mqh, Size *nbytesp, void **datap, bool nowait)
Definition: shm_mq.c:573
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
void appendBinaryStringInfo(StringInfo str, const void *data, int datalen)
Definition: stringinfo.c:227
void initStringInfo(StringInfo str)
Definition: stringinfo.c:59
shm_mq_handle * error_mq_handle

References ALLOCSET_DEFAULT_SIZES, AllocSetContextCreate, appendBinaryStringInfo(), StringInfoData::data, data, ereport, errcode(), errmsg(), ERROR, ParallelApplyWorkerInfo::error_mq_handle, HandleParallelApplyMessage(), HOLD_INTERRUPTS, initStringInfo(), lfirst, MemoryContextReset(), MemoryContextSwitchTo(), ParallelApplyMessagePending, ParallelApplyWorkerPool, pfree(), res, RESUME_INTERRUPTS, shm_mq_receive(), SHM_MQ_SUCCESS, SHM_MQ_WOULD_BLOCK, and TopMemoryContext.

Referenced by ProcessInterrupts().

◆ LogicalParallelApplyLoop()

static void LogicalParallelApplyLoop ( shm_mq_handle mqh)
static

Definition at line 735 of file applyparallelworker.c.

736 {
737  shm_mq_result shmq_res;
738  ErrorContextCallback errcallback;
740 
741  /*
742  * Init the ApplyMessageContext which we clean up after each replication
743  * protocol message.
744  */
746  "ApplyMessageContext",
748 
749  /*
750  * Push apply error context callback. Fields will be filled while applying
751  * a change.
752  */
753  errcallback.callback = apply_error_callback;
754  errcallback.previous = error_context_stack;
755  error_context_stack = &errcallback;
756 
757  for (;;)
758  {
759  void *data;
760  Size len;
761 
763 
764  /* Ensure we are reading the data into our memory context. */
766 
767  shmq_res = shm_mq_receive(mqh, &len, &data, true);
768 
769  if (shmq_res == SHM_MQ_SUCCESS)
770  {
771  StringInfoData s;
772  int c;
773 
774  if (len == 0)
775  elog(ERROR, "invalid message length");
776 
777  s.cursor = 0;
778  s.maxlen = -1;
779  s.data = (char *) data;
780  s.len = len;
781 
782  /*
783  * The first byte of messages sent from leader apply worker to
784  * parallel apply workers can only be 'w'.
785  */
786  c = pq_getmsgbyte(&s);
787  if (c != 'w')
788  elog(ERROR, "unexpected message \"%c\"", c);
789 
790  /*
791  * Ignore statistics fields that have been updated by the leader
792  * apply worker.
793  *
794  * XXX We can avoid sending the statistics fields from the leader
795  * apply worker but for that, it needs to rebuild the entire
796  * message by removing these fields which could be more work than
797  * simply ignoring these fields in the parallel apply worker.
798  */
800 
801  apply_dispatch(&s);
802  }
803  else if (shmq_res == SHM_MQ_WOULD_BLOCK)
804  {
805  /* Replay the changes from the file, if any. */
807  {
808  int rc;
809 
810  /* Wait for more work. */
811  rc = WaitLatch(MyLatch,
813  1000L,
814  WAIT_EVENT_LOGICAL_PARALLEL_APPLY_MAIN);
815 
816  if (rc & WL_LATCH_SET)
818  }
819  }
820  else
821  {
822  Assert(shmq_res == SHM_MQ_DETACHED);
823 
824  ereport(ERROR,
825  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
826  errmsg("lost connection to the logical replication apply worker")));
827  }
828 
830  MemoryContextSwitchTo(oldcxt);
831  }
832 
833  /* Pop the error context stack. */
834  error_context_stack = errcallback.previous;
835 
836  MemoryContextSwitchTo(oldcxt);
837 }
static void ProcessParallelApplyInterrupts(void)
#define SIZE_STATS_MESSAGE
static bool pa_process_spooled_messages_if_required(void)
MemoryContext ApplyMessageContext
Definition: worker.c:308
void apply_dispatch(StringInfo s)
Definition: worker.c:3292
MemoryContext ApplyContext
Definition: worker.c:309
void apply_error_callback(void *arg)
Definition: worker.c:4893
void ResetLatch(Latch *latch)
Definition: latch.c:697
int WaitLatch(Latch *latch, int wakeEvents, long timeout, uint32 wait_event_info)
Definition: latch.c:490
#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
Assert(fmt[strlen(fmt) - 1] !='\n')
MemoryContext CurrentMemoryContext
Definition: mcxt.c:135
const void size_t len
char * c
@ SHM_MQ_DETACHED
Definition: shm_mq.h:40
struct ErrorContextCallback * previous
Definition: elog.h:295
void(* callback)(void *arg)
Definition: elog.h:296

References ALLOCSET_DEFAULT_SIZES, AllocSetContextCreate, apply_dispatch(), apply_error_callback(), ApplyContext, ApplyMessageContext, Assert(), ErrorContextCallback::callback, CurrentMemoryContext, StringInfoData::cursor, StringInfoData::data, data, elog(), ereport, errcode(), errmsg(), ERROR, error_context_stack, StringInfoData::len, len, StringInfoData::maxlen, MemoryContextReset(), MemoryContextSwitchTo(), MyLatch, pa_process_spooled_messages_if_required(), pq_getmsgbyte(), ErrorContextCallback::previous, ProcessParallelApplyInterrupts(), ResetLatch(), SHM_MQ_DETACHED, shm_mq_receive(), SHM_MQ_SUCCESS, SHM_MQ_WOULD_BLOCK, SIZE_STATS_MESSAGE, WaitLatch(), WL_EXIT_ON_PM_DEATH, WL_LATCH_SET, and WL_TIMEOUT.

Referenced by ParallelApplyWorkerMain().

◆ 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  entry->xid = xid;
513 }
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:1009
uint32 TransactionId
Definition: c.h:641
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
#define SpinLockRelease(lock)
Definition: spin.h:64
#define SpinLockAcquire(lock)
Definition: spin.h:62
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
@ PARALLEL_TRANS_UNKNOWN

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

static bool pa_can_start ( void  )
static

Definition at line 265 of file applyparallelworker.c.

266 {
267  /* Only leader apply workers can start parallel apply workers. */
268  if (!am_leader_apply_worker())
269  return false;
270 
271  /*
272  * It is good to check for any change in the subscription parameter to
273  * avoid the case where for a very long time the change doesn't get
274  * reflected. This can happen when there is a constant flow of streaming
275  * transactions that are handled by parallel apply workers.
276  *
277  * It is better to do it before the below checks so that the latest values
278  * of subscription can be used for the checks.
279  */
281 
282  /*
283  * Don't start a new parallel apply worker if the subscription is not
284  * using parallel streaming mode, or if the publisher does not support
285  * parallel apply.
286  */
288  return false;
289 
290  /*
291  * Don't start a new parallel worker if user has set skiplsn as it's
292  * possible that they want to skip the streaming transaction. For
293  * streaming transactions, we need to serialize the transaction to a file
294  * so that we can get the last LSN of the transaction to judge whether to
295  * skip before starting to apply the change.
296  *
297  * One might think that we could allow parallelism if the first lsn of the
298  * transaction is greater than skiplsn, but we don't send it with the
299  * STREAM START message, and it doesn't seem worth sending the extra eight
300  * bytes with the STREAM START to enable parallelism for this case.
301  */
303  return false;
304 
305  /*
306  * For streaming transactions that are being applied using a parallel
307  * apply worker, we cannot decide whether to apply the change for a
308  * relation that is not in the READY state (see
309  * should_apply_changes_for_rel) as we won't know remote_final_lsn by that
310  * time. So, we don't start the new parallel apply worker in this case.
311  */
312  if (!AllTablesyncsReady())
313  return false;
314 
315  return true;
316 }
void maybe_reread_subscription(void)
Definition: worker.c:3885
Subscription * MySubscription
Definition: worker.c:316
LogicalRepWorker * MyLogicalRepWorker
Definition: launcher.c:61
XLogRecPtr skiplsn
bool AllTablesyncsReady(void)
Definition: tablesync.c:1697
static bool am_leader_apply_worker(void)
#define XLogRecPtrIsInvalid(r)
Definition: xlogdefs.h:29

References AllTablesyncsReady(), am_leader_apply_worker(), maybe_reread_subscription(), MyLogicalRepWorker, MySubscription, LogicalRepWorker::parallel_apply, Subscription::skiplsn, and XLogRecPtrIsInvalid.

Referenced by pa_allocate_worker().

◆ pa_decr_and_wait_stream_block()

void pa_decr_and_wait_stream_block ( void  )

Definition at line 1595 of file applyparallelworker.c.

1596 {
1598 
1599  /*
1600  * It is only possible to not have any pending stream chunks when we are
1601  * applying spooled messages.
1602  */
1604  {
1606  return;
1607 
1608  elog(ERROR, "invalid pending streaming chunk 0");
1609  }
1610 
1612  {
1615  }
1616 }
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
static bool am_parallel_apply_worker(void)

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

624 {
625  ListCell *lc;
626 
627  foreach(lc, ParallelApplyWorkerPool)
628  {
630 
631  if (winfo->error_mq_handle)
632  {
634  winfo->error_mq_handle = NULL;
635  }
636  }
637 }
void shm_mq_detach(shm_mq_handle *mqh)
Definition: shm_mq.c:844

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

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

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

Referenced by get_transaction_apply_action().

◆ pa_free_worker()

static void pa_free_worker ( ParallelApplyWorkerInfo winfo)
static

Definition at line 557 of file applyparallelworker.c.

558 {
560  Assert(winfo->in_use);
562 
563  if (!hash_search(ParallelApplyTxnHash, &winfo->shared->xid, HASH_REMOVE, NULL))
564  elog(ERROR, "hash table corrupted");
565 
566  /*
567  * Stop the worker if there are enough workers in the pool.
568  *
569  * XXX Additionally, we also stop the worker if the leader apply worker
570  * serialize part of the transaction data due to a send timeout. This is
571  * because the message could be partially written to the queue and there
572  * is no way to clean the queue other than resending the message until it
573  * succeeds. Instead of trying to send the data which anyway would have
574  * been serialized and then letting the parallel apply worker deal with
575  * the spurious message, we stop the worker.
576  */
577  if (winfo->serialize_changes ||
580  {
582  pa_free_worker_info(winfo);
583 
584  return;
585  }
586 
587  winfo->in_use = false;
588  winfo->serialize_changes = false;
589 }
static void pa_free_worker_info(ParallelApplyWorkerInfo *winfo)
static ParallelTransState pa_get_xact_state(ParallelApplyWorkerShared *wshared)
@ HASH_REMOVE
Definition: hsearch.h:115
void logicalrep_pa_worker_stop(ParallelApplyWorkerInfo *winfo)
Definition: launcher.c:639
int max_parallel_apply_workers_per_subscription
Definition: launcher.c:59
static int list_length(const List *l)
Definition: pg_list.h:152
@ PARALLEL_TRANS_FINISHED

References am_parallel_apply_worker(), Assert(), elog(), ERROR, HASH_REMOVE, hash_search(), ParallelApplyWorkerInfo::in_use, list_length(), logicalrep_pa_worker_stop(), max_parallel_apply_workers_per_subscription, pa_free_worker_info(), pa_get_xact_state(), PARALLEL_TRANS_FINISHED, ParallelApplyTxnHash, ParallelApplyWorkerPool, ParallelApplyWorkerInfo::serialize_changes, ParallelApplyWorkerInfo::shared, and ParallelApplyWorkerShared::xid.

Referenced by pa_xact_finish().

◆ pa_free_worker_info()

static void pa_free_worker_info ( ParallelApplyWorkerInfo winfo)
static

Definition at line 596 of file applyparallelworker.c.

597 {
598  Assert(winfo);
599 
600  if (winfo->mq_handle)
601  shm_mq_detach(winfo->mq_handle);
602 
603  if (winfo->error_mq_handle)
605 
606  /* Unlink the files with serialized changes. */
607  if (winfo->serialize_changes)
609 
610  if (winfo->dsm_seg)
611  dsm_detach(winfo->dsm_seg);
612 
613  /* Remove from the worker pool. */
615 
616  pfree(winfo);
617 }
void stream_cleanup_files(Oid subid, TransactionId xid)
Definition: worker.c:4210
void dsm_detach(dsm_segment *seg)
Definition: dsm.c:776
List * list_delete_ptr(List *list, void *datum)
Definition: list.c:871
shm_mq_handle * mq_handle

References Assert(), dsm_detach(), ParallelApplyWorkerInfo::dsm_seg, ParallelApplyWorkerInfo::error_mq_handle, list_delete_ptr(), ParallelApplyWorkerInfo::mq_handle, MyLogicalRepWorker, ParallelApplyWorkerPool, pfree(), ParallelApplyWorkerInfo::serialize_changes, ParallelApplyWorkerInfo::shared, shm_mq_detach(), stream_cleanup_files(), LogicalRepWorker::subid, and ParallelApplyWorkerShared::xid.

Referenced by pa_free_worker(), and pa_launch_parallel_worker().

◆ pa_get_fileset_state()

static PartialFileSetState pa_get_fileset_state ( void  )
static

◆ pa_get_xact_state()

static ParallelTransState pa_get_xact_state ( ParallelApplyWorkerShared wshared)
static

Definition at line 1323 of file applyparallelworker.c.

1324 {
1325  ParallelTransState xact_state;
1326 
1327  SpinLockAcquire(&wshared->mutex);
1328  xact_state = wshared->xact_state;
1329  SpinLockRelease(&wshared->mutex);
1330 
1331  return xact_state;
1332 }
ParallelTransState

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

Referenced by pa_free_worker(), pa_wait_for_xact_finish(), and pa_wait_for_xact_state().

◆ pa_has_spooled_message_pending()

static bool pa_has_spooled_message_pending ( )
static

Definition at line 643 of file applyparallelworker.c.

644 {
645  PartialFileSetState fileset_state;
646 
647  fileset_state = pa_get_fileset_state();
648 
649  return (fileset_state != FS_EMPTY);
650 }
static PartialFileSetState pa_get_fileset_state(void)
@ FS_EMPTY

References FS_EMPTY, and pa_get_fileset_state().

Referenced by pa_decr_and_wait_stream_block().

◆ pa_launch_parallel_worker()

static ParallelApplyWorkerInfo* pa_launch_parallel_worker ( void  )
static

Definition at line 404 of file applyparallelworker.c.

405 {
406  MemoryContext oldcontext;
407  bool launched;
409  ListCell *lc;
410 
411  /* Try to get an available parallel apply worker from the worker pool. */
412  foreach(lc, ParallelApplyWorkerPool)
413  {
414  winfo = (ParallelApplyWorkerInfo *) lfirst(lc);
415 
416  if (!winfo->in_use)
417  return winfo;
418  }
419 
420  /*
421  * Start a new parallel apply worker.
422  *
423  * The worker info can be used for the lifetime of the worker process, so
424  * create it in a permanent context.
425  */
426  oldcontext = MemoryContextSwitchTo(ApplyContext);
427 
429 
430  /* Setup shared memory. */
431  if (!pa_setup_dsm(winfo))
432  {
433  MemoryContextSwitchTo(oldcontext);
434  pfree(winfo);
435  return NULL;
436  }
437 
443  InvalidOid,
444  dsm_segment_handle(winfo->dsm_seg));
445 
446  if (launched)
447  {
449  }
450  else
451  {
452  pa_free_worker_info(winfo);
453  winfo = NULL;
454  }
455 
456  MemoryContextSwitchTo(oldcontext);
457 
458  return winfo;
459 }
static bool pa_setup_dsm(ParallelApplyWorkerInfo *winfo)
dsm_handle dsm_segment_handle(dsm_segment *seg)
Definition: dsm.c:1094
bool logicalrep_worker_launch(LogicalRepWorkerType wtype, Oid dbid, Oid subid, const char *subname, Oid userid, Oid relid, dsm_handle subworker_dsm)
Definition: launcher.c:306
List * lappend(List *list, void *datum)
Definition: list.c:338
void * palloc0(Size size)
Definition: mcxt.c:1257
#define InvalidOid
Definition: postgres_ext.h:36
@ WORKERTYPE_PARALLEL_APPLY

References ApplyContext, LogicalRepWorker::dbid, ParallelApplyWorkerInfo::dsm_seg, dsm_segment_handle(), ParallelApplyWorkerInfo::in_use, InvalidOid, lappend(), lfirst, logicalrep_worker_launch(), MemoryContextSwitchTo(), MyLogicalRepWorker, MySubscription, Subscription::name, Subscription::oid, pa_free_worker_info(), pa_setup_dsm(), palloc0(), ParallelApplyWorkerPool, pfree(), LogicalRepWorker::userid, and WORKERTYPE_PARALLEL_APPLY.

Referenced by pa_allocate_worker().

◆ pa_lock_stream()

void pa_lock_stream ( TransactionId  xid,
LOCKMODE  lockmode 
)

◆ pa_lock_transaction()

void pa_lock_transaction ( TransactionId  xid,
LOCKMODE  lockmode 
)

◆ pa_process_spooled_messages_if_required()

static bool pa_process_spooled_messages_if_required ( void  )
static

Definition at line 659 of file applyparallelworker.c.

660 {
661  PartialFileSetState fileset_state;
662 
663  fileset_state = pa_get_fileset_state();
664 
665  if (fileset_state == FS_EMPTY)
666  return false;
667 
668  /*
669  * If the leader apply worker is busy serializing the partial changes then
670  * acquire the stream lock now and wait for the leader worker to finish
671  * serializing the changes. Otherwise, the parallel apply worker won't get
672  * a chance to receive a STREAM_STOP (and acquire the stream lock) until
673  * the leader had serialized all changes which can lead to undetected
674  * deadlock.
675  *
676  * Note that the fileset state can be FS_SERIALIZE_DONE once the leader
677  * worker has finished serializing the changes.
678  */
679  if (fileset_state == FS_SERIALIZE_IN_PROGRESS)
680  {
683 
684  fileset_state = pa_get_fileset_state();
685  }
686 
687  /*
688  * We cannot read the file immediately after the leader has serialized all
689  * changes to the file because there may still be messages in the memory
690  * queue. We will apply all spooled messages the next time we call this
691  * function and that will ensure there are no messages left in the memory
692  * queue.
693  */
694  if (fileset_state == FS_SERIALIZE_DONE)
695  {
697  }
698  else if (fileset_state == FS_READY)
699  {
704  }
705 
706  return true;
707 }
void pa_set_fileset_state(ParallelApplyWorkerShared *wshared, PartialFileSetState fileset_state)
void apply_spooled_messages(FileSet *stream_fileset, TransactionId xid, XLogRecPtr lsn)
Definition: worker.c:2019
@ FS_SERIALIZE_DONE
@ FS_READY
@ FS_SERIALIZE_IN_PROGRESS
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28

References AccessShareLock, apply_spooled_messages(), ParallelApplyWorkerShared::fileset, FS_EMPTY, FS_READY, FS_SERIALIZE_DONE, FS_SERIALIZE_IN_PROGRESS, InvalidXLogRecPtr, MyParallelShared, pa_get_fileset_state(), pa_lock_stream(), pa_set_fileset_state(), pa_unlock_stream(), and ParallelApplyWorkerShared::xid.

Referenced by LogicalParallelApplyLoop().

◆ pa_reset_subtrans()

void pa_reset_subtrans ( void  )

Definition at line 1406 of file applyparallelworker.c.

1407 {
1408  /*
1409  * We don't need to free this explicitly as the allocated memory will be
1410  * freed at the transaction end.
1411  */
1412  subxactlist = NIL;
1413 }
static List * subxactlist
#define NIL
Definition: pg_list.h:68

References NIL, and subxactlist.

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

◆ pa_savepoint_name()

static void pa_savepoint_name ( Oid  suboid,
TransactionId  xid,
char *  spname,
Size  szsp 
)
static

Definition at line 1352 of file applyparallelworker.c.

1353 {
1354  snprintf(spname, szsp, "pg_sp_%u_%u", suboid, xid);
1355 }
#define snprintf
Definition: port.h:238

References snprintf.

Referenced by pa_start_subtrans(), and pa_stream_abort().

◆ pa_send_data()

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

Definition at line 1150 of file applyparallelworker.c.

1151 {
1152  int rc;
1153  shm_mq_result result;
1154  TimestampTz startTime = 0;
1155 
1157  Assert(!winfo->serialize_changes);
1158 
1159  /*
1160  * We don't try to send data to parallel worker for 'immediate' mode. This
1161  * is primarily used for testing purposes.
1162  */
1164  return false;
1165 
1166 /*
1167  * This timeout is a bit arbitrary but testing revealed that it is sufficient
1168  * to send the message unless the parallel apply worker is waiting on some
1169  * lock or there is a serious resource crunch. See the comments atop this file
1170  * to know why we are using a non-blocking way to send the message.
1171  */
1172 #define SHM_SEND_RETRY_INTERVAL_MS 1000
1173 #define SHM_SEND_TIMEOUT_MS (10000 - SHM_SEND_RETRY_INTERVAL_MS)
1174 
1175  for (;;)
1176  {
1177  result = shm_mq_send(winfo->mq_handle, nbytes, data, true, true);
1178 
1179  if (result == SHM_MQ_SUCCESS)
1180  return true;
1181  else if (result == SHM_MQ_DETACHED)
1182  ereport(ERROR,
1183  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1184  errmsg("could not send data to shared-memory queue")));
1185 
1186  Assert(result == SHM_MQ_WOULD_BLOCK);
1187 
1188  /* Wait before retrying. */
1189  rc = WaitLatch(MyLatch,
1192  WAIT_EVENT_LOGICAL_APPLY_SEND_DATA);
1193 
1194  if (rc & WL_LATCH_SET)
1195  {
1198  }
1199 
1200  if (startTime == 0)
1201  startTime = GetCurrentTimestamp();
1202  else if (TimestampDifferenceExceeds(startTime, GetCurrentTimestamp(),
1204  return false;
1205  }
1206 }
#define SHM_SEND_TIMEOUT_MS
#define SHM_SEND_RETRY_INTERVAL_MS
bool TimestampDifferenceExceeds(TimestampTz start_time, TimestampTz stop_time, int msec)
Definition: timestamp.c:1719
TimestampTz GetCurrentTimestamp(void)
Definition: timestamp.c:1583
#define unlikely(x)
Definition: c.h:300
int64 TimestampTz
Definition: timestamp.h:39
#define CHECK_FOR_INTERRUPTS()
Definition: miscadmin.h:121
int debug_logical_replication_streaming
@ DEBUG_LOGICAL_REP_STREAMING_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
bool IsTransactionState(void)
Definition: xact.c:378

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

1339 {
1340  stream_apply_worker = winfo;
1341 }

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

static bool pa_setup_dsm ( ParallelApplyWorkerInfo winfo)
static

Definition at line 327 of file applyparallelworker.c.

328 {
330  Size segsize;
331  dsm_segment *seg;
332  shm_toc *toc;
334  shm_mq *mq;
335  Size queue_size = DSM_QUEUE_SIZE;
336  Size error_queue_size = DSM_ERROR_QUEUE_SIZE;
337 
338  /*
339  * Estimate how much shared memory we need.
340  *
341  * Because the TOC machinery may choose to insert padding of oddly-sized
342  * requests, we must estimate each chunk separately.
343  *
344  * We need one key to register the location of the header, and two other
345  * keys to track the locations of the message queue and the error message
346  * queue.
347  */
350  shm_toc_estimate_chunk(&e, queue_size);
351  shm_toc_estimate_chunk(&e, error_queue_size);
352 
354  segsize = shm_toc_estimate(&e);
355 
356  /* Create the shared memory segment and establish a table of contents. */
357  seg = dsm_create(shm_toc_estimate(&e), 0);
358  if (!seg)
359  return false;
360 
362  segsize);
363 
364  /* Set up the header region. */
365  shared = shm_toc_allocate(toc, sizeof(ParallelApplyWorkerShared));
366  SpinLockInit(&shared->mutex);
367 
371  shared->fileset_state = FS_EMPTY;
372 
374 
375  /* Set up message queue for the worker. */
376  mq = shm_mq_create(shm_toc_allocate(toc, queue_size), queue_size);
379 
380  /* Attach the queue. */
381  winfo->mq_handle = shm_mq_attach(mq, seg, NULL);
382 
383  /* Set up error queue for the worker. */
384  mq = shm_mq_create(shm_toc_allocate(toc, error_queue_size),
385  error_queue_size);
388 
389  /* Attach the queue. */
390  winfo->error_mq_handle = shm_mq_attach(mq, seg, NULL);
391 
392  /* Return results to caller. */
393  winfo->dsm_seg = seg;
394  winfo->shared = shared;
395 
396  return true;
397 }
#define DSM_ERROR_QUEUE_SIZE
#define DSM_QUEUE_SIZE
#define PARALLEL_APPLY_KEY_SHARED
#define PARALLEL_APPLY_KEY_ERROR_QUEUE
#define PARALLEL_APPLY_KEY_MQ
#define PG_LOGICAL_APPLY_SHM_MAGIC
static void pg_atomic_init_u32(volatile pg_atomic_uint32 *ptr, uint32 val)
Definition: atomics.h:218
void * dsm_segment_address(dsm_segment *seg)
Definition: dsm.c:1066
dsm_segment * dsm_create(Size size, int flags)
Definition: dsm.c:489
e
Definition: preproc-init.c:82
shm_mq_handle * shm_mq_attach(shm_mq *mq, dsm_segment *seg, BackgroundWorkerHandle *handle)
Definition: shm_mq.c:291
void shm_mq_set_sender(shm_mq *mq, PGPROC *proc)
Definition: shm_mq.c:225
shm_mq * shm_mq_create(void *address, Size size)
Definition: shm_mq.c:178
void shm_mq_set_receiver(shm_mq *mq, PGPROC *proc)
Definition: shm_mq.c:207
shm_toc * shm_toc_create(uint64 magic, void *address, Size nbytes)
Definition: shm_toc.c:40
Size shm_toc_estimate(shm_toc_estimator *e)
Definition: shm_toc.c:263
void shm_toc_insert(shm_toc *toc, uint64 key, void *address)
Definition: shm_toc.c:171
void * shm_toc_allocate(shm_toc *toc, Size nbytes)
Definition: shm_toc.c:88
#define shm_toc_estimate_chunk(e, sz)
Definition: shm_toc.h:51
#define shm_toc_initialize_estimator(e)
Definition: shm_toc.h:49
#define shm_toc_estimate_keys(e, cnt)
Definition: shm_toc.h:53
#define SpinLockInit(lock)
Definition: spin.h:60
PGPROC * MyProc
Definition: proc.c:66
Definition: shm_mq.c:73

References dsm_create(), DSM_ERROR_QUEUE_SIZE, DSM_QUEUE_SIZE, ParallelApplyWorkerInfo::dsm_seg, dsm_segment_address(), ParallelApplyWorkerInfo::error_mq_handle, ParallelApplyWorkerShared::fileset_state, FS_EMPTY, InvalidXLogRecPtr, ParallelApplyWorkerShared::last_commit_end, ParallelApplyWorkerInfo::mq_handle, ParallelApplyWorkerShared::mutex, MyProc, PARALLEL_APPLY_KEY_ERROR_QUEUE, PARALLEL_APPLY_KEY_MQ, PARALLEL_APPLY_KEY_SHARED, PARALLEL_TRANS_UNKNOWN, ParallelApplyWorkerShared::pending_stream_count, pg_atomic_init_u32(), PG_LOGICAL_APPLY_SHM_MAGIC, ParallelApplyWorkerInfo::shared, shm_mq_attach(), shm_mq_create(), shm_mq_set_receiver(), shm_mq_set_sender(), shm_toc_allocate(), shm_toc_create(), shm_toc_estimate(), shm_toc_estimate_chunk, shm_toc_estimate_keys, shm_toc_initialize_estimator, shm_toc_insert(), SpinLockInit, and ParallelApplyWorkerShared::xact_state.

Referenced by pa_launch_parallel_worker().

◆ pa_shutdown()

static void pa_shutdown ( int  code,
Datum  arg 
)
static

Definition at line 848 of file applyparallelworker.c.

849 {
853 
855 }
#define InvalidBackendId
Definition: backendid.h:23
void * arg
static Pointer DatumGetPointer(Datum X)
Definition: postgres.h:312
int SendProcSignal(pid_t pid, ProcSignalReason reason, BackendId backendId)
Definition: procsignal.c:262
@ PROCSIG_PARALLEL_APPLY_MESSAGE
Definition: procsignal.h:38

References arg, DatumGetPointer(), dsm_detach(), InvalidBackendId, LogicalRepWorker::leader_pid, MyLogicalRepWorker, PROCSIG_PARALLEL_APPLY_MESSAGE, and SendProcSignal().

Referenced by ParallelApplyWorkerMain().

◆ pa_start_subtrans()

void pa_start_subtrans ( TransactionId  current_xid,
TransactionId  top_xid 
)

Definition at line 1366 of file applyparallelworker.c.

1367 {
1368  if (current_xid != top_xid &&
1369  !list_member_xid(subxactlist, current_xid))
1370  {
1371  MemoryContext oldctx;
1372  char spname[NAMEDATALEN];
1373 
1374  pa_savepoint_name(MySubscription->oid, current_xid,
1375  spname, sizeof(spname));
1376 
1377  elog(DEBUG1, "defining savepoint %s in logical replication parallel apply worker", spname);
1378 
1379  /* We must be in transaction block to define the SAVEPOINT. */
1380  if (!IsTransactionBlock())
1381  {
1382  if (!IsTransactionState())
1384 
1387  }
1388 
1389  DefineSavepoint(spname);
1390 
1391  /*
1392  * CommitTransactionCommand is needed to start a subtransaction after
1393  * issuing a SAVEPOINT inside a transaction block (see
1394  * StartSubTransaction()).
1395  */
1397 
1399  subxactlist = lappend_xid(subxactlist, current_xid);
1400  MemoryContextSwitchTo(oldctx);
1401  }
1402 }
static void pa_savepoint_name(Oid suboid, TransactionId xid, char *spname, Size szsp)
#define DEBUG1
Definition: elog.h:30
List * lappend_xid(List *list, TransactionId datum)
Definition: list.c:392
bool list_member_xid(const List *list, TransactionId datum)
Definition: list.c:741
MemoryContext TopTransactionContext
Definition: mcxt.c:146
#define NAMEDATALEN
void DefineSavepoint(const char *name)
Definition: xact.c:4219
void StartTransactionCommand(void)
Definition: xact.c:2937
bool IsTransactionBlock(void)
Definition: xact.c:4816
void BeginTransactionBlock(void)
Definition: xact.c:3770
void CommitTransactionCommand(void)
Definition: xact.c:3034

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

1421 {
1422  TransactionId xid = abort_data->xid;
1423  TransactionId subxid = abort_data->subxid;
1424 
1425  /*
1426  * Update origin state so we can restart streaming from correct position
1427  * in case of crash.
1428  */
1431 
1432  /*
1433  * If the two XIDs are the same, it's in fact abort of toplevel xact, so
1434  * just free the subxactlist.
1435  */
1436  if (subxid == xid)
1437  {
1439 
1440  /*
1441  * Release the lock as we might be processing an empty streaming
1442  * transaction in which case the lock won't be released during
1443  * transaction rollback.
1444  *
1445  * Note that it's ok to release the transaction lock before aborting
1446  * the transaction because even if the parallel apply worker dies due
1447  * to crash or some other reason, such a transaction would still be
1448  * considered aborted.
1449  */
1451 
1453 
1454  if (IsTransactionBlock())
1455  {
1456  EndTransactionBlock(false);
1458  }
1459 
1461 
1463  }
1464  else
1465  {
1466  /* OK, so it's a subxact. Rollback to the savepoint. */
1467  int i;
1468  char spname[NAMEDATALEN];
1469 
1470  pa_savepoint_name(MySubscription->oid, subxid, spname, sizeof(spname));
1471 
1472  elog(DEBUG1, "rolling back to savepoint %s in logical replication parallel apply worker", spname);
1473 
1474  /*
1475  * Search the subxactlist, determine the offset tracked for the
1476  * subxact, and truncate the list.
1477  *
1478  * Note that for an empty sub-transaction we won't find the subxid
1479  * here.
1480  */
1481  for (i = list_length(subxactlist) - 1; i >= 0; i--)
1482  {
1484 
1485  if (xid_tmp == subxid)
1486  {
1487  RollbackToSavepoint(spname);
1490  break;
1491  }
1492  }
1493  }
1494 }
void pa_set_xact_state(ParallelApplyWorkerShared *wshared, ParallelTransState xact_state)
void pa_reset_subtrans(void)
void pa_unlock_transaction(TransactionId xid, LOCKMODE lockmode)
void pgstat_report_activity(BackendState state, const char *cmd_str)
@ STATE_IDLE
int i
Definition: isn.c:73
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 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:4413
bool EndTransactionBlock(bool chain)
Definition: xact.c:3890
void AbortCurrentTransaction(void)
Definition: xact.c:3305

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

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

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

static void pa_wait_for_xact_finish ( ParallelApplyWorkerInfo winfo)
static

Definition at line 1278 of file applyparallelworker.c.

1279 {
1280  /*
1281  * Wait until the parallel apply worker set the state to
1282  * PARALLEL_TRANS_STARTED which means it has acquired the transaction
1283  * lock. This is to prevent leader apply worker from acquiring the
1284  * transaction lock earlier than the parallel apply worker.
1285  */
1287 
1288  /*
1289  * Wait for the transaction lock to be released. This is required to
1290  * detect deadlock among leader and parallel apply workers. Refer to the
1291  * comments atop this file.
1292  */
1295 
1296  /*
1297  * Check if the state becomes PARALLEL_TRANS_FINISHED in case the parallel
1298  * apply worker failed while applying changes causing the lock to be
1299  * released.
1300  */
1302  ereport(ERROR,
1303  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1304  errmsg("lost connection to the logical replication parallel apply worker")));
1305 }
void pa_lock_transaction(TransactionId xid, LOCKMODE lockmode)
static void pa_wait_for_xact_state(ParallelApplyWorkerInfo *winfo, ParallelTransState xact_state)
@ PARALLEL_TRANS_STARTED

References AccessShareLock, ereport, errcode(), errmsg(), ERROR, pa_get_xact_state(), pa_lock_transaction(), pa_unlock_transaction(), pa_wait_for_xact_state(), PARALLEL_TRANS_FINISHED, PARALLEL_TRANS_STARTED, ParallelApplyWorkerInfo::shared, and ParallelApplyWorkerShared::xid.

Referenced by pa_xact_finish().

◆ pa_wait_for_xact_state()

static void pa_wait_for_xact_state ( ParallelApplyWorkerInfo winfo,
ParallelTransState  xact_state 
)
static

Definition at line 1248 of file applyparallelworker.c.

1250 {
1251  for (;;)
1252  {
1253  /*
1254  * Stop if the transaction state has reached or exceeded the given
1255  * xact_state.
1256  */
1257  if (pa_get_xact_state(winfo->shared) >= xact_state)
1258  break;
1259 
1260  /* Wait to be signalled. */
1261  (void) WaitLatch(MyLatch,
1263  10L,
1264  WAIT_EVENT_LOGICAL_PARALLEL_APPLY_STATE_CHANGE);
1265 
1266  /* Reset the latch so we don't spin. */
1268 
1269  /* An interrupt may have occurred while we were waiting. */
1271  }
1272 }

References CHECK_FOR_INTERRUPTS, MyLatch, pa_get_xact_state(), ResetLatch(), ParallelApplyWorkerInfo::shared, WaitLatch(), WL_EXIT_ON_PM_DEATH, WL_LATCH_SET, and WL_TIMEOUT.

Referenced by pa_wait_for_xact_finish().

◆ pa_xact_finish()

void pa_xact_finish ( ParallelApplyWorkerInfo winfo,
XLogRecPtr  remote_lsn 
)

Definition at line 1622 of file applyparallelworker.c.

1623 {
1625 
1626  /*
1627  * Unlock the shared object lock so that parallel apply worker can
1628  * continue to receive and apply changes.
1629  */
1631 
1632  /*
1633  * Wait for that worker to finish. This is necessary to maintain commit
1634  * order which avoids failures due to transaction dependencies and
1635  * deadlocks.
1636  */
1637  pa_wait_for_xact_finish(winfo);
1638 
1639  if (!XLogRecPtrIsInvalid(remote_lsn))
1640  store_flush_position(remote_lsn, winfo->shared->last_commit_end);
1641 
1642  pa_free_worker(winfo);
1643 }
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:3456

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

◆ ParallelApplyWorkerMain()

void ParallelApplyWorkerMain ( Datum  main_arg)

Definition at line 861 of file applyparallelworker.c.

862 {
864  dsm_handle handle;
865  dsm_segment *seg;
866  shm_toc *toc;
867  shm_mq *mq;
868  shm_mq_handle *mqh;
869  shm_mq_handle *error_mqh;
870  RepOriginId originid;
871  int worker_slot = DatumGetInt32(main_arg);
872  char originname[NAMEDATALEN];
873 
875 
876  /* Setup signal handling. */
879  pqsignal(SIGTERM, die);
881 
882  /*
883  * Attach to the dynamic shared memory segment for the parallel apply, and
884  * find its table of contents.
885  *
886  * Like parallel query, we don't need resource owner by this time. See
887  * ParallelWorkerMain.
888  */
889  memcpy(&handle, MyBgworkerEntry->bgw_extra, sizeof(dsm_handle));
890  seg = dsm_attach(handle);
891  if (!seg)
892  ereport(ERROR,
893  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
894  errmsg("could not map dynamic shared memory segment")));
895 
897  if (!toc)
898  ereport(ERROR,
899  (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
900  errmsg("invalid magic number in dynamic shared memory segment")));
901 
902  /* Look up the shared information. */
903  shared = shm_toc_lookup(toc, PARALLEL_APPLY_KEY_SHARED, false);
904  MyParallelShared = shared;
905 
906  /*
907  * Attach to the message queue.
908  */
909  mq = shm_toc_lookup(toc, PARALLEL_APPLY_KEY_MQ, false);
911  mqh = shm_mq_attach(mq, seg, NULL);
912 
913  /*
914  * Primary initialization is complete. Now, we can attach to our slot.
915  * This is to ensure that the leader apply worker does not write data to
916  * the uninitialized memory queue.
917  */
918  logicalrep_worker_attach(worker_slot);
919 
920  /*
921  * Register the shutdown callback after we are attached to the worker
922  * slot. This is to ensure that MyLogicalRepWorker remains valid when this
923  * callback is invoked.
924  */
926 
931 
932  /*
933  * Attach to the error queue.
934  */
937  error_mqh = shm_mq_attach(mq, seg, NULL);
938 
939  pq_redirect_to_shm_mq(seg, error_mqh);
942 
945 
947 
948  InitializingApplyWorker = false;
949 
950  /* Setup replication origin tracking. */
953  originname, sizeof(originname));
954  originid = replorigin_by_name(originname, false);
955 
956  /*
957  * The parallel apply worker doesn't need to monopolize this replication
958  * origin which was already acquired by its leader process.
959  */
961  replorigin_session_origin = originid;
963 
964  /*
965  * Setup callback for syscache so that we know when something changes in
966  * the subscription relation state.
967  */
970  (Datum) 0);
971 
972  set_apply_error_context_origin(originname);
973 
975 
976  /*
977  * The parallel apply worker must not get here because the parallel apply
978  * worker will only stop when it receives a SIGTERM or SIGINT from the
979  * leader, or when there is an error. None of these cases will allow the
980  * code to reach here.
981  */
982  Assert(false);
983 }
static void pa_shutdown(int code, Datum arg)
static void LogicalParallelApplyLoop(shm_mq_handle *mqh)
bool InitializingApplyWorker
Definition: worker.c:336
void ReplicationOriginNameForLogicalRep(Oid suboid, Oid relid, char *originname, Size szoriginname)
Definition: worker.c:446
void set_apply_error_context_origin(char *originname)
Definition: worker.c:5035
void InitializeLogRepWorker(void)
Definition: worker.c:4568
dsm_segment * dsm_attach(dsm_handle h)
Definition: dsm.c:638
uint32 dsm_handle
Definition: dsm_impl.h:55
void SignalHandlerForShutdownRequest(SIGNAL_ARGS)
Definition: interrupt.c:109
void SignalHandlerForConfigReload(SIGNAL_ARGS)
Definition: interrupt.c:61
void CacheRegisterSyscacheCallback(int cacheid, SyscacheCallbackFunction func, Datum arg)
Definition: inval.c:1519
void before_shmem_exit(pg_on_exit_callback function, Datum arg)
Definition: ipc.c:333
void logicalrep_worker_attach(int slot)
Definition: launcher.c:713
RepOriginId replorigin_by_name(const char *roname, bool missing_ok)
Definition: origin.c:221
RepOriginId replorigin_session_origin
Definition: origin.c:156
void replorigin_session_setup(RepOriginId node, int acquired_by)
Definition: origin.c:1095
#define die(msg)
Definition: pg_test_fsync.c:95
pqsigfunc pqsignal(int signo, pqsigfunc func)
static Datum PointerGetDatum(const void *X)
Definition: postgres.h:322
uintptr_t Datum
Definition: postgres.h:64
static int32 DatumGetInt32(Datum X)
Definition: postgres.h:202
void BackgroundWorkerUnblockSignals(void)
Definition: postmaster.c:5638
BackgroundWorker * MyBgworkerEntry
Definition: postmaster.c:194
void pq_set_parallel_leader(pid_t pid, BackendId backend_id)
Definition: pqmq.c:78
void pq_redirect_to_shm_mq(dsm_segment *seg, shm_mq_handle *mqh)
Definition: pqmq.c:53
shm_toc * shm_toc_attach(uint64 magic, void *address)
Definition: shm_toc.c:64
void * shm_toc_lookup(shm_toc *toc, uint64 key, bool noError)
Definition: shm_toc.c:232
char bgw_extra[BGW_EXTRALEN]
Definition: bgworker.h:99
TimestampTz last_recv_time
TimestampTz reply_time
TimestampTz last_send_time
@ SUBSCRIPTIONRELMAP
Definition: syscache.h:100
void invalidate_syncing_table_states(Datum arg, int cacheid, uint32 hashvalue)
Definition: tablesync.c:274
#define SIGHUP
Definition: win32_port.h:168
uint16 RepOriginId
Definition: xlogdefs.h:65

References Assert(), BackgroundWorkerUnblockSignals(), before_shmem_exit(), BackgroundWorker::bgw_extra, CacheRegisterSyscacheCallback(), CommitTransactionCommand(), DatumGetInt32(), die, dsm_attach(), dsm_segment_address(), ereport, errcode(), errmsg(), ERROR, LogicalRepWorker::generation, InitializeLogRepWorker(), InitializingApplyWorker, invalidate_syncing_table_states(), InvalidBackendId, InvalidOid, LogicalRepWorker::last_recv_time, LogicalRepWorker::last_send_time, LogicalRepWorker::leader_pid, LogicalParallelApplyLoop(), logicalrep_worker_attach(), ParallelApplyWorkerShared::logicalrep_worker_generation, ParallelApplyWorkerShared::logicalrep_worker_slot_no, ParallelApplyWorkerShared::mutex, MyBgworkerEntry, MyLogicalRepWorker, MyParallelShared, MyProc, MySubscription, NAMEDATALEN, Subscription::oid, pa_shutdown(), PARALLEL_APPLY_KEY_ERROR_QUEUE, PARALLEL_APPLY_KEY_MQ, PARALLEL_APPLY_KEY_SHARED, PG_LOGICAL_APPLY_SHM_MAGIC, PointerGetDatum(), pq_redirect_to_shm_mq(), pq_set_parallel_leader(), pqsignal(), ReplicationOriginNameForLogicalRep(), replorigin_by_name(), replorigin_session_origin, replorigin_session_setup(), LogicalRepWorker::reply_time, set_apply_error_context_origin(), shm_mq_attach(), shm_mq_set_receiver(), shm_mq_set_sender(), shm_toc_attach(), shm_toc_lookup(), SIGHUP, SignalHandlerForConfigReload(), SignalHandlerForShutdownRequest(), SpinLockAcquire, SpinLockRelease, StartTransactionCommand(), and SUBSCRIPTIONRELMAP.

◆ ProcessParallelApplyInterrupts()

static void ProcessParallelApplyInterrupts ( void  )
static

Definition at line 713 of file applyparallelworker.c.

714 {
716 
718  {
719  ereport(LOG,
720  (errmsg("logical replication parallel apply worker for subscription \"%s\" has finished",
721  MySubscription->name)));
722 
723  proc_exit(0);
724  }
725 
727  {
728  ConfigReloadPending = false;
730  }
731 }
@ PGC_SIGHUP
Definition: guc.h:71
void ProcessConfigFile(GucContext context)
volatile sig_atomic_t ShutdownRequestPending
Definition: interrupt.c:28
volatile sig_atomic_t ConfigReloadPending
Definition: interrupt.c:27
void proc_exit(int code)
Definition: ipc.c:104

References CHECK_FOR_INTERRUPTS, ConfigReloadPending, ereport, errmsg(), LOG, MySubscription, Subscription::name, PGC_SIGHUP, proc_exit(), ProcessConfigFile(), and ShutdownRequestPending.

Referenced by LogicalParallelApplyLoop().

Variable Documentation

◆ MyParallelShared

◆ ParallelApplyMessagePending

volatile sig_atomic_t ParallelApplyMessagePending = false

◆ ParallelApplyTxnHash

HTAB* ParallelApplyTxnHash = NULL
static

Definition at line 225 of file applyparallelworker.c.

Referenced by pa_allocate_worker(), pa_find_worker(), and pa_free_worker().

◆ ParallelApplyWorkerPool

◆ stream_apply_worker

ParallelApplyWorkerInfo* stream_apply_worker = NULL
static

Definition at line 252 of file applyparallelworker.c.

Referenced by pa_find_worker(), and pa_set_stream_apply_worker().

◆ subxactlist

List* subxactlist = NIL
static

Definition at line 255 of file applyparallelworker.c.

Referenced by pa_reset_subtrans(), pa_start_subtrans(), and pa_stream_abort().