PostgreSQL Source Code git master
worker_internal.h
Go to the documentation of this file.
1/*-------------------------------------------------------------------------
2 *
3 * worker_internal.h
4 * Internal headers shared by logical replication workers.
5 *
6 * Portions Copyright (c) 2016-2025, PostgreSQL Global Development Group
7 *
8 * src/include/replication/worker_internal.h
9 *
10 *-------------------------------------------------------------------------
11 */
12#ifndef WORKER_INTERNAL_H
13#define WORKER_INTERNAL_H
14
15#include "access/xlogdefs.h"
17#include "datatype/timestamp.h"
18#include "miscadmin.h"
21#include "storage/buffile.h"
22#include "storage/fileset.h"
23#include "storage/lock.h"
24#include "storage/shm_mq.h"
25#include "storage/shm_toc.h"
26#include "storage/spin.h"
27
28/* Different types of worker */
30{
36
37typedef struct LogicalRepWorker
38{
39 /* What type of worker is this? */
41
42 /* Time at which this worker was launched. */
44
45 /* Indicates if this slot is used or free. */
46 bool in_use;
47
48 /* Increased every time the slot is taken by new worker. */
50
51 /* Pointer to proc array. NULL if not running. */
53
54 /* Database id to connect to. */
56
57 /* User to use for connection (will be same as owner of subscription). */
59
60 /* Subscription id for the worker. */
62
63 /* Used for initial table synchronization. */
67 slock_t relmutex;
68
69 /*
70 * Used to create the changes and subxact files for the streaming
71 * transactions. Upon the arrival of the first streaming transaction or
72 * when the first-time leader apply worker times out while sending changes
73 * to the parallel apply worker, the fileset will be initialized, and it
74 * will be deleted when the worker exits. Under this, separate buffiles
75 * would be created for each transaction which will be deleted after the
76 * transaction is finished.
77 */
79
80 /*
81 * PID of leader apply worker if this slot is used for a parallel apply
82 * worker, InvalidPid otherwise.
83 */
85
86 /* Indicates whether apply can be performed in parallel. */
88
89 /* Stats. */
96
97/*
98 * State of the transaction in parallel apply worker.
99 *
100 * The enum values must have the same order as the transaction state
101 * transitions.
102 */
104{
109
110/*
111 * State of fileset used to communicate changes from leader to parallel
112 * apply worker.
113 *
114 * FS_EMPTY indicates an initial state where the leader doesn't need to use
115 * the file to communicate with the parallel apply worker.
116 *
117 * FS_SERIALIZE_IN_PROGRESS indicates that the leader is serializing changes
118 * to the file.
119 *
120 * FS_SERIALIZE_DONE indicates that the leader has serialized all changes to
121 * the file.
122 *
123 * FS_READY indicates that it is now ok for a parallel apply worker to
124 * read the file.
125 */
127{
133
134/*
135 * Struct for sharing information between leader apply worker and parallel
136 * apply workers.
137 */
139{
140 slock_t mutex;
141
143
144 /*
145 * State used to ensure commit ordering.
146 *
147 * The parallel apply worker will set it to PARALLEL_TRANS_FINISHED after
148 * handling the transaction finish commands while the apply leader will
149 * wait for it to become PARALLEL_TRANS_FINISHED before proceeding in
150 * transaction finish commands (e.g. STREAM_COMMIT/STREAM_PREPARE/
151 * STREAM_ABORT).
152 */
154
155 /* Information from the corresponding LogicalRepWorker slot. */
158
159 /*
160 * Indicates whether there are pending streaming blocks in the queue. The
161 * parallel apply worker will check it before starting to wait.
162 */
164
165 /*
166 * XactLastCommitEnd from the parallel apply worker. This is required by
167 * the leader worker so it can update the lsn_mappings.
168 */
170
171 /*
172 * After entering PARTIAL_SERIALIZE mode, the leader apply worker will
173 * serialize changes to the file, and share the fileset with the parallel
174 * apply worker when processing the transaction finish command. Then the
175 * parallel apply worker will apply all the spooled messages.
176 *
177 * FileSet is used here instead of SharedFileSet because we need it to
178 * survive after releasing the shared memory so that the leader apply
179 * worker can re-use the same fileset for the next streaming transaction.
180 */
184
185/*
186 * Information which is used to manage the parallel apply worker.
187 */
189{
190 /*
191 * This queue is used to send changes from the leader apply worker to the
192 * parallel apply worker.
193 */
195
196 /*
197 * This queue is used to transfer error messages from the parallel apply
198 * worker to the leader apply worker.
199 */
201
203
204 /*
205 * Indicates whether the leader apply worker needs to serialize the
206 * remaining changes to a file due to timeout when attempting to send data
207 * to the parallel apply worker via shared memory.
208 */
210
211 /*
212 * True if the worker is being used to process a parallel apply
213 * transaction. False indicates this worker is available for re-use.
214 */
215 bool in_use;
216
219
220/* Main memory context for apply worker. Permanent during worker lifetime. */
222
224
226
228
229/* libpqreceiver connection */
231
232/* Worker and subscription objects. */
235
237
239
240extern void logicalrep_worker_attach(int slot);
242 bool only_running);
243extern List *logicalrep_workers_find(Oid subid, bool only_running,
244 bool acquire_lock);
246 Oid dbid, Oid subid, const char *subname,
247 Oid userid, Oid relid,
248 dsm_handle subworker_dsm);
249extern void logicalrep_worker_stop(Oid subid, Oid relid);
251extern void logicalrep_worker_wakeup(Oid subid, Oid relid);
253
254extern int logicalrep_sync_worker_count(Oid subid);
255
256extern void ReplicationOriginNameForLogicalRep(Oid suboid, Oid relid,
257 char *originname, Size szoriginname);
258
259extern bool AllTablesyncsReady(void);
260extern void UpdateTwoPhaseState(Oid suboid, char new_state);
261
262extern void process_syncing_tables(XLogRecPtr current_lsn);
263extern void invalidate_syncing_table_states(Datum arg, int cacheid,
264 uint32 hashvalue);
265
266extern void stream_start_internal(TransactionId xid, bool first_segment);
267extern void stream_stop_internal(TransactionId xid);
268
269/* Common streaming function to apply all the spooled messages */
270extern void apply_spooled_messages(FileSet *stream_fileset, TransactionId xid,
271 XLogRecPtr lsn);
272
273extern void apply_dispatch(StringInfo s);
274
275extern void maybe_reread_subscription(void);
276
277extern void stream_cleanup_files(Oid subid, TransactionId xid);
278
280 char *slotname,
281 XLogRecPtr *origin_startpos);
282
283extern void start_apply(XLogRecPtr origin_startpos);
284
285extern void InitializeLogRepWorker(void);
286
287extern void SetupApplyOrSyncWorker(int worker_slot);
288
289extern void DisableSubscriptionAndExit(void);
290
291extern void store_flush_position(XLogRecPtr remote_lsn, XLogRecPtr local_lsn);
292
293/* Function for apply error callback */
294extern void apply_error_callback(void *arg);
295extern void set_apply_error_context_origin(char *originname);
296
297/* Parallel apply worker setup and interactions */
298extern void pa_allocate_worker(TransactionId xid);
300extern void pa_detach_all_error_mq(void);
301
302extern bool pa_send_data(ParallelApplyWorkerInfo *winfo, Size nbytes,
303 const void *data);
305 bool stream_locked);
306
308 ParallelTransState xact_state);
310
311extern void pa_start_subtrans(TransactionId current_xid,
312 TransactionId top_xid);
313extern void pa_reset_subtrans(void);
314extern void pa_stream_abort(LogicalRepStreamAbortData *abort_data);
316 PartialFileSetState fileset_state);
317
318extern void pa_lock_stream(TransactionId xid, LOCKMODE lockmode);
319extern void pa_unlock_stream(TransactionId xid, LOCKMODE lockmode);
320
321extern void pa_lock_transaction(TransactionId xid, LOCKMODE lockmode);
322extern void pa_unlock_transaction(TransactionId xid, LOCKMODE lockmode);
323
324extern void pa_decr_and_wait_stream_block(void);
325
326extern void pa_xact_finish(ParallelApplyWorkerInfo *winfo,
327 XLogRecPtr remote_lsn);
328
329#define isParallelApplyWorker(worker) ((worker)->in_use && \
330 (worker)->type == WORKERTYPE_PARALLEL_APPLY)
331#define isTablesyncWorker(worker) ((worker)->in_use && \
332 (worker)->type == WORKERTYPE_TABLESYNC)
333
334static inline bool
336{
338}
339
340static inline bool
342{
345}
346
347static inline bool
349{
352}
353
354#endif /* WORKER_INTERNAL_H */
#define PGDLLIMPORT
Definition: c.h:1277
#define Assert(condition)
Definition: c.h:815
uint16_t uint16
Definition: c.h:487
uint32_t uint32
Definition: c.h:488
uint32 TransactionId
Definition: c.h:609
size_t Size
Definition: c.h:562
int64 TimestampTz
Definition: timestamp.h:39
uint32 dsm_handle
Definition: dsm_impl.h:55
int LOCKMODE
Definition: lockdefs.h:26
void * arg
const void * data
NameData subname
uintptr_t Datum
Definition: postgres.h:69
unsigned int Oid
Definition: postgres_ext.h:32
Definition: pg_list.h:54
XLogRecPtr relstate_lsn
TimestampTz last_recv_time
LogicalRepWorkerType type
TimestampTz launch_time
TimestampTz reply_time
FileSet * stream_fileset
XLogRecPtr reply_lsn
XLogRecPtr last_lsn
TimestampTz last_send_time
Definition: proc.h:162
shm_mq_handle * error_mq_handle
shm_mq_handle * mq_handle
ParallelApplyWorkerShared * shared
pg_atomic_uint32 pending_stream_count
PartialFileSetState fileset_state
ParallelTransState xact_state
bool AllTablesyncsReady(void)
Definition: tablesync.c:1739
ParallelTransState
@ PARALLEL_TRANS_UNKNOWN
@ PARALLEL_TRANS_STARTED
@ PARALLEL_TRANS_FINISHED
#define isParallelApplyWorker(worker)
void set_stream_options(WalRcvStreamOptions *options, char *slotname, XLogRecPtr *origin_startpos)
Definition: worker.c:4434
bool logicalrep_worker_launch(LogicalRepWorkerType wtype, Oid dbid, Oid subid, const char *subname, Oid userid, Oid relid, dsm_handle subworker_dsm)
Definition: launcher.c:297
List * logicalrep_workers_find(Oid subid, bool only_running, bool acquire_lock)
Definition: launcher.c:266
void pa_set_xact_state(ParallelApplyWorkerShared *wshared, ParallelTransState xact_state)
void stream_cleanup_files(Oid subid, TransactionId xid)
Definition: worker.c:4301
void pa_unlock_stream(TransactionId xid, LOCKMODE lockmode)
PGDLLIMPORT ErrorContextCallback * apply_error_context_stack
Definition: worker.c:289
PGDLLIMPORT bool in_remote_transaction
Definition: worker.c:304
void logicalrep_worker_wakeup_ptr(LogicalRepWorker *worker)
Definition: launcher.c:693
PGDLLIMPORT MemoryContext ApplyMessageContext
Definition: worker.c:291
void invalidate_syncing_table_states(Datum arg, int cacheid, uint32 hashvalue)
Definition: tablesync.c:281
static bool am_parallel_apply_worker(void)
PGDLLIMPORT LogicalRepWorker * MyLogicalRepWorker
Definition: launcher.c:54
struct ParallelApplyWorkerShared ParallelApplyWorkerShared
void logicalrep_worker_attach(int slot)
Definition: launcher.c:704
void stream_stop_internal(TransactionId xid)
Definition: worker.c:1625
void start_apply(XLogRecPtr origin_startpos)
Definition: worker.c:4503
void pa_stream_abort(LogicalRepStreamAbortData *abort_data)
struct ParallelApplyWorkerInfo ParallelApplyWorkerInfo
void pa_lock_stream(TransactionId xid, LOCKMODE lockmode)
void apply_dispatch(StringInfo s)
Definition: worker.c:3365
void pa_set_fileset_state(ParallelApplyWorkerShared *wshared, PartialFileSetState fileset_state)
void pa_reset_subtrans(void)
void logicalrep_pa_worker_stop(ParallelApplyWorkerInfo *winfo)
Definition: launcher.c:630
void DisableSubscriptionAndExit(void)
Definition: worker.c:4808
void process_syncing_tables(XLogRecPtr current_lsn)
Definition: tablesync.c:667
void pa_lock_transaction(TransactionId xid, LOCKMODE lockmode)
void ReplicationOriginNameForLogicalRep(Oid suboid, Oid relid, char *originname, Size szoriginname)
Definition: worker.c:426
LogicalRepWorker * logicalrep_worker_find(Oid subid, Oid relid, bool only_running)
Definition: launcher.c:234
void pa_detach_all_error_mq(void)
PGDLLIMPORT struct WalReceiverConn * LogRepWorkerWalRcvConn
Definition: worker.c:297
PGDLLIMPORT bool InitializingApplyWorker
Definition: worker.c:319
void stream_start_internal(TransactionId xid, bool first_segment)
Definition: worker.c:1451
void pa_start_subtrans(TransactionId current_xid, TransactionId top_xid)
void pa_switch_to_partial_serialize(ParallelApplyWorkerInfo *winfo, bool stream_locked)
void logicalrep_worker_wakeup(Oid subid, Oid relid)
Definition: launcher.c:673
void logicalrep_worker_stop(Oid subid, Oid relid)
Definition: launcher.c:606
void set_apply_error_context_origin(char *originname)
Definition: worker.c:5149
LogicalRepWorkerType
@ WORKERTYPE_TABLESYNC
@ WORKERTYPE_UNKNOWN
@ WORKERTYPE_PARALLEL_APPLY
@ WORKERTYPE_APPLY
struct LogicalRepWorker LogicalRepWorker
void pa_xact_finish(ParallelApplyWorkerInfo *winfo, XLogRecPtr remote_lsn)
PartialFileSetState
@ FS_EMPTY
@ FS_SERIALIZE_DONE
@ FS_READY
@ FS_SERIALIZE_IN_PROGRESS
#define isTablesyncWorker(worker)
void SetupApplyOrSyncWorker(int worker_slot)
Definition: worker.c:4734
static bool am_tablesync_worker(void)
static bool am_leader_apply_worker(void)
PGDLLIMPORT Subscription * MySubscription
Definition: worker.c:299
void apply_error_callback(void *arg)
Definition: worker.c:4996
void store_flush_position(XLogRecPtr remote_lsn, XLogRecPtr local_lsn)
Definition: worker.c:3529
int logicalrep_sync_worker_count(Oid subid)
Definition: launcher.c:845
void maybe_reread_subscription(void)
Definition: worker.c:3958
void InitializeLogRepWorker(void)
Definition: worker.c:4640
bool pa_send_data(ParallelApplyWorkerInfo *winfo, Size nbytes, const void *data)
void pa_allocate_worker(TransactionId xid)
PGDLLIMPORT ParallelApplyWorkerShared * MyParallelShared
void apply_spooled_messages(FileSet *stream_fileset, TransactionId xid, XLogRecPtr lsn)
Definition: worker.c:2023
void UpdateTwoPhaseState(Oid suboid, char new_state)
Definition: tablesync.c:1764
void pa_set_stream_apply_worker(ParallelApplyWorkerInfo *winfo)
ParallelApplyWorkerInfo * pa_find_worker(TransactionId xid)
void pa_unlock_transaction(TransactionId xid, LOCKMODE lockmode)
void pa_decr_and_wait_stream_block(void)
PGDLLIMPORT MemoryContext ApplyContext
Definition: worker.c:292
uint64 XLogRecPtr
Definition: xlogdefs.h:21