LCOV - code coverage report
Current view: top level - src/include/replication - worker_internal.h (source / functions) Hit Total Coverage
Test: PostgreSQL 19devel Lines: 6 6 100.0 %
Date: 2025-07-29 03:18:01 Functions: 3 3 100.0 %
Legend: Lines: hit not hit

          Line data    Source code
       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"
      16             : #include "catalog/pg_subscription.h"
      17             : #include "datatype/timestamp.h"
      18             : #include "miscadmin.h"
      19             : #include "replication/logicalrelation.h"
      20             : #include "replication/walreceiver.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 */
      29             : typedef enum LogicalRepWorkerType
      30             : {
      31             :     WORKERTYPE_UNKNOWN = 0,
      32             :     WORKERTYPE_TABLESYNC,
      33             :     WORKERTYPE_APPLY,
      34             :     WORKERTYPE_PARALLEL_APPLY,
      35             : } LogicalRepWorkerType;
      36             : 
      37             : typedef struct LogicalRepWorker
      38             : {
      39             :     /* What type of worker is this? */
      40             :     LogicalRepWorkerType type;
      41             : 
      42             :     /* Time at which this worker was launched. */
      43             :     TimestampTz launch_time;
      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. */
      49             :     uint16      generation;
      50             : 
      51             :     /* Pointer to proc array. NULL if not running. */
      52             :     PGPROC     *proc;
      53             : 
      54             :     /* Database id to connect to. */
      55             :     Oid         dbid;
      56             : 
      57             :     /* User to use for connection (will be same as owner of subscription). */
      58             :     Oid         userid;
      59             : 
      60             :     /* Subscription id for the worker. */
      61             :     Oid         subid;
      62             : 
      63             :     /* Used for initial table synchronization. */
      64             :     Oid         relid;
      65             :     char        relstate;
      66             :     XLogRecPtr  relstate_lsn;
      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             :      */
      78             :     FileSet    *stream_fileset;
      79             : 
      80             :     /*
      81             :      * PID of leader apply worker if this slot is used for a parallel apply
      82             :      * worker, InvalidPid otherwise.
      83             :      */
      84             :     pid_t       leader_pid;
      85             : 
      86             :     /* Indicates whether apply can be performed in parallel. */
      87             :     bool        parallel_apply;
      88             : 
      89             :     /*
      90             :      * The changes made by this and later transactions must be retained to
      91             :      * ensure reliable conflict detection during the apply phase.
      92             :      *
      93             :      * The logical replication launcher manages an internal replication slot
      94             :      * named "pg_conflict_detection". It asynchronously collects this ID to
      95             :      * decide when to advance the xmin value of the slot.
      96             :      */
      97             :     TransactionId oldest_nonremovable_xid;
      98             : 
      99             :     /* Stats. */
     100             :     XLogRecPtr  last_lsn;
     101             :     TimestampTz last_send_time;
     102             :     TimestampTz last_recv_time;
     103             :     XLogRecPtr  reply_lsn;
     104             :     TimestampTz reply_time;
     105             : } LogicalRepWorker;
     106             : 
     107             : /*
     108             :  * State of the transaction in parallel apply worker.
     109             :  *
     110             :  * The enum values must have the same order as the transaction state
     111             :  * transitions.
     112             :  */
     113             : typedef enum ParallelTransState
     114             : {
     115             :     PARALLEL_TRANS_UNKNOWN,
     116             :     PARALLEL_TRANS_STARTED,
     117             :     PARALLEL_TRANS_FINISHED,
     118             : } ParallelTransState;
     119             : 
     120             : /*
     121             :  * State of fileset used to communicate changes from leader to parallel
     122             :  * apply worker.
     123             :  *
     124             :  * FS_EMPTY indicates an initial state where the leader doesn't need to use
     125             :  * the file to communicate with the parallel apply worker.
     126             :  *
     127             :  * FS_SERIALIZE_IN_PROGRESS indicates that the leader is serializing changes
     128             :  * to the file.
     129             :  *
     130             :  * FS_SERIALIZE_DONE indicates that the leader has serialized all changes to
     131             :  * the file.
     132             :  *
     133             :  * FS_READY indicates that it is now ok for a parallel apply worker to
     134             :  * read the file.
     135             :  */
     136             : typedef enum PartialFileSetState
     137             : {
     138             :     FS_EMPTY,
     139             :     FS_SERIALIZE_IN_PROGRESS,
     140             :     FS_SERIALIZE_DONE,
     141             :     FS_READY,
     142             : } PartialFileSetState;
     143             : 
     144             : /*
     145             :  * Struct for sharing information between leader apply worker and parallel
     146             :  * apply workers.
     147             :  */
     148             : typedef struct ParallelApplyWorkerShared
     149             : {
     150             :     slock_t     mutex;
     151             : 
     152             :     TransactionId xid;
     153             : 
     154             :     /*
     155             :      * State used to ensure commit ordering.
     156             :      *
     157             :      * The parallel apply worker will set it to PARALLEL_TRANS_FINISHED after
     158             :      * handling the transaction finish commands while the apply leader will
     159             :      * wait for it to become PARALLEL_TRANS_FINISHED before proceeding in
     160             :      * transaction finish commands (e.g. STREAM_COMMIT/STREAM_PREPARE/
     161             :      * STREAM_ABORT).
     162             :      */
     163             :     ParallelTransState xact_state;
     164             : 
     165             :     /* Information from the corresponding LogicalRepWorker slot. */
     166             :     uint16      logicalrep_worker_generation;
     167             :     int         logicalrep_worker_slot_no;
     168             : 
     169             :     /*
     170             :      * Indicates whether there are pending streaming blocks in the queue. The
     171             :      * parallel apply worker will check it before starting to wait.
     172             :      */
     173             :     pg_atomic_uint32 pending_stream_count;
     174             : 
     175             :     /*
     176             :      * XactLastCommitEnd from the parallel apply worker. This is required by
     177             :      * the leader worker so it can update the lsn_mappings.
     178             :      */
     179             :     XLogRecPtr  last_commit_end;
     180             : 
     181             :     /*
     182             :      * After entering PARTIAL_SERIALIZE mode, the leader apply worker will
     183             :      * serialize changes to the file, and share the fileset with the parallel
     184             :      * apply worker when processing the transaction finish command. Then the
     185             :      * parallel apply worker will apply all the spooled messages.
     186             :      *
     187             :      * FileSet is used here instead of SharedFileSet because we need it to
     188             :      * survive after releasing the shared memory so that the leader apply
     189             :      * worker can re-use the same fileset for the next streaming transaction.
     190             :      */
     191             :     PartialFileSetState fileset_state;
     192             :     FileSet     fileset;
     193             : } ParallelApplyWorkerShared;
     194             : 
     195             : /*
     196             :  * Information which is used to manage the parallel apply worker.
     197             :  */
     198             : typedef struct ParallelApplyWorkerInfo
     199             : {
     200             :     /*
     201             :      * This queue is used to send changes from the leader apply worker to the
     202             :      * parallel apply worker.
     203             :      */
     204             :     shm_mq_handle *mq_handle;
     205             : 
     206             :     /*
     207             :      * This queue is used to transfer error messages from the parallel apply
     208             :      * worker to the leader apply worker.
     209             :      */
     210             :     shm_mq_handle *error_mq_handle;
     211             : 
     212             :     dsm_segment *dsm_seg;
     213             : 
     214             :     /*
     215             :      * Indicates whether the leader apply worker needs to serialize the
     216             :      * remaining changes to a file due to timeout when attempting to send data
     217             :      * to the parallel apply worker via shared memory.
     218             :      */
     219             :     bool        serialize_changes;
     220             : 
     221             :     /*
     222             :      * True if the worker is being used to process a parallel apply
     223             :      * transaction. False indicates this worker is available for re-use.
     224             :      */
     225             :     bool        in_use;
     226             : 
     227             :     ParallelApplyWorkerShared *shared;
     228             : } ParallelApplyWorkerInfo;
     229             : 
     230             : /* Main memory context for apply worker. Permanent during worker lifetime. */
     231             : extern PGDLLIMPORT MemoryContext ApplyContext;
     232             : 
     233             : extern PGDLLIMPORT MemoryContext ApplyMessageContext;
     234             : 
     235             : extern PGDLLIMPORT ErrorContextCallback *apply_error_context_stack;
     236             : 
     237             : extern PGDLLIMPORT ParallelApplyWorkerShared *MyParallelShared;
     238             : 
     239             : /* libpqreceiver connection */
     240             : extern PGDLLIMPORT struct WalReceiverConn *LogRepWorkerWalRcvConn;
     241             : 
     242             : /* Worker and subscription objects. */
     243             : extern PGDLLIMPORT Subscription *MySubscription;
     244             : extern PGDLLIMPORT LogicalRepWorker *MyLogicalRepWorker;
     245             : 
     246             : extern PGDLLIMPORT bool in_remote_transaction;
     247             : 
     248             : extern PGDLLIMPORT bool InitializingApplyWorker;
     249             : 
     250             : extern void logicalrep_worker_attach(int slot);
     251             : extern LogicalRepWorker *logicalrep_worker_find(Oid subid, Oid relid,
     252             :                                                 bool only_running);
     253             : extern List *logicalrep_workers_find(Oid subid, bool only_running,
     254             :                                      bool acquire_lock);
     255             : extern bool logicalrep_worker_launch(LogicalRepWorkerType wtype,
     256             :                                      Oid dbid, Oid subid, const char *subname,
     257             :                                      Oid userid, Oid relid,
     258             :                                      dsm_handle subworker_dsm,
     259             :                                      bool retain_dead_tuples);
     260             : extern void logicalrep_worker_stop(Oid subid, Oid relid);
     261             : extern void logicalrep_pa_worker_stop(ParallelApplyWorkerInfo *winfo);
     262             : extern void logicalrep_worker_wakeup(Oid subid, Oid relid);
     263             : extern void logicalrep_worker_wakeup_ptr(LogicalRepWorker *worker);
     264             : 
     265             : extern int  logicalrep_sync_worker_count(Oid subid);
     266             : 
     267             : extern void ReplicationOriginNameForLogicalRep(Oid suboid, Oid relid,
     268             :                                                char *originname, Size szoriginname);
     269             : 
     270             : extern bool AllTablesyncsReady(void);
     271             : extern void UpdateTwoPhaseState(Oid suboid, char new_state);
     272             : 
     273             : extern void process_syncing_tables(XLogRecPtr current_lsn);
     274             : extern void invalidate_syncing_table_states(Datum arg, int cacheid,
     275             :                                             uint32 hashvalue);
     276             : 
     277             : extern void stream_start_internal(TransactionId xid, bool first_segment);
     278             : extern void stream_stop_internal(TransactionId xid);
     279             : 
     280             : /* Common streaming function to apply all the spooled messages */
     281             : extern void apply_spooled_messages(FileSet *stream_fileset, TransactionId xid,
     282             :                                    XLogRecPtr lsn);
     283             : 
     284             : extern void apply_dispatch(StringInfo s);
     285             : 
     286             : extern void maybe_reread_subscription(void);
     287             : 
     288             : extern void stream_cleanup_files(Oid subid, TransactionId xid);
     289             : 
     290             : extern void set_stream_options(WalRcvStreamOptions *options,
     291             :                                char *slotname,
     292             :                                XLogRecPtr *origin_startpos);
     293             : 
     294             : extern void start_apply(XLogRecPtr origin_startpos);
     295             : 
     296             : extern void InitializeLogRepWorker(void);
     297             : 
     298             : extern void SetupApplyOrSyncWorker(int worker_slot);
     299             : 
     300             : extern void DisableSubscriptionAndExit(void);
     301             : 
     302             : extern void store_flush_position(XLogRecPtr remote_lsn, XLogRecPtr local_lsn);
     303             : 
     304             : /* Function for apply error callback */
     305             : extern void apply_error_callback(void *arg);
     306             : extern void set_apply_error_context_origin(char *originname);
     307             : 
     308             : /* Parallel apply worker setup and interactions */
     309             : extern void pa_allocate_worker(TransactionId xid);
     310             : extern ParallelApplyWorkerInfo *pa_find_worker(TransactionId xid);
     311             : extern void pa_detach_all_error_mq(void);
     312             : 
     313             : extern bool pa_send_data(ParallelApplyWorkerInfo *winfo, Size nbytes,
     314             :                          const void *data);
     315             : extern void pa_switch_to_partial_serialize(ParallelApplyWorkerInfo *winfo,
     316             :                                            bool stream_locked);
     317             : 
     318             : extern void pa_set_xact_state(ParallelApplyWorkerShared *wshared,
     319             :                               ParallelTransState xact_state);
     320             : extern void pa_set_stream_apply_worker(ParallelApplyWorkerInfo *winfo);
     321             : 
     322             : extern void pa_start_subtrans(TransactionId current_xid,
     323             :                               TransactionId top_xid);
     324             : extern void pa_reset_subtrans(void);
     325             : extern void pa_stream_abort(LogicalRepStreamAbortData *abort_data);
     326             : extern void pa_set_fileset_state(ParallelApplyWorkerShared *wshared,
     327             :                                  PartialFileSetState fileset_state);
     328             : 
     329             : extern void pa_lock_stream(TransactionId xid, LOCKMODE lockmode);
     330             : extern void pa_unlock_stream(TransactionId xid, LOCKMODE lockmode);
     331             : 
     332             : extern void pa_lock_transaction(TransactionId xid, LOCKMODE lockmode);
     333             : extern void pa_unlock_transaction(TransactionId xid, LOCKMODE lockmode);
     334             : 
     335             : extern void pa_decr_and_wait_stream_block(void);
     336             : 
     337             : extern void pa_xact_finish(ParallelApplyWorkerInfo *winfo,
     338             :                            XLogRecPtr remote_lsn);
     339             : 
     340             : #define isParallelApplyWorker(worker) ((worker)->in_use && \
     341             :                                        (worker)->type == WORKERTYPE_PARALLEL_APPLY)
     342             : #define isTablesyncWorker(worker) ((worker)->in_use && \
     343             :                                    (worker)->type == WORKERTYPE_TABLESYNC)
     344             : 
     345             : static inline bool
     346        1122 : am_tablesync_worker(void)
     347             : {
     348        1122 :     return isTablesyncWorker(MyLogicalRepWorker);
     349             : }
     350             : 
     351             : static inline bool
     352      512710 : am_leader_apply_worker(void)
     353             : {
     354             :     Assert(MyLogicalRepWorker->in_use);
     355      512710 :     return (MyLogicalRepWorker->type == WORKERTYPE_APPLY);
     356             : }
     357             : 
     358             : static inline bool
     359      659798 : am_parallel_apply_worker(void)
     360             : {
     361             :     Assert(MyLogicalRepWorker->in_use);
     362      659798 :     return isParallelApplyWorker(MyLogicalRepWorker);
     363             : }
     364             : 
     365             : #endif                          /* WORKER_INTERNAL_H */

Generated by: LCOV version 1.16