worker_internal.h

Go to the documentation of this file.

/*-------------------------------------------------------------------------
 *
 * worker_internal.h
 *    Internal headers shared by logical replication workers.
 *
 * Portions Copyright (c) 2016-2025, PostgreSQL Global Development Group
 *
 * src/include/replication/worker_internal.h
 *
 *-------------------------------------------------------------------------
 */
#ifndef WORKER_INTERNAL_H
#define WORKER_INTERNAL_H
 
#include "access/xlogdefs.h"
#include "catalog/pg_subscription.h"
#include "datatype/timestamp.h"
#include "miscadmin.h"
#include "replication/logicalrelation.h"
#include "replication/walreceiver.h"
#include "storage/buffile.h"
#include "storage/fileset.h"
#include "storage/lock.h"
#include "storage/shm_mq.h"
#include "storage/shm_toc.h"
#include "storage/spin.h"
 
/* Different types of worker */
typedef enum LogicalRepWorkerType
{
    WORKERTYPE_UNKNOWN = 0,
    WORKERTYPE_TABLESYNC,
    WORKERTYPE_SEQUENCESYNC,
    WORKERTYPE_APPLY,
    WORKERTYPE_PARALLEL_APPLY,
} LogicalRepWorkerType;
 
typedef struct LogicalRepWorker
{
    /* What type of worker is this? */
    LogicalRepWorkerType type;
 
    /* Time at which this worker was launched. */
    TimestampTz launch_time;
 
    /* Indicates if this slot is used or free. */
    bool        in_use;
 
    /* Increased every time the slot is taken by new worker. */
    uint16      generation;
 
    /* Pointer to proc array. NULL if not running. */
    PGPROC     *proc;
 
    /* Database id to connect to. */
    Oid         dbid;
 
    /* User to use for connection (will be same as owner of subscription). */
    Oid         userid;
 
    /* Subscription id for the worker. */
    Oid         subid;
 
    /* Used for initial table synchronization. */
    Oid         relid;
    char        relstate;
    XLogRecPtr  relstate_lsn;
    slock_t     relmutex;
 
    /*
     * Used to create the changes and subxact files for the streaming
     * transactions.  Upon the arrival of the first streaming transaction or
     * when the first-time leader apply worker times out while sending changes
     * to the parallel apply worker, the fileset will be initialized, and it
     * will be deleted when the worker exits.  Under this, separate buffiles
     * would be created for each transaction which will be deleted after the
     * transaction is finished.
     */
    FileSet    *stream_fileset;
 
    /*
     * PID of leader apply worker if this slot is used for a parallel apply
     * worker, InvalidPid otherwise.
     */
    pid_t       leader_pid;
 
    /* Indicates whether apply can be performed in parallel. */
    bool        parallel_apply;
 
    /*
     * Changes made by this transaction and subsequent ones must be preserved.
     * This ensures that update_deleted conflicts can be accurately detected
     * during the apply phase of logical replication by this worker.
     *
     * The logical replication launcher manages an internal replication slot
     * named "pg_conflict_detection". It asynchronously collects this ID to
     * decide when to advance the xmin value of the slot.
     *
     * This ID is set to InvalidTransactionId when the apply worker stops
     * retaining information needed for conflict detection.
     */
    TransactionId oldest_nonremovable_xid;
 
    /* Stats. */
    XLogRecPtr  last_lsn;
    TimestampTz last_send_time;
    TimestampTz last_recv_time;
    XLogRecPtr  reply_lsn;
    TimestampTz reply_time;
 
    TimestampTz last_seqsync_start_time;
} LogicalRepWorker;
 
/*
 * State of the transaction in parallel apply worker.
 *
 * The enum values must have the same order as the transaction state
 * transitions.
 */
typedef enum ParallelTransState
{
    PARALLEL_TRANS_UNKNOWN,
    PARALLEL_TRANS_STARTED,
    PARALLEL_TRANS_FINISHED,
} ParallelTransState;
 
/*
 * State of fileset used to communicate changes from leader to parallel
 * apply worker.
 *
 * FS_EMPTY indicates an initial state where the leader doesn't need to use
 * the file to communicate with the parallel apply worker.
 *
 * FS_SERIALIZE_IN_PROGRESS indicates that the leader is serializing changes
 * to the file.
 *
 * FS_SERIALIZE_DONE indicates that the leader has serialized all changes to
 * the file.
 *
 * FS_READY indicates that it is now ok for a parallel apply worker to
 * read the file.
 */
typedef enum PartialFileSetState
{
    FS_EMPTY,
    FS_SERIALIZE_IN_PROGRESS,
    FS_SERIALIZE_DONE,
    FS_READY,
} PartialFileSetState;
 
/*
 * Struct for sharing information between leader apply worker and parallel
 * apply workers.
 */
