PostgreSQL Source Code  git master
walreceiver.h
Go to the documentation of this file.
1 /*-------------------------------------------------------------------------
2  *
3  * walreceiver.h
4  * Exports from replication/walreceiverfuncs.c.
5  *
6  * Portions Copyright (c) 2010-2021, PostgreSQL Global Development Group
7  *
8  * src/include/replication/walreceiver.h
9  *
10  *-------------------------------------------------------------------------
11  */
12 #ifndef _WALRECEIVER_H
13 #define _WALRECEIVER_H
14 
15 #include "access/xlog.h"
16 #include "access/xlogdefs.h"
17 #include "getaddrinfo.h" /* for NI_MAXHOST */
18 #include "pgtime.h"
19 #include "port/atomics.h"
21 #include "replication/walsender.h"
23 #include "storage/latch.h"
24 #include "storage/spin.h"
25 #include "utils/tuplestore.h"
26 
27 /* user-settable parameters */
29 extern int wal_receiver_timeout;
30 extern bool hot_standby_feedback;
31 
32 /*
33  * MAXCONNINFO: maximum size of a connection string.
34  *
35  * XXX: Should this move to pg_config_manual.h?
36  */
37 #define MAXCONNINFO 1024
38 
39 /* Can we allow the standby to accept replication connection from another standby? */
40 #define AllowCascadeReplication() (EnableHotStandby && max_wal_senders > 0)
41 
42 /*
43  * Values for WalRcv->walRcvState.
44  */
45 typedef enum
46 {
47  WALRCV_STOPPED, /* stopped and mustn't start up again */
48  WALRCV_STARTING, /* launched, but the process hasn't
49  * initialized yet */
50  WALRCV_STREAMING, /* walreceiver is streaming */
51  WALRCV_WAITING, /* stopped streaming, waiting for orders */
52  WALRCV_RESTARTING, /* asked to restart streaming */
53  WALRCV_STOPPING /* requested to stop, but still running */
54 } WalRcvState;
55 
56 /* Shared memory area for management of walreceiver process */
57 typedef struct
58 {
59  /*
60  * PID of currently active walreceiver process, its current state and
61  * start time (actually, the time at which it was requested to be
62  * started).
63  */
64  pid_t pid;
68 
69  /*
70  * receiveStart and receiveStartTLI indicate the first byte position and
71  * timeline that will be received. When startup process starts the
72  * walreceiver, it sets these to the point where it wants the streaming to
73  * begin.
74  */
77 
78  /*
79  * flushedUpto-1 is the last byte position that has already been received,
80  * and receivedTLI is the timeline it came from. At the first startup of
81  * walreceiver, these are set to receiveStart and receiveStartTLI. After
82  * that, walreceiver updates these whenever it flushes the received WAL to
83  * disk.
84  */
87 
88  /*
89  * latestChunkStart is the starting byte position of the current "batch"
90  * of received WAL. It's actually the same as the previous value of
91  * flushedUpto before the last flush to disk. Startup process can use
92  * this to detect whether it's keeping up or not.
93  */
95 
96  /*
97  * Time of send and receive of any message received.
98  */
101 
102  /*
103  * Latest reported end of WAL on the sender
104  */
107 
108  /*
109  * connection string; initially set to connect to the primary, and later
110  * clobbered to hide security-sensitive fields.
111  */
112  char conninfo[MAXCONNINFO];
113 
114  /*
115  * Host name (this can be a host name, an IP address, or a directory path)
116  * and port number of the active replication connection.
117  */
118  char sender_host[NI_MAXHOST];
120 
121  /*
122  * replication slot name; is also used for walreceiver to connect with the
123  * primary
124  */
125  char slotname[NAMEDATALEN];
126 
127  /*
128  * If it's a temporary replication slot, it needs to be recreated when
129  * connecting.
130  */
132 
133  /* set true once conninfo is ready to display (obfuscated pwds etc) */
135 
136  /*
137  * Latch used by startup process to wake up walreceiver after telling it
138  * where to start streaming (after setting receiveStart and
139  * receiveStartTLI), and also to tell it to send apply feedback to the
140  * primary whenever specially marked commit records are applied. This is
141  * normally mapped to procLatch when walreceiver is running.
142  */
144 
145  slock_t mutex; /* locks shared variables shown above */
146 
147  /*
148  * Like flushedUpto, but advanced after writing and before flushing,
149  * without the need to acquire the spin lock. Data can be read by another
150  * process up to this point, but shouldn't be used for data integrity
151  * purposes.
152  */
154 
155  /*
156  * force walreceiver reply? This doesn't need to be locked; memory
157  * barriers for ordering are sufficient. But we do need atomic fetch and
158  * store semantics, so use sig_atomic_t.
159  */
160  sig_atomic_t force_reply; /* used as a bool */
161 } WalRcvData;
162 
163 extern WalRcvData *WalRcv;
164 
165 typedef struct
166 {
167  bool logical; /* True if this is logical replication stream,
168  * false if physical stream. */
169  char *slotname; /* Name of the replication slot or NULL. */
170  XLogRecPtr startpoint; /* LSN of starting point. */
171 
172  union
173  {
174  struct
175  {
176  TimeLineID startpointTLI; /* Starting timeline */
177  } physical;
178  struct
179  {
180  uint32 proto_version; /* Logical protocol version */
181  List *publication_names; /* String list of publications */
182  bool binary; /* Ask publisher to use binary */
183  bool streaming; /* Streaming of large transactions */
184  bool twophase; /* Streaming of two-phase transactions at
185  * prepare time */
186  } logical;
187  } proto;
189 
190 struct WalReceiverConn;
192 
193 /*
194  * Status of walreceiver query execution.
195  *
196  * We only define statuses that are currently used.
197  */
198 typedef enum
199 {
200  WALRCV_ERROR, /* There was error when executing the query. */
201  WALRCV_OK_COMMAND, /* Query executed utility or replication
202  * command. */
203  WALRCV_OK_TUPLES, /* Query returned tuples. */
204  WALRCV_OK_COPY_IN, /* Query started COPY FROM. */
205  WALRCV_OK_COPY_OUT, /* Query started COPY TO. */
206  WALRCV_OK_COPY_BOTH /* Query started COPY BOTH replication
207  * protocol. */
209 
210 /*
211  * Return value for walrcv_exec, returns the status of the execution and
212  * tuples if any.
213  */
214 typedef struct WalRcvExecResult
215 {
217  int sqlstate;
218  char *err;
222 
223 /* WAL receiver - libpqwalreceiver hooks */
224 
225 /*
226  * walrcv_connect_fn
227  *
228  * Establish connection to a cluster. 'logical' is true if the
229  * connection is logical, and false if the connection is physical.
230  * 'appname' is a name associated to the connection, to use for example
231  * with fallback_application_name or application_name. Returns the
232  * details about the connection established, as defined by
233  * WalReceiverConn for each WAL receiver module. On error, NULL is
234  * returned with 'err' including the error generated.
235  */
236 typedef WalReceiverConn *(*walrcv_connect_fn) (const char *conninfo,
237  bool logical,
238  const char *appname,
239  char **err);
240 
241 /*
242  * walrcv_check_conninfo_fn
243  *
244  * Parse and validate the connection string given as of 'conninfo'.
245  */
246 typedef void (*walrcv_check_conninfo_fn) (const char *conninfo);
247 
248 /*
249  * walrcv_get_conninfo_fn
250  *
251  * Returns a user-displayable conninfo string. Note that any
252  * security-sensitive fields should be obfuscated.
253  */
254 typedef char *(*walrcv_get_conninfo_fn) (WalReceiverConn *conn);
255 
256 /*
257  * walrcv_get_senderinfo_fn
258  *
259  * Provide information of the WAL sender this WAL receiver is connected
260  * to, as of 'sender_host' for the host of the sender and 'sender_port'
261  * for its port.
262  */
264  char **sender_host,
265  int *sender_port);
266 
267 /*
268  * walrcv_identify_system_fn
269  *
270  * Run IDENTIFY_SYSTEM on the cluster connected to and validate the
271  * identity of the cluster. Returns the system ID of the cluster
272  * connected to. 'primary_tli' is the timeline ID of the sender.
273  */
274 typedef char *(*walrcv_identify_system_fn) (WalReceiverConn *conn,
275  TimeLineID *primary_tli);
276 
277 /*
278  * walrcv_server_version_fn
279  *
280  * Returns the version number of the cluster connected to.
281  */
283 
284 /*
285  * walrcv_readtimelinehistoryfile_fn
286  *
287  * Fetch from cluster the timeline history file for timeline 'tli'.
288  * Returns the name of the timeline history file as of 'filename', its
289  * contents as of 'content' and its 'size'.
290  */
292  TimeLineID tli,
293  char **filename,
294  char **content,
295  int *size);
296 
297 /*
298  * walrcv_startstreaming_fn
299  *
300  * Start streaming WAL data from given streaming options. Returns true
301  * if the connection has switched successfully to copy-both mode and false
302  * if the server received the command and executed it successfully, but
303  * didn't switch to copy-mode.
304  */
307 
308 /*
309  * walrcv_endstreaming_fn
310  *
311  * Stop streaming of WAL data. Returns the next timeline ID of the cluster
312  * connected to in 'next_tli', or 0 if there was no report.
313  */
315  TimeLineID *next_tli);
316 
317 /*
318  * walrcv_receive_fn
319  *
320  * Receive a message available from the WAL stream. 'buffer' is a pointer
321  * to a buffer holding the message received. Returns the length of the data,
322  * 0 if no data is available yet ('wait_fd' is a socket descriptor which can
323  * be waited on before a retry), and -1 if the cluster ended the COPY.
324  */
326  char **buffer,
327  pgsocket *wait_fd);
328 
329 /*
330  * walrcv_send_fn
331  *
332  * Send a message of size 'nbytes' to the WAL stream with 'buffer' as
333  * contents.
334  */
336  const char *buffer,
337  int nbytes);
338 
339 /*
340  * walrcv_create_slot_fn
341  *
342  * Create a new replication slot named 'slotname'. 'temporary' defines
343  * if the slot is temporary. 'snapshot_action' defines the behavior wanted
344  * for an exported snapshot (see replication protocol for more details).
345  * 'lsn' includes the LSN position at which the created slot became
346  * consistent. Returns the name of the exported snapshot for a logical
347  * slot, or NULL for a physical slot.
348  */
349 typedef char *(*walrcv_create_slot_fn) (WalReceiverConn *conn,
350  const char *slotname,
351  bool temporary,
352  bool two_phase,
353  CRSSnapshotAction snapshot_action,
354  XLogRecPtr *lsn);
355 
356 /*
357  * walrcv_get_backend_pid_fn
358  *
359  * Returns the PID of the remote backend process.
360  */
362 
363 /*
364  * walrcv_exec_fn
365  *
366  * Send generic queries (and commands) to the remote cluster. 'nRetTypes'
367  * is the expected number of returned attributes, and 'retTypes' an array
368  * including their type OIDs. Returns the status of the execution and
369  * tuples if any.
370  */
371 typedef WalRcvExecResult *(*walrcv_exec_fn) (WalReceiverConn *conn,
372  const char *query,
373  const int nRetTypes,
374  const Oid *retTypes);
375 
376 /*
377  * walrcv_disconnect_fn
378  *
379  * Disconnect with the cluster.
380  */
382 
384 {
401 
403 
404 #define walrcv_connect(conninfo, logical, appname, err) \
405  WalReceiverFunctions->walrcv_connect(conninfo, logical, appname, err)
406 #define walrcv_check_conninfo(conninfo) \
407  WalReceiverFunctions->walrcv_check_conninfo(conninfo)
408 #define walrcv_get_conninfo(conn) \
409  WalReceiverFunctions->walrcv_get_conninfo(conn)
410 #define walrcv_get_senderinfo(conn, sender_host, sender_port) \
411  WalReceiverFunctions->walrcv_get_senderinfo(conn, sender_host, sender_port)
412 #define walrcv_identify_system(conn, primary_tli) \
413  WalReceiverFunctions->walrcv_identify_system(conn, primary_tli)
414 #define walrcv_server_version(conn) \
415  WalReceiverFunctions->walrcv_server_version(conn)
416 #define walrcv_readtimelinehistoryfile(conn, tli, filename, content, size) \
417  WalReceiverFunctions->walrcv_readtimelinehistoryfile(conn, tli, filename, content, size)
418 #define walrcv_startstreaming(conn, options) \
419  WalReceiverFunctions->walrcv_startstreaming(conn, options)
420 #define walrcv_endstreaming(conn, next_tli) \
421  WalReceiverFunctions->walrcv_endstreaming(conn, next_tli)
422 #define walrcv_receive(conn, buffer, wait_fd) \
423  WalReceiverFunctions->walrcv_receive(conn, buffer, wait_fd)
424 #define walrcv_send(conn, buffer, nbytes) \
425  WalReceiverFunctions->walrcv_send(conn, buffer, nbytes)
426 #define walrcv_create_slot(conn, slotname, temporary, two_phase, snapshot_action, lsn) \
427  WalReceiverFunctions->walrcv_create_slot(conn, slotname, temporary, two_phase, snapshot_action, lsn)
428 #define walrcv_get_backend_pid(conn) \
429  WalReceiverFunctions->walrcv_get_backend_pid(conn)
430 #define walrcv_exec(conn, exec, nRetTypes, retTypes) \
431  WalReceiverFunctions->walrcv_exec(conn, exec, nRetTypes, retTypes)
432 #define walrcv_disconnect(conn) \
433  WalReceiverFunctions->walrcv_disconnect(conn)
434 
435 static inline void
437 {
438  if (!walres)
439  return;
440 
441  if (walres->err)
442  pfree(walres->err);
443 
444  if (walres->tuplestore)
445  tuplestore_end(walres->tuplestore);
446 
447  if (walres->tupledesc)
448  FreeTupleDesc(walres->tupledesc);
449 
450  pfree(walres);
451 }
452 
453 /* prototypes for functions in walreceiver.c */
454 extern void WalReceiverMain(void) pg_attribute_noreturn();
455 extern void ProcessWalRcvInterrupts(void);
456 
457 /* prototypes for functions in walreceiverfuncs.c */
458 extern Size WalRcvShmemSize(void);
459 extern void WalRcvShmemInit(void);
460 extern void ShutdownWalRcv(void);
461 extern bool WalRcvStreaming(void);
462 extern bool WalRcvRunning(void);
463 extern void RequestXLogStreaming(TimeLineID tli, XLogRecPtr recptr,
464  const char *conninfo, const char *slotname,
465  bool create_temp_slot);
467 extern XLogRecPtr GetWalRcvWriteRecPtr(void);
468 extern int GetReplicationApplyDelay(void);
469 extern int GetReplicationTransferLatency(void);
470 extern void WalRcvForceReply(void);
471 
472 #endif /* _WALRECEIVER_H */
walrcv_identify_system_fn walrcv_identify_system
Definition: walreceiver.h:389
int sender_port
Definition: walreceiver.h:119
int slock_t
Definition: s_lock.h:958
#define pg_attribute_noreturn()
Definition: c.h:179
bool hot_standby_feedback
Definition: walreceiver.c:91
bool WalRcvStreaming(void)
XLogRecPtr GetWalRcvWriteRecPtr(void)
walrcv_server_version_fn walrcv_server_version
Definition: walreceiver.h:390
uint32 TimeLineID
Definition: xlogdefs.h:59
int64 pg_time_t
Definition: pgtime.h:23
void ShutdownWalRcv(void)
slock_t mutex
Definition: walreceiver.h:145
walrcv_startstreaming_fn walrcv_startstreaming
Definition: walreceiver.h:392
PGDLLIMPORT WalReceiverFunctionsType * WalReceiverFunctions
Definition: walreceiver.c:95
int64 TimestampTz
Definition: timestamp.h:39
sig_atomic_t force_reply
Definition: walreceiver.h:160
WalRcvState walRcvState
Definition: walreceiver.h:65
WalReceiverConn *(* walrcv_connect_fn)(const char *conninfo, bool logical, const char *appname, char **err)
Definition: walreceiver.h:236
void(* walrcv_disconnect_fn)(WalReceiverConn *conn)
Definition: walreceiver.h:381
WalRcvExecStatus
Definition: walreceiver.h:198
TimeLineID receivedTLI
Definition: walreceiver.h:86
void ProcessWalRcvInterrupts(void)
Definition: walreceiver.c:147
void RequestXLogStreaming(TimeLineID tli, XLogRecPtr recptr, const char *conninfo, const char *slotname, bool create_temp_slot)
unsigned int Oid
Definition: postgres_ext.h:31
TimestampTz lastMsgReceiptTime
Definition: walreceiver.h:100
TimestampTz lastMsgSendTime
Definition: walreceiver.h:99
pg_time_t startTime
Definition: walreceiver.h:67
int GetReplicationApplyDelay(void)
#define PGDLLIMPORT
Definition: c.h:1317
void WalRcvShmemInit(void)
WalRcvState
Definition: walreceiver.h:45
walrcv_receive_fn walrcv_receive
Definition: walreceiver.h:394
pg_atomic_uint64 writtenUpto
Definition: walreceiver.h:153
#define NAMEDATALEN
WalRcvData * WalRcv
static void walrcv_clear_result(WalRcvExecResult *walres)
Definition: walreceiver.h:436
void pfree(void *pointer)
Definition: mcxt.c:1169
#define NI_MAXHOST
Definition: getaddrinfo.h:88
struct WalRcvExecResult WalRcvExecResult
TupleDesc tupledesc
Definition: walreceiver.h:220
walrcv_get_backend_pid_fn walrcv_get_backend_pid
Definition: walreceiver.h:397
void WalRcvForceReply(void)
Definition: walreceiver.c:1286
PGconn * conn
Definition: streamutil.c:54
pid_t pid
Definition: walreceiver.h:64
#define MAXCONNINFO
Definition: walreceiver.h:37
XLogRecPtr latestChunkStart
Definition: walreceiver.h:94
int wal_receiver_timeout
Definition: walreceiver.c:90
int GetReplicationTransferLatency(void)
Definition: latch.h:110
void(* walrcv_send_fn)(WalReceiverConn *conn, const char *buffer, int nbytes)
Definition: walreceiver.h:335
walrcv_send_fn walrcv_send
Definition: walreceiver.h:395
char *(* walrcv_create_slot_fn)(WalReceiverConn *conn, const char *slotname, bool temporary, bool two_phase, CRSSnapshotAction snapshot_action, XLogRecPtr *lsn)
Definition: walreceiver.h:349
XLogRecPtr startpoint
Definition: walreceiver.h:170
unsigned int uint32
Definition: c.h:441
int pgsocket
Definition: port.h:31
pid_t(* walrcv_get_backend_pid_fn)(WalReceiverConn *conn)
Definition: walreceiver.h:361
TimeLineID receiveStartTLI
Definition: walreceiver.h:76
void(* walrcv_get_senderinfo_fn)(WalReceiverConn *conn, char **sender_host, int *sender_port)
Definition: walreceiver.h:263
Latch * latch
Definition: walreceiver.h:143
static char ** options
WalRcvExecResult *(* walrcv_exec_fn)(WalReceiverConn *conn, const char *query, const int nRetTypes, const Oid *retTypes)
Definition: walreceiver.h:371
void(* walrcv_check_conninfo_fn)(const char *conninfo)
Definition: walreceiver.h:246
static TimeLineID receiveTLI
Definition: xlog.c:200
void(* walrcv_readtimelinehistoryfile_fn)(WalReceiverConn *conn, TimeLineID tli, char **filename, char **content, int *size)
Definition: walreceiver.h:291
int(* walrcv_receive_fn)(WalReceiverConn *conn, char **buffer, pgsocket *wait_fd)
Definition: walreceiver.h:325
bool ready_to_display
Definition: walreceiver.h:134
bool(* walrcv_startstreaming_fn)(WalReceiverConn *conn, const WalRcvStreamOptions *options)
Definition: walreceiver.h:305
walrcv_get_conninfo_fn walrcv_get_conninfo
Definition: walreceiver.h:387
static bool two_phase
walrcv_disconnect_fn walrcv_disconnect
Definition: walreceiver.h:399
TimestampTz latestWalEndTime
Definition: walreceiver.h:106
walrcv_endstreaming_fn walrcv_endstreaming
Definition: walreceiver.h:393
int wal_receiver_status_interval
Definition: walreceiver.c:89
XLogRecPtr latestWalEnd
Definition: walreceiver.h:105
Tuplestorestate * tuplestore
Definition: walreceiver.h:219
walrcv_readtimelinehistoryfile_fn walrcv_readtimelinehistoryfile
Definition: walreceiver.h:391
uint64 XLogRecPtr
Definition: xlogdefs.h:21
WalRcvExecStatus status
Definition: walreceiver.h:216
walrcv_get_senderinfo_fn walrcv_get_senderinfo
Definition: walreceiver.h:388
void WalReceiverMain(void) pg_attribute_noreturn()
Definition: walreceiver.c:168
void(* walrcv_endstreaming_fn)(WalReceiverConn *conn, TimeLineID *next_tli)
Definition: walreceiver.h:314
size_t Size
Definition: c.h:540
Size WalRcvShmemSize(void)
void tuplestore_end(Tuplestorestate *state)
Definition: tuplestore.c:453
walrcv_exec_fn walrcv_exec
Definition: walreceiver.h:398
struct WalReceiverFunctionsType WalReceiverFunctionsType
bool is_temp_slot
Definition: walreceiver.h:131
void FreeTupleDesc(TupleDesc tupdesc)
Definition: tupdesc.c:309
static char * filename
Definition: pg_dumpall.c:92
walrcv_create_slot_fn walrcv_create_slot
Definition: walreceiver.h:396
bool WalRcvRunning(void)
CRSSnapshotAction
Definition: walsender.h:20
TimeLineID startpointTLI
Definition: walreceiver.h:176
walrcv_check_conninfo_fn walrcv_check_conninfo
Definition: walreceiver.h:386
XLogRecPtr receiveStart
Definition: walreceiver.h:75
XLogRecPtr flushedUpto
Definition: walreceiver.h:85
char *(* walrcv_get_conninfo_fn)(WalReceiverConn *conn)
Definition: walreceiver.h:254
int(* walrcv_server_version_fn)(WalReceiverConn *conn)
Definition: walreceiver.h:282
walrcv_connect_fn walrcv_connect
Definition: walreceiver.h:385
Definition: pg_list.h:50
XLogRecPtr GetWalRcvFlushRecPtr(XLogRecPtr *latestChunkStart, TimeLineID *receiveTLI)
ConditionVariable walRcvStoppedCV
Definition: walreceiver.h:66
unsigned char bool
Definition: c.h:391
char *(* walrcv_identify_system_fn)(WalReceiverConn *conn, TimeLineID *primary_tli)
Definition: walreceiver.h:274