#include <signal.h>

Include dependency graph for logicalworker.h:

This graph shows which files directly or indirectly include this file:

Functions
void	ApplyWorkerMain (Datum main_arg)

void	ParallelApplyWorkerMain (Datum main_arg)

void	TablesyncWorkerMain (Datum main_arg)

bool	IsLogicalWorker (void)

bool	IsLogicalParallelApplyWorker (void)

void	HandleParallelApplyMessageInterrupt (void)

void	HandleParallelApplyMessages (void)

void	LogicalRepWorkersWakeupAtCommit (Oid subid)

void	AtEOXact_LogicalRepWorkers (bool isCommit)

Variables
PGDLLIMPORT volatile sig_atomic_t	ParallelApplyMessagePending

Function Documentation

◆ ApplyWorkerMain()

void ApplyWorkerMain ( Datum main_arg )

Definition at line 4703 of file worker.c.

 {
     int         worker_slot = DatumGetInt32(main_arg);
  
     InitializingApplyWorker = true;
  
     SetupApplyOrSyncWorker(worker_slot);
  
     InitializingApplyWorker = false;
  
     run_apply_worker();
  
     proc_exit(0);
 }

References DatumGetInt32(), InitializingApplyWorker, proc_exit(), run_apply_worker(), and SetupApplyOrSyncWorker().

◆ AtEOXact_LogicalRepWorkers()

void AtEOXact_LogicalRepWorkers ( bool isCommit )

Definition at line 5021 of file worker.c.

 {
     if (isCommit && on_commit_wakeup_workers_subids != NIL)
     {
         ListCell   *lc;
  
         LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
         foreach(lc, on_commit_wakeup_workers_subids)
         {
             Oid         subid = lfirst_oid(lc);
             List       *workers;
             ListCell   *lc2;
  
             workers = logicalrep_workers_find(subid, true);
             foreach(lc2, workers)
             {
                 LogicalRepWorker *worker = (LogicalRepWorker *) lfirst(lc2);
  
                 logicalrep_worker_wakeup_ptr(worker);
             }
         }
         LWLockRelease(LogicalRepWorkerLock);
     }
  
     /* The List storage will be reclaimed automatically in xact cleanup. */
     on_commit_wakeup_workers_subids = NIL;
 }

References lfirst, lfirst_oid, logicalrep_worker_wakeup_ptr(), logicalrep_workers_find(), LW_SHARED, LWLockAcquire(), LWLockRelease(), NIL, and on_commit_wakeup_workers_subids.

Referenced by AbortTransaction(), CommitTransaction(), and PrepareTransaction().

◆ HandleParallelApplyMessageInterrupt()

void HandleParallelApplyMessageInterrupt ( void )

Definition at line 990 of file applyparallelworker.c.

 {
     InterruptPending = true;
     ParallelApplyMessagePending = true;
     SetLatch(MyLatch);
 }

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

Referenced by procsignal_sigusr1_handler().

◆ HandleParallelApplyMessages()

void HandleParallelApplyMessages ( void )

