codekingpro/portable-devtools

1/*-------------------------------------------------------------------------2 *3 * walreceiver.h4 *	  Exports from replication/walreceiverfuncs.c.5 *6 * Portions Copyright (c) 2010-2023, PostgreSQL Global Development Group7 *8 * src/include/replication/walreceiver.h9 *10 *-------------------------------------------------------------------------11 */12#ifndef _WALRECEIVER_H13#define _WALRECEIVER_H14 15#include <netdb.h>16#include <sys/socket.h>17 18#include "access/xlog.h"19#include "access/xlogdefs.h"20#include "pgtime.h"21#include "port/atomics.h"22#include "replication/logicalproto.h"23#include "replication/walsender.h"24#include "storage/condition_variable.h"25#include "storage/latch.h"26#include "storage/spin.h"27#include "utils/tuplestore.h"28 29/* user-settable parameters */30extern PGDLLIMPORT int wal_receiver_status_interval;31extern PGDLLIMPORT int wal_receiver_timeout;32extern PGDLLIMPORT bool hot_standby_feedback;33 34/*35 * MAXCONNINFO: maximum size of a connection string.36 *37 * XXX: Should this move to pg_config_manual.h?38 */39#define MAXCONNINFO		102440 41/* Can we allow the standby to accept replication connection from another standby? */42#define AllowCascadeReplication() (EnableHotStandby && max_wal_senders > 0)43 44/*45 * Values for WalRcv->walRcvState.46 */47typedef enum48{49	WALRCV_STOPPED,				/* stopped and mustn't start up again */50	WALRCV_STARTING,			/* launched, but the process hasn't51								 * initialized yet */52	WALRCV_STREAMING,			/* walreceiver is streaming */53	WALRCV_WAITING,				/* stopped streaming, waiting for orders */54	WALRCV_RESTARTING,			/* asked to restart streaming */55	WALRCV_STOPPING				/* requested to stop, but still running */56} WalRcvState;57 58/* Shared memory area for management of walreceiver process */59typedef struct60{61	/*62	 * PID of currently active walreceiver process, its current state and63	 * start time (actually, the time at which it was requested to be64	 * started).65	 */66	pid_t		pid;67	WalRcvState walRcvState;68	ConditionVariable walRcvStoppedCV;69	pg_time_t	startTime;70 71	/*72	 * receiveStart and receiveStartTLI indicate the first byte position and73	 * timeline that will be received. When startup process starts the74	 * walreceiver, it sets these to the point where it wants the streaming to75	 * begin.76	 */77	XLogRecPtr	receiveStart;78	TimeLineID	receiveStartTLI;79 80	/*81	 * flushedUpto-1 is the last byte position that has already been received,82	 * and receivedTLI is the timeline it came from.  At the first startup of83	 * walreceiver, these are set to receiveStart and receiveStartTLI. After84	 * that, walreceiver updates these whenever it flushes the received WAL to85	 * disk.86	 */87	XLogRecPtr	flushedUpto;88	TimeLineID	receivedTLI;89 90	/*91	 * latestChunkStart is the starting byte position of the current "batch"92	 * of received WAL.  It's actually the same as the previous value of93	 * flushedUpto before the last flush to disk.  Startup process can use94	 * this to detect whether it's keeping up or not.95	 */96	XLogRecPtr	latestChunkStart;97 98	/*99	 * Time of send and receive of any message received.100	 */101	TimestampTz lastMsgSendTime;102	TimestampTz lastMsgReceiptTime;103 104	/*105	 * Latest reported end of WAL on the sender106	 */107	XLogRecPtr	latestWalEnd;108	TimestampTz latestWalEndTime;109 110	/*111	 * connection string; initially set to connect to the primary, and later112	 * clobbered to hide security-sensitive fields.113	 */114	char		conninfo[MAXCONNINFO];115 116	/*117	 * Host name (this can be a host name, an IP address, or a directory path)118	 * and port number of the active replication connection.119	 */120	char		sender_host[NI_MAXHOST];121	int			sender_port;122 123	/*124	 * replication slot name; is also used for walreceiver to connect with the125	 * primary126	 */127	char		slotname[NAMEDATALEN];128 129	/*130	 * If it's a temporary replication slot, it needs to be recreated when131	 * connecting.132	 */133	bool		is_temp_slot;134 135	/* set true once conninfo is ready to display (obfuscated pwds etc) */136	bool		ready_to_display;137 138	/*139	 * Latch used by startup process to wake up walreceiver after telling it140	 * where to start streaming (after setting receiveStart and141	 * receiveStartTLI), and also to tell it to send apply feedback to the142	 * primary whenever specially marked commit records are applied. This is143	 * normally mapped to procLatch when walreceiver is running.144	 */145	Latch	   *latch;146 147	slock_t		mutex;			/* locks shared variables shown above */148 149	/*150	 * Like flushedUpto, but advanced after writing and before flushing,151	 * without the need to acquire the spin lock.  Data can be read by another152	 * process up to this point, but shouldn't be used for data integrity153	 * purposes.154	 */155	pg_atomic_uint64 writtenUpto;156 157	/*158	 * force walreceiver reply?  This doesn't need to be locked; memory159	 * barriers for ordering are sufficient.  But we do need atomic fetch and160	 * store semantics, so use sig_atomic_t.161	 */162	sig_atomic_t force_reply;	/* used as a bool */163} WalRcvData;164 165extern PGDLLIMPORT WalRcvData *WalRcv;166 167typedef struct168{169	bool		logical;		/* True if this is logical replication stream,170								 * false if physical stream.  */171	char	   *slotname;		/* Name of the replication slot or NULL. */172	XLogRecPtr	startpoint;		/* LSN of starting point. */173 174	union175	{176		struct177		{178			TimeLineID	startpointTLI;	/* Starting timeline */179		}			physical;180		struct181		{182			uint32		proto_version;	/* Logical protocol version */183			List	   *publication_names;	/* String list of publications */184			bool		binary; /* Ask publisher to use binary */185			char	   *streaming_str;	/* Streaming of large transactions */186			bool		twophase;	/* Streaming of two-phase transactions at187									 * prepare time */188			char	   *origin; /* Only publish data originating from the189								 * specified origin */190		}			logical;191	}			proto;192} WalRcvStreamOptions;193 194struct WalReceiverConn;195typedef struct WalReceiverConn WalReceiverConn;196 197/*198 * Status of walreceiver query execution.199 *200 * We only define statuses that are currently used.201 */202typedef enum203{204	WALRCV_ERROR,				/* There was error when executing the query. */205	WALRCV_OK_COMMAND,			/* Query executed utility or replication206								 * command. */207	WALRCV_OK_TUPLES,			/* Query returned tuples. */208	WALRCV_OK_COPY_IN,			/* Query started COPY FROM. */209	WALRCV_OK_COPY_OUT,			/* Query started COPY TO. */210	WALRCV_OK_COPY_BOTH			/* Query started COPY BOTH replication211								 * protocol. */212} WalRcvExecStatus;213 214/*215 * Return value for walrcv_exec, returns the status of the execution and216 * tuples if any.217 */218typedef struct WalRcvExecResult219{220	WalRcvExecStatus status;221	int			sqlstate;222	char	   *err;223	Tuplestorestate *tuplestore;224	TupleDesc	tupledesc;225} WalRcvExecResult;226 227/* WAL receiver - libpqwalreceiver hooks */228 229/*230 * walrcv_connect_fn231 *232 * Establish connection to a cluster.  'logical' is true if the233 * connection is logical, and false if the connection is physical.234 * 'appname' is a name associated to the connection, to use for example235 * with fallback_application_name or application_name.  Returns the236 * details about the connection established, as defined by237 * WalReceiverConn for each WAL receiver module.  On error, NULL is238 * returned with 'err' including the error generated.239 */240typedef WalReceiverConn *(*walrcv_connect_fn) (const char *conninfo,241											   bool logical,242											   bool must_use_password,243											   const char *appname,244											   char **err);245 246/*247 * walrcv_check_conninfo_fn248 *249 * Parse and validate the connection string given as of 'conninfo'.250 */251typedef void (*walrcv_check_conninfo_fn) (const char *conninfo,252										  bool must_use_password);253 254/*255 * walrcv_get_conninfo_fn256 *257 * Returns a user-displayable conninfo string.  Note that any258 * security-sensitive fields should be obfuscated.259 */260typedef char *(*walrcv_get_conninfo_fn) (WalReceiverConn *conn);261 262/*263 * walrcv_get_senderinfo_fn264 *265 * Provide information of the WAL sender this WAL receiver is connected266 * to, as of 'sender_host' for the host of the sender and 'sender_port'267 * for its port.268 */269typedef void (*walrcv_get_senderinfo_fn) (WalReceiverConn *conn,270										  char **sender_host,271										  int *sender_port);272 273/*274 * walrcv_identify_system_fn275 *276 * Run IDENTIFY_SYSTEM on the cluster connected to and validate the277 * identity of the cluster.  Returns the system ID of the cluster278 * connected to.  'primary_tli' is the timeline ID of the sender.279 */280typedef char *(*walrcv_identify_system_fn) (WalReceiverConn *conn,281											TimeLineID *primary_tli);282 283/*284 * walrcv_server_version_fn285 *286 * Returns the version number of the cluster connected to.287 */288typedef int (*walrcv_server_version_fn) (WalReceiverConn *conn);289 290/*291 * walrcv_readtimelinehistoryfile_fn292 *293 * Fetch from cluster the timeline history file for timeline 'tli'.294 * Returns the name of the timeline history file as of 'filename', its295 * contents as of 'content' and its 'size'.296 */297typedef void (*walrcv_readtimelinehistoryfile_fn) (WalReceiverConn *conn,298												   TimeLineID tli,299												   char **filename,300												   char **content,301												   int *size);302 303/*304 * walrcv_startstreaming_fn305 *306 * Start streaming WAL data from given streaming options.  Returns true307 * if the connection has switched successfully to copy-both mode and false308 * if the server received the command and executed it successfully, but309 * didn't switch to copy-mode.310 */311typedef bool (*walrcv_startstreaming_fn) (WalReceiverConn *conn,312										  const WalRcvStreamOptions *options);313 314/*315 * walrcv_endstreaming_fn316 *317 * Stop streaming of WAL data.  Returns the next timeline ID of the cluster318 * connected to in 'next_tli', or 0 if there was no report.319 */320typedef void (*walrcv_endstreaming_fn) (WalReceiverConn *conn,321										TimeLineID *next_tli);322 323/*324 * walrcv_receive_fn325 *326 * Receive a message available from the WAL stream.  'buffer' is a pointer327 * to a buffer holding the message received.  Returns the length of the data,328 * 0 if no data is available yet ('wait_fd' is a socket descriptor which can329 * be waited on before a retry), and -1 if the cluster ended the COPY.330 */331typedef int (*walrcv_receive_fn) (WalReceiverConn *conn,332								  char **buffer,333								  pgsocket *wait_fd);334 335/*336 * walrcv_send_fn337 *338 * Send a message of size 'nbytes' to the WAL stream with 'buffer' as339 * contents.340 */341typedef void (*walrcv_send_fn) (WalReceiverConn *conn,342								const char *buffer,343								int nbytes);344 345/*346 * walrcv_create_slot_fn347 *348 * Create a new replication slot named 'slotname'.  'temporary' defines349 * if the slot is temporary.  'snapshot_action' defines the behavior wanted350 * for an exported snapshot (see replication protocol for more details).351 * 'lsn' includes the LSN position at which the created slot became352 * consistent.  Returns the name of the exported snapshot for a logical353 * slot, or NULL for a physical slot.354 */355typedef char *(*walrcv_create_slot_fn) (WalReceiverConn *conn,356										const char *slotname,357										bool temporary,358										bool two_phase,359										CRSSnapshotAction snapshot_action,360										XLogRecPtr *lsn);361 362/*363 * walrcv_get_backend_pid_fn364 *365 * Returns the PID of the remote backend process.366 */367typedef pid_t (*walrcv_get_backend_pid_fn) (WalReceiverConn *conn);368 369/*370 * walrcv_exec_fn371 *372 * Send generic queries (and commands) to the remote cluster.  'nRetTypes'373 * is the expected number of returned attributes, and 'retTypes' an array374 * including their type OIDs.  Returns the status of the execution and375 * tuples if any.376 */377typedef WalRcvExecResult *(*walrcv_exec_fn) (WalReceiverConn *conn,378											 const char *query,379											 const int nRetTypes,380											 const Oid *retTypes);381 382/*383 * walrcv_disconnect_fn384 *385 * Disconnect with the cluster.386 */387typedef void (*walrcv_disconnect_fn) (WalReceiverConn *conn);388 389typedef struct WalReceiverFunctionsType390{391	walrcv_connect_fn walrcv_connect;392	walrcv_check_conninfo_fn walrcv_check_conninfo;393	walrcv_get_conninfo_fn walrcv_get_conninfo;394	walrcv_get_senderinfo_fn walrcv_get_senderinfo;395	walrcv_identify_system_fn walrcv_identify_system;396	walrcv_server_version_fn walrcv_server_version;397	walrcv_readtimelinehistoryfile_fn walrcv_readtimelinehistoryfile;398	walrcv_startstreaming_fn walrcv_startstreaming;399	walrcv_endstreaming_fn walrcv_endstreaming;400	walrcv_receive_fn walrcv_receive;401	walrcv_send_fn walrcv_send;402	walrcv_create_slot_fn walrcv_create_slot;403	walrcv_get_backend_pid_fn walrcv_get_backend_pid;404	walrcv_exec_fn walrcv_exec;405	walrcv_disconnect_fn walrcv_disconnect;406} WalReceiverFunctionsType;407 408extern PGDLLIMPORT WalReceiverFunctionsType *WalReceiverFunctions;409 410#define walrcv_connect(conninfo, logical, must_use_password, appname, err) \411	WalReceiverFunctions->walrcv_connect(conninfo, logical, must_use_password, appname, err)412#define walrcv_check_conninfo(conninfo, must_use_password) \413	WalReceiverFunctions->walrcv_check_conninfo(conninfo, must_use_password)414#define walrcv_get_conninfo(conn) \415	WalReceiverFunctions->walrcv_get_conninfo(conn)416#define walrcv_get_senderinfo(conn, sender_host, sender_port) \417	WalReceiverFunctions->walrcv_get_senderinfo(conn, sender_host, sender_port)418#define walrcv_identify_system(conn, primary_tli) \419	WalReceiverFunctions->walrcv_identify_system(conn, primary_tli)420#define walrcv_server_version(conn) \421	WalReceiverFunctions->walrcv_server_version(conn)422#define walrcv_readtimelinehistoryfile(conn, tli, filename, content, size) \423	WalReceiverFunctions->walrcv_readtimelinehistoryfile(conn, tli, filename, content, size)424#define walrcv_startstreaming(conn, options) \425	WalReceiverFunctions->walrcv_startstreaming(conn, options)426#define walrcv_endstreaming(conn, next_tli) \427	WalReceiverFunctions->walrcv_endstreaming(conn, next_tli)428#define walrcv_receive(conn, buffer, wait_fd) \429	WalReceiverFunctions->walrcv_receive(conn, buffer, wait_fd)430#define walrcv_send(conn, buffer, nbytes) \431	WalReceiverFunctions->walrcv_send(conn, buffer, nbytes)432#define walrcv_create_slot(conn, slotname, temporary, two_phase, snapshot_action, lsn) \433	WalReceiverFunctions->walrcv_create_slot(conn, slotname, temporary, two_phase, snapshot_action, lsn)434#define walrcv_get_backend_pid(conn) \435	WalReceiverFunctions->walrcv_get_backend_pid(conn)436#define walrcv_exec(conn, exec, nRetTypes, retTypes) \437	WalReceiverFunctions->walrcv_exec(conn, exec, nRetTypes, retTypes)438#define walrcv_disconnect(conn) \439	WalReceiverFunctions->walrcv_disconnect(conn)440 441static inline void442walrcv_clear_result(WalRcvExecResult *walres)443{444	if (!walres)445		return;446 447	if (walres->err)448		pfree(walres->err);449 450	if (walres->tuplestore)451		tuplestore_end(walres->tuplestore);452 453	if (walres->tupledesc)454		FreeTupleDesc(walres->tupledesc);455 456	pfree(walres);457}458 459/* prototypes for functions in walreceiver.c */460extern void WalReceiverMain(void) pg_attribute_noreturn();461extern void ProcessWalRcvInterrupts(void);462extern void WalRcvForceReply(void);463 464/* prototypes for functions in walreceiverfuncs.c */465extern Size WalRcvShmemSize(void);466extern void WalRcvShmemInit(void);467extern void ShutdownWalRcv(void);468extern bool WalRcvStreaming(void);469extern bool WalRcvRunning(void);470extern void RequestXLogStreaming(TimeLineID tli, XLogRecPtr recptr,471								 const char *conninfo, const char *slotname,472								 bool create_temp_slot);473extern XLogRecPtr GetWalRcvFlushRecPtr(XLogRecPtr *latestChunkStart, TimeLineID *receiveTLI);474extern XLogRecPtr GetWalRcvWriteRecPtr(void);475extern int	GetReplicationApplyDelay(void);476extern int	GetReplicationTransferLatency(void);477 478#endif							/* _WALRECEIVER_H */479