Line data    Source code

       1             : /*-------------------------------------------------------------------------
       2             :  *
       3             :  * worker_internal.h
       4             :  *    Internal headers shared by logical replication workers.
       5             :  *
       6             :  * Portions Copyright (c) 2016-2021, 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 <signal.h>
      16             : 
      17             : #include "access/xlogdefs.h"
      18             : #include "catalog/pg_subscription.h"
      19             : #include "datatype/timestamp.h"
      20             : #include "storage/fileset.h"
      21             : #include "storage/lock.h"
      22             : #include "storage/spin.h"
      23             : 
      24             : 
      25             : typedef struct LogicalRepWorker
      26             : {
      27             :     /* Time at which this worker was launched. */
      28             :     TimestampTz launch_time;
      29             : 
      30             :     /* Indicates if this slot is used or free. */
      31             :     bool        in_use;
      32             : 
      33             :     /* Increased every time the slot is taken by new worker. */
      34             :     uint16      generation;
      35             : 
      36             :     /* Pointer to proc array. NULL if not running. */
      37             :     PGPROC     *proc;
      38             : 
      39             :     /* Database id to connect to. */
      40             :     Oid         dbid;
      41             : 
      42             :     /* User to use for connection (will be same as owner of subscription). */
      43             :     Oid         userid;
      44             : 
      45             :     /* Subscription id for the worker. */
      46             :     Oid         subid;
      47             : 
      48             :     /* Used for initial table synchronization. */
      49             :     Oid         relid;
      50             :     char        relstate;
      51             :     XLogRecPtr  relstate_lsn;
      52             :     slock_t     relmutex;
      53             : 
      54             :     /*
      55             :      * Used to create the changes and subxact files for the streaming
      56             :      * transactions.  Upon the arrival of the first streaming transaction, the
      57             :      * fileset will be initialized, and it will be deleted when the worker
      58             :      * exits.  Under this, separate buffiles would be created for each
      59             :      * transaction which will be deleted after the transaction is finished.
      60             :      */
      61             :     FileSet    *stream_fileset;
      62             : 
      63             :     /* Stats. */
      64             :     XLogRecPtr  last_lsn;
      65             :     TimestampTz last_send_time;
      66             :     TimestampTz last_recv_time;
      67             :     XLogRecPtr  reply_lsn;
      68             :     TimestampTz reply_time;
      69             : } LogicalRepWorker;
      70             : 
      71             : /* Main memory context for apply worker. Permanent during worker lifetime. */
      72             : extern MemoryContext ApplyContext;
      73             : 
      74             : /* libpqreceiver connection */
      75             : extern struct WalReceiverConn *LogRepWorkerWalRcvConn;
      76             : 
      77             : /* Worker and subscription objects. */
      78             : extern Subscription *MySubscription;
      79             : extern LogicalRepWorker *MyLogicalRepWorker;
      80             : 
      81             : extern bool in_remote_transaction;
      82             : 
      83             : extern void logicalrep_worker_attach(int slot);
      84             : extern LogicalRepWorker *logicalrep_worker_find(Oid subid, Oid relid,
      85             :                                                 bool only_running);
      86             : extern List *logicalrep_workers_find(Oid subid, bool only_running);
      87             : extern void logicalrep_worker_launch(Oid dbid, Oid subid, const char *subname,
      88             :                                      Oid userid, Oid relid);
      89             : extern void logicalrep_worker_stop(Oid subid, Oid relid);
      90             : extern void logicalrep_worker_wakeup(Oid subid, Oid relid);
      91             : extern void logicalrep_worker_wakeup_ptr(LogicalRepWorker *worker);
      92             : 
      93             : extern int  logicalrep_sync_worker_count(Oid subid);
      94             : 
      95             : extern void ReplicationOriginNameForTablesync(Oid suboid, Oid relid,
      96             :                                               char *originname, int szorgname);
      97             : extern char *LogicalRepSyncTableStart(XLogRecPtr *origin_startpos);
      98             : 
      99             : extern bool AllTablesyncsReady(void);
     100             : extern void UpdateTwoPhaseState(Oid suboid, char new_state);
     101             : 
     102             : void        process_syncing_tables(XLogRecPtr current_lsn);
     103             : void        invalidate_syncing_table_states(Datum arg, int cacheid,
     104             :                                             uint32 hashvalue);
     105             : 
     106             : static inline bool
     107      230042 : am_tablesync_worker(void)
     108             : {
     109      230042 :     return OidIsValid(MyLogicalRepWorker->relid);
     110             : }
     111             : 
     112             : #endif                          /* WORKER_INTERNAL_H */