codekingpro/portable-devtools

1/*-------------------------------------------------------------------------2 *3 * worker_internal.h4 *	  Internal headers shared by logical replication workers.5 *6 * Portions Copyright (c) 2016-2023, PostgreSQL Global Development Group7 *8 * src/include/replication/worker_internal.h9 *10 *-------------------------------------------------------------------------11 */12#ifndef WORKER_INTERNAL_H13#define WORKER_INTERNAL_H14 15#include <signal.h>16 17#include "access/xlogdefs.h"18#include "catalog/pg_subscription.h"19#include "datatype/timestamp.h"20#include "miscadmin.h"21#include "replication/logicalrelation.h"22#include "storage/buffile.h"23#include "storage/fileset.h"24#include "storage/lock.h"25#include "storage/shm_mq.h"26#include "storage/shm_toc.h"27#include "storage/spin.h"28 29 30typedef struct LogicalRepWorker31{32	/* Time at which this worker was launched. */33	TimestampTz launch_time;34 35	/* Indicates if this slot is used or free. */36	bool		in_use;37 38	/* Increased every time the slot is taken by new worker. */39	uint16		generation;40 41	/* Pointer to proc array. NULL if not running. */42	PGPROC	   *proc;43 44	/* Database id to connect to. */45	Oid			dbid;46 47	/* User to use for connection (will be same as owner of subscription). */48	Oid			userid;49 50	/* Subscription id for the worker. */51	Oid			subid;52 53	/* Used for initial table synchronization. */54	Oid			relid;55	char		relstate;56	XLogRecPtr	relstate_lsn;57	slock_t		relmutex;58 59	/*60	 * Used to create the changes and subxact files for the streaming61	 * transactions.  Upon the arrival of the first streaming transaction or62	 * when the first-time leader apply worker times out while sending changes63	 * to the parallel apply worker, the fileset will be initialized, and it64	 * will be deleted when the worker exits.  Under this, separate buffiles65	 * would be created for each transaction which will be deleted after the66	 * transaction is finished.67	 */68	FileSet    *stream_fileset;69 70	/*71	 * PID of leader apply worker if this slot is used for a parallel apply72	 * worker, InvalidPid otherwise.73	 */74	pid_t		leader_pid;75 76	/* Indicates whether apply can be performed in parallel. */77	bool		parallel_apply;78 79	/* Stats. */80	XLogRecPtr	last_lsn;81	TimestampTz last_send_time;82	TimestampTz last_recv_time;83	XLogRecPtr	reply_lsn;84	TimestampTz reply_time;85} LogicalRepWorker;86 87/*88 * State of the transaction in parallel apply worker.89 *90 * The enum values must have the same order as the transaction state91 * transitions.92 */93typedef enum ParallelTransState94{95	PARALLEL_TRANS_UNKNOWN,96	PARALLEL_TRANS_STARTED,97	PARALLEL_TRANS_FINISHED98} ParallelTransState;99 100/*101 * State of fileset used to communicate changes from leader to parallel102 * apply worker.103 *104 * FS_EMPTY indicates an initial state where the leader doesn't need to use105 * the file to communicate with the parallel apply worker.106 *107 * FS_SERIALIZE_IN_PROGRESS indicates that the leader is serializing changes108 * to the file.109 *110 * FS_SERIALIZE_DONE indicates that the leader has serialized all changes to111 * the file.112 *113 * FS_READY indicates that it is now ok for a parallel apply worker to114 * read the file.115 */116typedef enum PartialFileSetState117{118	FS_EMPTY,119	FS_SERIALIZE_IN_PROGRESS,120	FS_SERIALIZE_DONE,121	FS_READY122} PartialFileSetState;123 124/*125 * Struct for sharing information between leader apply worker and parallel126 * apply workers.127 */128typedef struct ParallelApplyWorkerShared129{130	slock_t		mutex;131 132	TransactionId xid;133 134	/*135	 * State used to ensure commit ordering.136	 *137	 * The parallel apply worker will set it to PARALLEL_TRANS_FINISHED after138	 * handling the transaction finish commands while the apply leader will139	 * wait for it to become PARALLEL_TRANS_FINISHED before proceeding in140	 * transaction finish commands (e.g. STREAM_COMMIT/STREAM_PREPARE/141	 * STREAM_ABORT).142	 */143	ParallelTransState xact_state;144 145	/* Information from the corresponding LogicalRepWorker slot. */146	uint16		logicalrep_worker_generation;147	int			logicalrep_worker_slot_no;148 149	/*150	 * Indicates whether there are pending streaming blocks in the queue. The151	 * parallel apply worker will check it before starting to wait.152	 */153	pg_atomic_uint32 pending_stream_count;154 155	/*156	 * XactLastCommitEnd from the parallel apply worker. This is required by157	 * the leader worker so it can update the lsn_mappings.158	 */159	XLogRecPtr	last_commit_end;160 161	/*162	 * After entering PARTIAL_SERIALIZE mode, the leader apply worker will163	 * serialize changes to the file, and share the fileset with the parallel164	 * apply worker when processing the transaction finish command. Then the165	 * parallel apply worker will apply all the spooled messages.166	 *167	 * FileSet is used here instead of SharedFileSet because we need it to168	 * survive after releasing the shared memory so that the leader apply169	 * worker can re-use the same fileset for the next streaming transaction.170	 */171	PartialFileSetState fileset_state;172	FileSet		fileset;173} ParallelApplyWorkerShared;174 175/*176 * Information which is used to manage the parallel apply worker.177 */178typedef struct ParallelApplyWorkerInfo179{180	/*181	 * This queue is used to send changes from the leader apply worker to the182	 * parallel apply worker.183	 */184	shm_mq_handle *mq_handle;185 186	/*187	 * This queue is used to transfer error messages from the parallel apply188	 * worker to the leader apply worker.189	 */190	shm_mq_handle *error_mq_handle;191 192	dsm_segment *dsm_seg;193 194	/*195	 * Indicates whether the leader apply worker needs to serialize the196	 * remaining changes to a file due to timeout when attempting to send data197	 * to the parallel apply worker via shared memory.198	 */199	bool		serialize_changes;200 201	/*202	 * True if the worker is being used to process a parallel apply203	 * transaction. False indicates this worker is available for re-use.204	 */205	bool		in_use;206 207	ParallelApplyWorkerShared *shared;208} ParallelApplyWorkerInfo;209 210/* Main memory context for apply worker. Permanent during worker lifetime. */211extern PGDLLIMPORT MemoryContext ApplyContext;212 213extern PGDLLIMPORT MemoryContext ApplyMessageContext;214 215extern PGDLLIMPORT ErrorContextCallback *apply_error_context_stack;216 217extern PGDLLIMPORT ParallelApplyWorkerShared *MyParallelShared;218 219/* libpqreceiver connection */220extern PGDLLIMPORT struct WalReceiverConn *LogRepWorkerWalRcvConn;221 222/* Worker and subscription objects. */223extern PGDLLIMPORT Subscription *MySubscription;224extern PGDLLIMPORT LogicalRepWorker *MyLogicalRepWorker;225 226extern PGDLLIMPORT bool in_remote_transaction;227 228extern PGDLLIMPORT bool InitializingApplyWorker;229 230extern void logicalrep_worker_attach(int slot);231extern LogicalRepWorker *logicalrep_worker_find(Oid subid, Oid relid,232												bool only_running);233extern List *logicalrep_workers_find(Oid subid, bool only_running);234extern bool logicalrep_worker_launch(Oid dbid, Oid subid, const char *subname,235									 Oid userid, Oid relid,236									 dsm_handle subworker_dsm);237extern void logicalrep_worker_stop(Oid subid, Oid relid);238extern void logicalrep_pa_worker_stop(ParallelApplyWorkerInfo *winfo);239extern void logicalrep_worker_wakeup(Oid subid, Oid relid);240extern void logicalrep_worker_wakeup_ptr(LogicalRepWorker *worker);241 242extern int	logicalrep_sync_worker_count(Oid subid);243 244extern void ReplicationOriginNameForLogicalRep(Oid suboid, Oid relid,245											   char *originname, Size szoriginname);246extern char *LogicalRepSyncTableStart(XLogRecPtr *origin_startpos);247 248extern bool AllTablesyncsReady(void);249extern void UpdateTwoPhaseState(Oid suboid, char new_state);250 251extern void process_syncing_tables(XLogRecPtr current_lsn);252extern void invalidate_syncing_table_states(Datum arg, int cacheid,253											uint32 hashvalue);254 255extern void stream_start_internal(TransactionId xid, bool first_segment);256extern void stream_stop_internal(TransactionId xid);257 258/* Common streaming function to apply all the spooled messages */259extern void apply_spooled_messages(FileSet *stream_fileset, TransactionId xid,260								   XLogRecPtr lsn);261 262extern void apply_dispatch(StringInfo s);263 264extern void maybe_reread_subscription(void);265 266extern void stream_cleanup_files(Oid subid, TransactionId xid);267 268extern void InitializeApplyWorker(void);269 270extern void store_flush_position(XLogRecPtr remote_lsn, XLogRecPtr local_lsn);271 272/* Function for apply error callback */273extern void apply_error_callback(void *arg);274extern void set_apply_error_context_origin(char *originname);275 276/* Parallel apply worker setup and interactions */277extern void pa_allocate_worker(TransactionId xid);278extern ParallelApplyWorkerInfo *pa_find_worker(TransactionId xid);279extern void pa_detach_all_error_mq(void);280 281extern bool pa_send_data(ParallelApplyWorkerInfo *winfo, Size nbytes,282						 const void *data);283extern void pa_switch_to_partial_serialize(ParallelApplyWorkerInfo *winfo,284										   bool stream_locked);285 286extern void pa_set_xact_state(ParallelApplyWorkerShared *wshared,287							  ParallelTransState xact_state);288extern void pa_set_stream_apply_worker(ParallelApplyWorkerInfo *winfo);289 290extern void pa_start_subtrans(TransactionId current_xid,291							  TransactionId top_xid);292extern void pa_reset_subtrans(void);293extern void pa_stream_abort(LogicalRepStreamAbortData *abort_data);294extern void pa_set_fileset_state(ParallelApplyWorkerShared *wshared,295								 PartialFileSetState fileset_state);296 297extern void pa_lock_stream(TransactionId xid, LOCKMODE lockmode);298extern void pa_unlock_stream(TransactionId xid, LOCKMODE lockmode);299 300extern void pa_lock_transaction(TransactionId xid, LOCKMODE lockmode);301extern void pa_unlock_transaction(TransactionId xid, LOCKMODE lockmode);302 303extern void pa_decr_and_wait_stream_block(void);304 305extern void pa_xact_finish(ParallelApplyWorkerInfo *winfo,306						   XLogRecPtr remote_lsn);307 308#define isParallelApplyWorker(worker) ((worker)->leader_pid != InvalidPid)309 310static inline bool311am_tablesync_worker(void)312{313	return OidIsValid(MyLogicalRepWorker->relid);314}315 316static inline bool317am_leader_apply_worker(void)318{319	return (!am_tablesync_worker() &&320			!isParallelApplyWorker(MyLogicalRepWorker));321}322 323static inline bool324am_parallel_apply_worker(void)325{326	return isParallelApplyWorker(MyLogicalRepWorker);327}328 329#endif							/* WORKER_INTERNAL_H */330