Definition at line 1064 of file applyparallelworker.c.

 {
     ListCell   *lc;
     MemoryContext oldcontext;
  
     static MemoryContext hpam_context = NULL;
  
     /*
      * This is invoked from ProcessInterrupts(), and since some of the
      * functions it calls contain CHECK_FOR_INTERRUPTS(), there is a potential
      * for recursive calls if more signals are received while this runs. It's
      * unclear that recursive entry would be safe, and it doesn't seem useful
      * even if it is safe, so let's block interrupts until done.
      */
     HOLD_INTERRUPTS();
  
     /*
      * Moreover, CurrentMemoryContext might be pointing almost anywhere. We
      * don't want to risk leaking data into long-lived contexts, so let's do
      * our work here in a private context that we can reset on each use.
      */
     if (!hpam_context)          /* first time through? */
         hpam_context = AllocSetContextCreate(TopMemoryContext,
                                              "HandleParallelApplyMessages",
                                              ALLOCSET_DEFAULT_SIZES);
     else
         MemoryContextReset(hpam_context);
  
     oldcontext = MemoryContextSwitchTo(hpam_context);
  
     ParallelApplyMessagePending = false;
  
     foreach(lc, ParallelApplyWorkerPool)
     {
         shm_mq_result res;
         Size        nbytes;
         void       *data;
         ParallelApplyWorkerInfo *winfo = (ParallelApplyWorkerInfo *) lfirst(lc);
  
         /*
          * The leader will detach from the error queue and set it to NULL
          * before preparing to stop all parallel apply workers, so we don't
          * need to handle error messages anymore. See
          * logicalrep_worker_detach.
          */
         if (!winfo->error_mq_handle)
             continue;
  
         res = shm_mq_receive(winfo->error_mq_handle, &nbytes, &data, true);
  
         if (res == SHM_MQ_WOULD_BLOCK)
             continue;
         else if (res == SHM_MQ_SUCCESS)
         {
             StringInfoData msg;
  
             initStringInfo(&msg);
             appendBinaryStringInfo(&msg, data, nbytes);
             HandleParallelApplyMessage(&msg);
             pfree(msg.data);
         }
         else
             ereport(ERROR,
                     (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                      errmsg("lost connection to the logical replication parallel apply worker")));
     }
  
     MemoryContextSwitchTo(oldcontext);
  
     /* Might as well clear the context on our way out */
     MemoryContextReset(hpam_context);
  
     RESUME_INTERRUPTS();
 }

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

◆ IsLogicalParallelApplyWorker()

bool IsLogicalParallelApplyWorker ( void )

Definition at line 4771 of file worker.c.

 {
     return IsLogicalWorker() && am_parallel_apply_worker();
 }

References am_parallel_apply_worker(), and IsLogicalWorker().

Referenced by mq_putmessage().

◆ IsLogicalWorker()

bool IsLogicalWorker ( void )

Definition at line 4762 of file worker.c.

 {
     return MyLogicalRepWorker != NULL;
 }

References MyLogicalRepWorker.

Referenced by IsLogicalParallelApplyWorker(), and ProcessInterrupts().

◆ LogicalRepWorkersWakeupAtCommit()

void LogicalRepWorkersWakeupAtCommit ( Oid subid )

Definition at line 5007 of file worker.c.

 {
     MemoryContext oldcxt;
  
     oldcxt = MemoryContextSwitchTo(TopTransactionContext);
     on_commit_wakeup_workers_subids =
         list_append_unique_oid(on_commit_wakeup_workers_subids, subid);
     MemoryContextSwitchTo(oldcxt);
 }

References list_append_unique_oid(), MemoryContextSwitchTo(), on_commit_wakeup_workers_subids, and TopTransactionContext.

Referenced by AlterObjectRename_internal(), AlterSubscription(), and AlterSubscriptionOwner_internal().

◆ ParallelApplyWorkerMain()

void ParallelApplyWorkerMain ( Datum main_arg )