typedef struct ParallelApplyWorkerShared
{
    slock_t     mutex;
 
    TransactionId xid;
 
    /*
     * State used to ensure commit ordering.
     *
     * The parallel apply worker will set it to PARALLEL_TRANS_FINISHED after
     * handling the transaction finish commands while the apply leader will
     * wait for it to become PARALLEL_TRANS_FINISHED before proceeding in
     * transaction finish commands (e.g. STREAM_COMMIT/STREAM_PREPARE/
     * STREAM_ABORT).
     */
    ParallelTransState xact_state;
 
    /* Information from the corresponding LogicalRepWorker slot. */
    uint16      logicalrep_worker_generation;
    int         logicalrep_worker_slot_no;
 
    /*
     * Indicates whether there are pending streaming blocks in the queue. The
     * parallel apply worker will check it before starting to wait.
     */
    pg_atomic_uint32 pending_stream_count;
 
    /*
     * XactLastCommitEnd from the parallel apply worker. This is required by
     * the leader worker so it can update the lsn_mappings.
     */
    XLogRecPtr  last_commit_end;
 
    /*
     * After entering PARTIAL_SERIALIZE mode, the leader apply worker will
     * serialize changes to the file, and share the fileset with the parallel
     * apply worker when processing the transaction finish command. Then the
     * parallel apply worker will apply all the spooled messages.
     *
     * FileSet is used here instead of SharedFileSet because we need it to
     * survive after releasing the shared memory so that the leader apply
     * worker can re-use the same fileset for the next streaming transaction.
     */
    PartialFileSetState fileset_state;
    FileSet     fileset;
} ParallelApplyWorkerShared;
 
/*
 * Information which is used to manage the parallel apply worker.
 */
typedef struct ParallelApplyWorkerInfo
{
    /*
     * This queue is used to send changes from the leader apply worker to the
     * parallel apply worker.
     */
    shm_mq_handle *mq_handle;
 
    /*
     * This queue is used to transfer error messages from the parallel apply
     * worker to the leader apply worker.
     */
    shm_mq_handle *error_mq_handle;
 
    dsm_segment *dsm_seg;
 
    /*
     * Indicates whether the leader apply worker needs to serialize the
     * remaining changes to a file due to timeout when attempting to send data
     * to the parallel apply worker via shared memory.
     */
    bool        serialize_changes;
 
    /*
     * True if the worker is being used to process a parallel apply
     * transaction. False indicates this worker is available for re-use.
     */
    bool        in_use;
 
    ParallelApplyWorkerShared *shared;
} ParallelApplyWorkerInfo;
 
/* Main memory context for apply worker. Permanent during worker lifetime. */
extern PGDLLIMPORT MemoryContext ApplyContext;
 
extern PGDLLIMPORT MemoryContext ApplyMessageContext;
 
extern PGDLLIMPORT ErrorContextCallback *apply_error_context_stack;
 
extern PGDLLIMPORT ParallelApplyWorkerShared *MyParallelShared;
 
/* libpqreceiver connection */
extern PGDLLIMPORT struct WalReceiverConn *LogRepWorkerWalRcvConn;
 
/* Worker and subscription objects. */
extern PGDLLIMPORT Subscription *MySubscription;
extern PGDLLIMPORT LogicalRepWorker *MyLogicalRepWorker;
 
extern PGDLLIMPORT bool in_remote_transaction;
 
extern PGDLLIMPORT bool InitializingApplyWorker;
 
extern PGDLLIMPORT List *table_states_not_ready;
 
extern void logicalrep_worker_attach(int slot);
extern LogicalRepWorker *logicalrep_worker_find(LogicalRepWorkerType wtype,
                                                Oid subid, Oid relid,
                                                bool only_running);
extern List *logicalrep_workers_find(Oid subid, bool only_running,
                                     bool acquire_lock);
extern bool logicalrep_worker_launch(LogicalRepWorkerType wtype,
                                     Oid dbid, Oid subid, const char *subname,
                                     Oid userid, Oid relid,
                                     dsm_handle subworker_dsm,
                                     bool retain_dead_tuples);
extern void logicalrep_worker_stop(LogicalRepWorkerType wtype, Oid subid,
                                   Oid relid);
extern void logicalrep_pa_worker_stop(ParallelApplyWorkerInfo *winfo);
extern void logicalrep_worker_wakeup(LogicalRepWorkerType wtype, Oid subid,
                                     Oid relid);
extern void logicalrep_worker_wakeup_ptr(LogicalRepWorker *worker);
 
extern void logicalrep_reset_seqsync_start_time(void);
extern int  logicalrep_sync_worker_count(Oid subid);
 
extern void ReplicationOriginNameForLogicalRep(Oid suboid, Oid relid,
                                               char *originname, Size szoriginname);
 
