PostgreSQL Source Code git master
Functions | Variables
logicalworker.h File Reference
#include <signal.h>
Include dependency graph for logicalworker.h:
This graph shows which files directly or indirectly include this file:

Go to the source code of 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  ProcessParallelApplyMessages (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 5885 of file worker.c.

5886{
5887 int worker_slot = DatumGetInt32(main_arg);
5888
5890
5891 SetupApplyOrSyncWorker(worker_slot);
5892
5894
5896
5897 proc_exit(0);
5898}
bool InitializingApplyWorker
Definition: worker.c:499
static void run_apply_worker()
Definition: worker.c:5592
void SetupApplyOrSyncWorker(int worker_slot)
Definition: worker.c:5831
void proc_exit(int code)
Definition: ipc.c:104
static int32 DatumGetInt32(Datum X)
Definition: postgres.h:212

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

◆ AtEOXact_LogicalRepWorkers()

void AtEOXact_LogicalRepWorkers ( bool  isCommit )

Definition at line 6228 of file worker.c.

6229{
6230 if (isCommit && on_commit_wakeup_workers_subids != NIL)
6231 {
6232 ListCell *lc;
6233
6234 LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
6236 {
6237 Oid subid = lfirst_oid(lc);
6238 List *workers;
6239 ListCell *lc2;
6240
6241 workers = logicalrep_workers_find(subid, true, false);
6242 foreach(lc2, workers)
6243 {
6244 LogicalRepWorker *worker = (LogicalRepWorker *) lfirst(lc2);
6245
6247 }
6248 }
6249 LWLockRelease(LogicalRepWorkerLock);
6250 }
6251
6252 /* The List storage will be reclaimed automatically in xact cleanup. */
6254}
static List * on_commit_wakeup_workers_subids
Definition: worker.c:482
List * logicalrep_workers_find(Oid subid, bool only_running, bool acquire_lock)
Definition: launcher.c:286
void logicalrep_worker_wakeup_ptr(LogicalRepWorker *worker)
Definition: launcher.c:720
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
Definition: lwlock.c:1174
void LWLockRelease(LWLock *lock)
Definition: lwlock.c:1894
@ LW_SHARED
Definition: lwlock.h:113
#define lfirst(lc)
Definition: pg_list.h:172
#define NIL
Definition: pg_list.h:68
#define lfirst_oid(lc)
Definition: pg_list.h:174
unsigned int Oid
Definition: postgres_ext.h:32
Definition: pg_list.h:54
Definition: pg_list.h:46

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

998{
999 InterruptPending = true;
1002}
volatile sig_atomic_t ParallelApplyMessagePending
volatile sig_atomic_t InterruptPending
Definition: globals.c:32
struct Latch * MyLatch
Definition: globals.c:63
void SetLatch(Latch *latch)
Definition: latch.c:290

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

Referenced by procsignal_sigusr1_handler().

◆ IsLogicalParallelApplyWorker()

bool IsLogicalParallelApplyWorker ( void  )

Definition at line 5970 of file worker.c.

5971{
5973}
bool IsLogicalWorker(void)
Definition: worker.c:5961
static bool am_parallel_apply_worker(void)

References am_parallel_apply_worker(), and IsLogicalWorker().

Referenced by mq_putmessage().

◆ IsLogicalWorker()

bool IsLogicalWorker ( void  )

Definition at line 5961 of file worker.c.

5962{
5963 return MyLogicalRepWorker != NULL;
5964}
LogicalRepWorker * MyLogicalRepWorker
Definition: launcher.c:56

References MyLogicalRepWorker.

Referenced by IsLogicalParallelApplyWorker(), and ProcessInterrupts().

◆ LogicalRepWorkersWakeupAtCommit()

void LogicalRepWorkersWakeupAtCommit ( Oid  subid )

Definition at line 6214 of file worker.c.

6215{
6216 MemoryContext oldcxt;
6217
6221 MemoryContextSwitchTo(oldcxt);
6222}
List * list_append_unique_oid(List *list, Oid datum)
Definition: list.c:1380
MemoryContext TopTransactionContext
Definition: mcxt.c:171
static MemoryContext MemoryContextSwitchTo(MemoryContext context)
Definition: palloc.h:124

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.

859{
861 dsm_handle handle;
862 dsm_segment *seg;
863 shm_toc *toc;
864 shm_mq *mq;
865 shm_mq_handle *mqh;
866 shm_mq_handle *error_mqh;
867 RepOriginId originid;
868 int worker_slot = DatumGetInt32(main_arg);
869 char originname[NAMEDATALEN];
870
872
873 /*
874 * Setup signal handling.
875 *
876 * Note: We intentionally used SIGUSR2 to trigger a graceful shutdown
877 * initiated by the leader apply worker. This helps to differentiate it
878 * from the case where we abort the current transaction and exit on
879 * receiving SIGTERM.
880 */
882 pqsignal(SIGTERM, die);
885
886 /*
887 * Attach to the dynamic shared memory segment for the parallel apply, and
888 * find its table of contents.
889 *
890 * Like parallel query, we don't need resource owner by this time. See
891 * ParallelWorkerMain.
892 */
893 memcpy(&handle, MyBgworkerEntry->bgw_extra, sizeof(dsm_handle));
894 seg = dsm_attach(handle);
895 if (!seg)
897 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
898 errmsg("could not map dynamic shared memory segment")));
899
901 if (!toc)
903 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
904 errmsg("invalid magic number in dynamic shared memory segment")));
905
906 /* Look up the shared information. */
907 shared = shm_toc_lookup(toc, PARALLEL_APPLY_KEY_SHARED, false);
908 MyParallelShared = shared;
909
910 /*
911 * Attach to the message queue.
912 */
913 mq = shm_toc_lookup(toc, PARALLEL_APPLY_KEY_MQ, false);
915 mqh = shm_mq_attach(mq, seg, NULL);
916
917 /*
918 * Primary initialization is complete. Now, we can attach to our slot.
919 * This is to ensure that the leader apply worker does not write data to
920 * the uninitialized memory queue.
921 */
922 logicalrep_worker_attach(worker_slot);
923
924 /*
925 * Register the shutdown callback after we are attached to the worker
926 * slot. This is to ensure that MyLogicalRepWorker remains valid when this
927 * callback is invoked.
928 */
930
935
936 /*
937 * Attach to the error queue.
938 */
941 error_mqh = shm_mq_attach(mq, seg, NULL);
942
943 pq_redirect_to_shm_mq(seg, error_mqh);
946
949
951
953
954 /* Setup replication origin tracking. */
957 originname, sizeof(originname));
958 originid = replorigin_by_name(originname, false);
959
960 /*
961 * The parallel apply worker doesn't need to monopolize this replication
962 * origin which was already acquired by its leader process.
963 */
965 replorigin_session_origin = originid;
967
968 /*
969 * Setup callback for syscache so that we know when something changes in
970 * the subscription relation state.
971 */
972 CacheRegisterSyscacheCallback(SUBSCRIPTIONRELMAP,
974 (Datum) 0);
975
977
979
980 /*
981 * The parallel apply worker must not get here because the parallel apply
982 * worker will only stop when it receives a SIGTERM or SIGUSR2 from the
983 * leader, or SIGINT from itself, or when there is an error. None of these
984 * cases will allow the code to reach here.
985 */
986 Assert(false);
987}
static void pa_shutdown(int code, Datum arg)
#define PARALLEL_APPLY_KEY_SHARED
ParallelApplyWorkerShared * MyParallelShared
static void LogicalParallelApplyLoop(shm_mq_handle *mqh)
#define PARALLEL_APPLY_KEY_ERROR_QUEUE
#define PARALLEL_APPLY_KEY_MQ
#define PG_LOGICAL_APPLY_SHM_MAGIC
void ReplicationOriginNameForLogicalRep(Oid suboid, Oid relid, char *originname, Size szoriginname)
Definition: worker.c:641
void set_apply_error_context_origin(char *originname)
Definition: worker.c:6260
void InitializeLogRepWorker(void)
Definition: worker.c:5705
Subscription * MySubscription
Definition: worker.c:479
void BackgroundWorkerUnblockSignals(void)
Definition: bgworker.c:927
void * dsm_segment_address(dsm_segment *seg)
Definition: dsm.c:1095
dsm_segment * dsm_attach(dsm_handle h)
Definition: dsm.c:665
uint32 dsm_handle
Definition: dsm_impl.h:55
int errcode(int sqlerrcode)
Definition: elog.c:854
int errmsg(const char *fmt,...)
Definition: elog.c:1071
#define ERROR
Definition: elog.h:39
#define ereport(elevel,...)
Definition: elog.h:150
Assert(PointerIsAligned(start, uint64))
void SignalHandlerForShutdownRequest(SIGNAL_ARGS)
Definition: interrupt.c:104
void SignalHandlerForConfigReload(SIGNAL_ARGS)
Definition: interrupt.c:61
void CacheRegisterSyscacheCallback(int cacheid, SyscacheCallbackFunction func, Datum arg)
Definition: inval.c:1812
void before_shmem_exit(pg_on_exit_callback function, Datum arg)
Definition: ipc.c:337
void logicalrep_worker_attach(int slot)
Definition: launcher.c:731
RepOriginId replorigin_by_name(const char *roname, bool missing_ok)
Definition: origin.c:226
RepOriginId replorigin_session_origin
Definition: origin.c:163
void replorigin_session_setup(RepOriginId node, int acquired_by)
Definition: origin.c:1120
#define NAMEDATALEN
#define die(msg)
Definition: pg_test_fsync.c:100
#define pqsignal
Definition: port.h:531
static Datum PointerGetDatum(const void *X)
Definition: postgres.h:332
uint64_t Datum
Definition: postgres.h:70
#define InvalidOid
Definition: postgres_ext.h:37
BackgroundWorker * MyBgworkerEntry
Definition: postmaster.c:200
void pq_set_parallel_leader(pid_t pid, ProcNumber procNumber)
Definition: pqmq.c:82
void pq_redirect_to_shm_mq(dsm_segment *seg, shm_mq_handle *mqh)
Definition: pqmq.c:53
#define INVALID_PROC_NUMBER
Definition: procnumber.h:26
void shm_mq_set_sender(shm_mq *mq, PGPROC *proc)
Definition: shm_mq.c:224
void shm_mq_set_receiver(shm_mq *mq, PGPROC *proc)
Definition: shm_mq.c:206
shm_mq_handle * shm_mq_attach(shm_mq *mq, dsm_segment *seg, BackgroundWorkerHandle *handle)
Definition: shm_mq.c:290
void * shm_toc_lookup(shm_toc *toc, uint64 key, bool noError)
Definition: shm_toc.c:232
shm_toc * shm_toc_attach(uint64 magic, void *address)
Definition: shm_toc.c:64
#define SpinLockRelease(lock)
Definition: spin.h:61
#define SpinLockAcquire(lock)
Definition: spin.h:59
PGPROC * MyProc
Definition: proc.c:66
char bgw_extra[BGW_EXTRALEN]
Definition: bgworker.h:99
TimestampTz last_recv_time
TimestampTz reply_time
TimestampTz last_send_time
Definition: dsm.c:67
Definition: shm_mq.c:138
Definition: shm_mq.c:72
Definition: shm_toc.c:27
void invalidate_syncing_table_states(Datum arg, int cacheid, uint32 hashvalue)
Definition: tablesync.c:280
#define SIGHUP
Definition: win32_port.h:158
#define SIGUSR2
Definition: win32_port.h:171
void StartTransactionCommand(void)
Definition: xact.c:3071
void CommitTransactionCommand(void)
Definition: xact.c:3169
uint16 RepOriginId
Definition: xlogdefs.h:68

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, INVALID_PROC_NUMBER, invalidate_syncing_table_states(), 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(), SIGUSR2, SpinLockAcquire, SpinLockRelease, and StartTransactionCommand().

◆ ProcessParallelApplyMessages()

void ProcessParallelApplyMessages ( void  )

Definition at line 1071 of file applyparallelworker.c.

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

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

Referenced by ProcessInterrupts().

◆ TablesyncWorkerMain()

void TablesyncWorkerMain ( Datum  main_arg )

Definition at line 1750 of file tablesync.c.

1751{
1752 int worker_slot = DatumGetInt32(main_arg);
1753
1754 SetupApplyOrSyncWorker(worker_slot);
1755
1757
1759}
static pg_noreturn void finish_sync_worker(void)
Definition: tablesync.c:143
static void run_tablesync_worker()
Definition: tablesync.c:1724

References DatumGetInt32(), finish_sync_worker(), 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(), ProcessInterrupts(), and ProcessParallelApplyMessages().

AltStyle によって変換されたページ (->オリジナル) /