Definition at line 858 of file applyparallelworker.c.

 {
     ParallelApplyWorkerShared *shared;
     dsm_handle  handle;
     dsm_segment *seg;
     shm_toc    *toc;
     shm_mq     *mq;
     shm_mq_handle *mqh;
     shm_mq_handle *error_mqh;
     RepOriginId originid;
     int         worker_slot = DatumGetInt32(main_arg);
     char        originname[NAMEDATALEN];
  
     InitializingApplyWorker = true;
  
     /* Setup signal handling. */
     pqsignal(SIGHUP, SignalHandlerForConfigReload);
     pqsignal(SIGINT, SignalHandlerForShutdownRequest);
     pqsignal(SIGTERM, die);
     BackgroundWorkerUnblockSignals();
  
     /*
      * Attach to the dynamic shared memory segment for the parallel apply, and
      * find its table of contents.
      *
      * Like parallel query, we don't need resource owner by this time. See
      * ParallelWorkerMain.
      */
     memcpy(&handle, MyBgworkerEntry->bgw_extra, sizeof(dsm_handle));
     seg = dsm_attach(handle);
     if (!seg)
         ereport(ERROR,
                 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                  errmsg("could not map dynamic shared memory segment")));
  
     toc = shm_toc_attach(PG_LOGICAL_APPLY_SHM_MAGIC, dsm_segment_address(seg));
     if (!toc)
         ereport(ERROR,
                 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                  errmsg("invalid magic number in dynamic shared memory segment")));
  
     /* Look up the shared information. */
     shared = shm_toc_lookup(toc, PARALLEL_APPLY_KEY_SHARED, false);
     MyParallelShared = shared;
  
     /*
      * Attach to the message queue.
      */
     mq = shm_toc_lookup(toc, PARALLEL_APPLY_KEY_MQ, false);
     shm_mq_set_receiver(mq, MyProc);
     mqh = shm_mq_attach(mq, seg, NULL);
  
     /*
      * Primary initialization is complete. Now, we can attach to our slot.
      * This is to ensure that the leader apply worker does not write data to
      * the uninitialized memory queue.
      */
     logicalrep_worker_attach(worker_slot);
  
     /*
      * Register the shutdown callback after we are attached to the worker
      * slot. This is to ensure that MyLogicalRepWorker remains valid when this
      * callback is invoked.
      */
     before_shmem_exit(pa_shutdown, PointerGetDatum(seg));
  
     SpinLockAcquire(&MyParallelShared->mutex);
     MyParallelShared->logicalrep_worker_generation = MyLogicalRepWorker->generation;
     MyParallelShared->logicalrep_worker_slot_no = worker_slot;
     SpinLockRelease(&MyParallelShared->mutex);
  
     /*
      * Attach to the error queue.
      */
     mq = shm_toc_lookup(toc, PARALLEL_APPLY_KEY_ERROR_QUEUE, false);
     shm_mq_set_sender(mq, MyProc);
     error_mqh = shm_mq_attach(mq, seg, NULL);
  
     pq_redirect_to_shm_mq(seg, error_mqh);
     pq_set_parallel_leader(MyLogicalRepWorker->leader_pid,
                            InvalidBackendId);
  
     MyLogicalRepWorker->last_send_time = MyLogicalRepWorker->last_recv_time =
         MyLogicalRepWorker->reply_time = 0;
  
     InitializeLogRepWorker();
  
     InitializingApplyWorker = false;
  
     /* Setup replication origin tracking. */
     StartTransactionCommand();
     ReplicationOriginNameForLogicalRep(MySubscription->oid, InvalidOid,
                                        originname, sizeof(originname));
     originid = replorigin_by_name(originname, false);
  
     /*
      * The parallel apply worker doesn't need to monopolize this replication
      * origin which was already acquired by its leader process.
      */
     replorigin_session_setup(originid, MyLogicalRepWorker->leader_pid);
     replorigin_session_origin = originid;
     CommitTransactionCommand();
  
     /*
      * Setup callback for syscache so that we know when something changes in
      * the subscription relation state.
      */
     CacheRegisterSyscacheCallback(SUBSCRIPTIONRELMAP,
                                   invalidate_syncing_table_states,
                                   (Datum) 0);
  
     set_apply_error_context_origin(originname);
  
     LogicalParallelApplyLoop(mqh);
  
     /*
      * The parallel apply worker must not get here because the parallel apply
      * worker will only stop when it receives a SIGTERM or SIGINT from the
      * leader, or when there is an error. None of these cases will allow the
      * code to reach here.
      */
     Assert(false);
 }

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.

◆ TablesyncWorkerMain()

void TablesyncWorkerMain ( Datum main_arg )

Definition at line 1683 of file tablesync.c.

 {
     int         worker_slot = DatumGetInt32(main_arg);
  
     SetupApplyOrSyncWorker(worker_slot);
  
     run_tablesync_worker();
  
     finish_sync_worker();
 }

References DatumGetInt32(), run_tablesync_worker(), and SetupApplyOrSyncWorker().

Variable Documentation

◆ ParallelApplyMessagePending

PGDLLIMPORT volatile sig_atomic_t ParallelApplyMessagePending

extern

Definition at line 245 of file applyparallelworker.c.

Referenced by HandleParallelApplyMessageInterrupt(), HandleParallelApplyMessages(), and ProcessInterrupts().

Functions

Variables