extern bool AllTablesyncsReady(void);
extern bool HasSubscriptionTablesCached(void);
extern void UpdateTwoPhaseState(Oid suboid, char new_state);
 
extern void ProcessSyncingTablesForSync(XLogRecPtr current_lsn);
extern void ProcessSyncingTablesForApply(XLogRecPtr current_lsn);
extern void ProcessSequencesForSync(void);
 
pg_noreturn extern void FinishSyncWorker(void);
extern void InvalidateSyncingRelStates(Datum arg, int cacheid, uint32 hashvalue);
extern void launch_sync_worker(LogicalRepWorkerType wtype, int nsyncworkers,
                               Oid relid, TimestampTz *last_start_time);
extern void ProcessSyncingRelations(XLogRecPtr current_lsn);
extern void FetchRelationStates(bool *has_pending_subtables,
                                bool *has_pending_sequences, bool *started_tx);
 
extern void stream_start_internal(TransactionId xid, bool first_segment);
extern void stream_stop_internal(TransactionId xid);
 
/* Common streaming function to apply all the spooled messages */
extern void apply_spooled_messages(FileSet *stream_fileset, TransactionId xid,
                                   XLogRecPtr lsn);
 
extern void apply_dispatch(StringInfo s);
 
extern void maybe_reread_subscription(void);
 
extern void stream_cleanup_files(Oid subid, TransactionId xid);
 
extern void set_stream_options(WalRcvStreamOptions *options,
                               char *slotname,
                               XLogRecPtr *origin_startpos);
 
extern void start_apply(XLogRecPtr origin_startpos);
 
extern void InitializeLogRepWorker(void);
 
extern void SetupApplyOrSyncWorker(int worker_slot);
 
extern void DisableSubscriptionAndExit(void);
 
extern void store_flush_position(XLogRecPtr remote_lsn, XLogRecPtr local_lsn);
 
/* Function for apply error callback */
extern void apply_error_callback(void *arg);
extern void set_apply_error_context_origin(char *originname);
 
/* Parallel apply worker setup and interactions */
extern void pa_allocate_worker(TransactionId xid);
extern ParallelApplyWorkerInfo *pa_find_worker(TransactionId xid);
extern void pa_detach_all_error_mq(void);
 
extern bool pa_send_data(ParallelApplyWorkerInfo *winfo, Size nbytes,
                         const void *data);
extern void pa_switch_to_partial_serialize(ParallelApplyWorkerInfo *winfo,
                                           bool stream_locked);
 
extern void pa_set_xact_state(ParallelApplyWorkerShared *wshared,
                              ParallelTransState xact_state);
extern void pa_set_stream_apply_worker(ParallelApplyWorkerInfo *winfo);
 
extern void pa_start_subtrans(TransactionId current_xid,
                              TransactionId top_xid);
extern void pa_reset_subtrans(void);
extern void pa_stream_abort(LogicalRepStreamAbortData *abort_data);
extern void pa_set_fileset_state(ParallelApplyWorkerShared *wshared,
                                 PartialFileSetState fileset_state);
 
extern void pa_lock_stream(TransactionId xid, LOCKMODE lockmode);
extern void pa_unlock_stream(TransactionId xid, LOCKMODE lockmode);
 
extern void pa_lock_transaction(TransactionId xid, LOCKMODE lockmode);
extern void pa_unlock_transaction(TransactionId xid, LOCKMODE lockmode);
 
extern void pa_decr_and_wait_stream_block(void);
 
extern void pa_xact_finish(ParallelApplyWorkerInfo *winfo,
                           XLogRecPtr remote_lsn);
 
#define isParallelApplyWorker(worker) ((worker)->in_use && \
                                       (worker)->type == WORKERTYPE_PARALLEL_APPLY)
#define isTableSyncWorker(worker) ((worker)->in_use && \
                                   (worker)->type == WORKERTYPE_TABLESYNC)
#define isSequenceSyncWorker(worker) ((worker)->in_use && \
                                      (worker)->type == WORKERTYPE_SEQUENCESYNC)
 
static inline bool
am_tablesync_worker(void)
{
    return isTableSyncWorker(MyLogicalRepWorker);
}
 
static inline bool
am_sequencesync_worker(void)
{
    return isSequenceSyncWorker(MyLogicalRepWorker);
}
 
static inline bool
am_leader_apply_worker(void)
{
    Assert(MyLogicalRepWorker->in_use);
    return (MyLogicalRepWorker->type == WORKERTYPE_APPLY);
}
 
static inline bool
am_parallel_apply_worker(void)
{
    Assert(MyLogicalRepWorker->in_use);
    return isParallelApplyWorker(MyLogicalRepWorker);
}
 
#endif                          /* WORKER_INTERNAL_H */