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-2020, 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"
20 #include "replication/walsender.h"
21 #include "storage/latch.h"
22 #include "storage/spin.h"
23 #include "utils/tuplestore.h"
24 
25 /* user-settable parameters */
28 extern int wal_receiver_timeout;
29 extern bool hot_standby_feedback;
30 
31 /*
32  * MAXCONNINFO: maximum size of a connection string.
33  *
34  * XXX: Should this move to pg_config_manual.h?
35  */
36 #define MAXCONNINFO 1024
37 
38 /* Can we allow the standby to accept replication connection from another standby? */
39 #define AllowCascadeReplication() (EnableHotStandby && max_wal_senders > 0)
40 
41 /*
42  * Values for WalRcv->walRcvState.
43  */
44 typedef enum
45 {
46  WALRCV_STOPPED, /* stopped and mustn't start up again */
47  WALRCV_STARTING, /* launched, but the process hasn't
48  * initialized yet */
49  WALRCV_STREAMING, /* walreceiver is streaming */
50  WALRCV_WAITING, /* stopped streaming, waiting for orders */
51  WALRCV_RESTARTING, /* asked to restart streaming */
52  WALRCV_STOPPING /* requested to stop, but still running */
53 } WalRcvState;
54 
55 /* Shared memory area for management of walreceiver process */
56 typedef struct
57 {
58  /*
59  * PID of currently active walreceiver process, its current state and
60  * start time (actually, the time at which it was requested to be
61  * started).
62  */
63  pid_t pid;
66 
67  /*
68  * receiveStart and receiveStartTLI indicate the first byte position and
69  * timeline that will be received. When startup process starts the
70  * walreceiver, it sets these to the point where it wants the streaming to
71  * begin.
72  */
75 
76  /*
77  * receivedUpto-1 is the last byte position that has already been
78  * received, and receivedTLI is the timeline it came from. At the first
79  * startup of walreceiver, these are set to receiveStart and
80  * receiveStartTLI. After that, walreceiver updates these whenever it
81  * flushes the received WAL to disk.
82  */
85 
86  /*
87  * latestChunkStart is the starting byte position of the current "batch"
88  * of received WAL. It's actually the same as the previous value of
89  * receivedUpto before the last flush to disk. Startup process can use
90  * this to detect whether it's keeping up or not.
91  */
93 
94  /*
95  * Time of send and receive of any message received.
96  */
99 
100  /*
101  * Latest reported end of WAL on the sender
102  */
105 
106  /*
107  * connection string; initially set to connect to the primary, and later
108  * clobbered to hide security-sensitive fields.
109  */
110  char conninfo[MAXCONNINFO];
111 
112  /*
113  * Host name (this can be a host name, an IP address, or a directory path)
114  * and port number of the active replication connection.
115  */
116  char sender_host[NI_MAXHOST];
118 
119  /*
120  * replication slot name; is also used for walreceiver to connect with the
121  * primary
122  */
123  char slotname[NAMEDATALEN];
124 
125  /*
126  * If it's a temporary replication slot, it needs to be recreated when
127  * connecting.
128  */
130 
131  /* set true once conninfo is ready to display (obfuscated pwds etc) */
133 
134  /*
135  * Latch used by startup process to wake up walreceiver after telling it
136  * where to start streaming (after setting receiveStart and
137  * receiveStartTLI), and also to tell it to send apply feedback to the
138  * primary whenever specially marked commit records are applied. This is
139  * normally mapped to procLatch when walreceiver is running.
140  */
142 
143  slock_t mutex; /* locks shared variables shown above */
144 
145  /*
146  * force walreceiver reply? This doesn't need to be locked; memory
147  * barriers for ordering are sufficient. But we do need atomic fetch and
148  * store semantics, so use sig_atomic_t.
149  */
150  sig_atomic_t force_reply; /* used as a bool */
151 } WalRcvData;
152 
153 extern WalRcvData *WalRcv;
154 
155 typedef struct
156 {
157  bool logical; /* True if this is logical replication stream,
158  * false if physical stream. */
159  char *slotname; /* Name of the replication slot or NULL. */
160  XLogRecPtr startpoint; /* LSN of starting point. */
161 
162  union
163  {
164  struct
165  {
166  TimeLineID startpointTLI; /* Starting timeline */
167  } physical;
168  struct
169  {
170  uint32 proto_version; /* Logical protocol version */
171  List *publication_names; /* String list of publications */
172  } logical;
173  } proto;
175 
176 struct WalReceiverConn;
178 
179 /*
180  * Status of walreceiver query execution.
181  *
182  * We only define statuses that are currently used.
183  */
184 typedef enum
185 {
186  WALRCV_ERROR, /* There was error when executing the query. */
187  WALRCV_OK_COMMAND, /* Query executed utility or replication
188  * command. */
189  WALRCV_OK_TUPLES, /* Query returned tuples. */
190  WALRCV_OK_COPY_IN, /* Query started COPY FROM. */
191  WALRCV_OK_COPY_OUT, /* Query started COPY TO. */
192  WALRCV_OK_COPY_BOTH /* Query started COPY BOTH replication
193  * protocol. */
195 
196 /*
197  * Return value for walrcv_exec, returns the status of the execution and
198  * tuples if any.
199  */
200 typedef struct WalRcvExecResult
201 {
203  char *err;
207 
208 /* libpqwalreceiver hooks */
209 typedef WalReceiverConn *(*walrcv_connect_fn) (const char *conninfo, bool logical,
210  const char *appname,
211  char **err);
212 typedef void (*walrcv_check_conninfo_fn) (const char *conninfo);
213 typedef char *(*walrcv_get_conninfo_fn) (WalReceiverConn *conn);
215  char **sender_host,
216  int *sender_port);
217 typedef char *(*walrcv_identify_system_fn) (WalReceiverConn *conn,
218  TimeLineID *primary_tli);
221  TimeLineID tli,
222  char **filename,
223  char **content, int *size);
227  TimeLineID *next_tli);
228 typedef int (*walrcv_receive_fn) (WalReceiverConn *conn, char **buffer,
229  pgsocket *wait_fd);
230 typedef void (*walrcv_send_fn) (WalReceiverConn *conn, const char *buffer,
231  int nbytes);
232 typedef char *(*walrcv_create_slot_fn) (WalReceiverConn *conn,
233  const char *slotname, bool temporary,
234  CRSSnapshotAction snapshot_action,
235  XLogRecPtr *lsn);
237 typedef WalRcvExecResult *(*walrcv_exec_fn) (WalReceiverConn *conn,
238  const char *query,
239  const int nRetTypes,
240  const Oid *retTypes);
242 
244 {
261 
263 
264 #define walrcv_connect(conninfo, logical, appname, err) \
265  WalReceiverFunctions->walrcv_connect(conninfo, logical, appname, err)
266 #define walrcv_check_conninfo(conninfo) \
267  WalReceiverFunctions->walrcv_check_conninfo(conninfo)
268 #define walrcv_get_conninfo(conn) \
269  WalReceiverFunctions->walrcv_get_conninfo(conn)
270 #define walrcv_get_senderinfo(conn, sender_host, sender_port) \
271  WalReceiverFunctions->walrcv_get_senderinfo(conn, sender_host, sender_port)
272 #define walrcv_identify_system(conn, primary_tli) \
273  WalReceiverFunctions->walrcv_identify_system(conn, primary_tli)
274 #define walrcv_server_version(conn) \
275  WalReceiverFunctions->walrcv_server_version(conn)
276 #define walrcv_readtimelinehistoryfile(conn, tli, filename, content, size) \
277  WalReceiverFunctions->walrcv_readtimelinehistoryfile(conn, tli, filename, content, size)
278 #define walrcv_startstreaming(conn, options) \
279  WalReceiverFunctions->walrcv_startstreaming(conn, options)
280 #define walrcv_endstreaming(conn, next_tli) \
281  WalReceiverFunctions->walrcv_endstreaming(conn, next_tli)
282 #define walrcv_receive(conn, buffer, wait_fd) \
283  WalReceiverFunctions->walrcv_receive(conn, buffer, wait_fd)
284 #define walrcv_send(conn, buffer, nbytes) \
285  WalReceiverFunctions->walrcv_send(conn, buffer, nbytes)
286 #define walrcv_create_slot(conn, slotname, temporary, snapshot_action, lsn) \
287  WalReceiverFunctions->walrcv_create_slot(conn, slotname, temporary, snapshot_action, lsn)
288 #define walrcv_get_backend_pid(conn) \
289  WalReceiverFunctions->walrcv_get_backend_pid(conn)
290 #define walrcv_exec(conn, exec, nRetTypes, retTypes) \
291  WalReceiverFunctions->walrcv_exec(conn, exec, nRetTypes, retTypes)
292 #define walrcv_disconnect(conn) \
293  WalReceiverFunctions->walrcv_disconnect(conn)
294 
295 static inline void
297 {
298  if (!walres)
299  return;
300 
301  if (walres->err)
302  pfree(walres->err);
303 
304  if (walres->tuplestore)
305  tuplestore_end(walres->tuplestore);
306 
307  if (walres->tupledesc)
308  FreeTupleDesc(walres->tupledesc);
309 
310  pfree(walres);
311 }
312 
313 /* prototypes for functions in walreceiver.c */
314 extern void WalReceiverMain(void) pg_attribute_noreturn();
315 extern void ProcessWalRcvInterrupts(void);
316 
317 /* prototypes for functions in walreceiverfuncs.c */
318 extern Size WalRcvShmemSize(void);
319 extern void WalRcvShmemInit(void);
320 extern void ShutdownWalRcv(void);
321 extern bool WalRcvStreaming(void);
322 extern bool WalRcvRunning(void);
323 extern void RequestXLogStreaming(TimeLineID tli, XLogRecPtr recptr,
324  const char *conninfo, const char *slotname);
326 extern int GetReplicationApplyDelay(void);
327 extern int GetReplicationTransferLatency(void);
328 extern void WalRcvForceReply(void);
329 
330 #endif /* _WALRECEIVER_H */
walrcv_identify_system_fn walrcv_identify_system
Definition: walreceiver.h:249
int sender_port
Definition: walreceiver.h:117
int slock_t
Definition: s_lock.h:934
XLogRecPtr GetWalRcvWriteRecPtr(XLogRecPtr *latestChunkStart, TimeLineID *receiveTLI)
bool hot_standby_feedback
Definition: walreceiver.c:79
bool WalRcvStreaming(void)
walrcv_server_version_fn walrcv_server_version
Definition: walreceiver.h:250
uint32 TimeLineID
Definition: xlogdefs.h:52
int64 pg_time_t
Definition: pgtime.h:23
void ShutdownWalRcv(void)
slock_t mutex
Definition: walreceiver.h:143
walrcv_startstreaming_fn walrcv_startstreaming
Definition: walreceiver.h:252
PGDLLIMPORT WalReceiverFunctionsType * WalReceiverFunctions
Definition: walreceiver.c:83
int64 TimestampTz
Definition: timestamp.h:39
sig_atomic_t force_reply
Definition: walreceiver.h:150
WalRcvState walRcvState
Definition: walreceiver.h:64
WalReceiverConn *(* walrcv_connect_fn)(const char *conninfo, bool logical, const char *appname, char **err)
Definition: walreceiver.h:209
void(* walrcv_disconnect_fn)(WalReceiverConn *conn)
Definition: walreceiver.h:241
WalRcvExecStatus
Definition: walreceiver.h:184
TimeLineID receivedTLI
Definition: walreceiver.h:84
void ProcessWalRcvInterrupts(void)
Definition: walreceiver.c:146
unsigned int Oid
Definition: postgres_ext.h:31
TimestampTz lastMsgReceiptTime
Definition: walreceiver.h:98
TimestampTz lastMsgSendTime
Definition: walreceiver.h:97
pg_time_t startTime
Definition: walreceiver.h:65
int GetReplicationApplyDelay(void)
#define PGDLLIMPORT
Definition: c.h:1276
void WalRcvShmemInit(void)
WalRcvState
Definition: walreceiver.h:44
walrcv_receive_fn walrcv_receive
Definition: walreceiver.h:254
#define NAMEDATALEN
WalRcvData * WalRcv
static void walrcv_clear_result(WalRcvExecResult *walres)
Definition: walreceiver.h:296
void pfree(void *pointer)
Definition: mcxt.c:1056
#define NI_MAXHOST
Definition: getaddrinfo.h:88
struct WalRcvExecResult WalRcvExecResult
TupleDesc tupledesc
Definition: walreceiver.h:205
walrcv_get_backend_pid_fn walrcv_get_backend_pid
Definition: walreceiver.h:257
void WalRcvForceReply(void)
Definition: walreceiver.c:1294
PGconn * conn
Definition: streamutil.c:54
pid_t pid
Definition: walreceiver.h:63
#define MAXCONNINFO
Definition: walreceiver.h:36
XLogRecPtr latestChunkStart
Definition: walreceiver.h:92
int wal_receiver_timeout
Definition: walreceiver.c:78
int GetReplicationTransferLatency(void)
Definition: latch.h:110
void(* walrcv_send_fn)(WalReceiverConn *conn, const char *buffer, int nbytes)
Definition: walreceiver.h:230
walrcv_send_fn walrcv_send
Definition: walreceiver.h:255
char *(* walrcv_create_slot_fn)(WalReceiverConn *conn, const char *slotname, bool temporary, CRSSnapshotAction snapshot_action, XLogRecPtr *lsn)
Definition: walreceiver.h:232
XLogRecPtr startpoint
Definition: walreceiver.h:160
unsigned int uint32
Definition: c.h:367
int pgsocket
Definition: port.h:31
pid_t(* walrcv_get_backend_pid_fn)(WalReceiverConn *conn)
Definition: walreceiver.h:236
XLogRecPtr receivedUpto
Definition: walreceiver.h:83
bool wal_receiver_create_temp_slot
Definition: walreceiver.c:76
TimeLineID receiveStartTLI
Definition: walreceiver.h:74
void(* walrcv_get_senderinfo_fn)(WalReceiverConn *conn, char **sender_host, int *sender_port)
Definition: walreceiver.h:214
Latch * latch
Definition: walreceiver.h:141
static char ** options
WalRcvExecResult *(* walrcv_exec_fn)(WalReceiverConn *conn, const char *query, const int nRetTypes, const Oid *retTypes)
Definition: walreceiver.h:237
void(* walrcv_check_conninfo_fn)(const char *conninfo)
Definition: walreceiver.h:212
static TimeLineID receiveTLI
Definition: xlog.c:209
void(* walrcv_readtimelinehistoryfile_fn)(WalReceiverConn *conn, TimeLineID tli, char **filename, char **content, int *size)
Definition: walreceiver.h:220
int(* walrcv_receive_fn)(WalReceiverConn *conn, char **buffer, pgsocket *wait_fd)
Definition: walreceiver.h:228
bool ready_to_display
Definition: walreceiver.h:132
bool(* walrcv_startstreaming_fn)(WalReceiverConn *conn, const WalRcvStreamOptions *options)
Definition: walreceiver.h:224
walrcv_get_conninfo_fn walrcv_get_conninfo
Definition: walreceiver.h:247
walrcv_disconnect_fn walrcv_disconnect
Definition: walreceiver.h:259
TimestampTz latestWalEndTime
Definition: walreceiver.h:104
walrcv_endstreaming_fn walrcv_endstreaming
Definition: walreceiver.h:253
int wal_receiver_status_interval
Definition: walreceiver.c:77
XLogRecPtr latestWalEnd
Definition: walreceiver.h:103
void RequestXLogStreaming(TimeLineID tli, XLogRecPtr recptr, const char *conninfo, const char *slotname)
Tuplestorestate * tuplestore
Definition: walreceiver.h:204
walrcv_readtimelinehistoryfile_fn walrcv_readtimelinehistoryfile
Definition: walreceiver.h:251
uint64 XLogRecPtr
Definition: xlogdefs.h:21
WalRcvExecStatus status
Definition: walreceiver.h:202
walrcv_get_senderinfo_fn walrcv_get_senderinfo
Definition: walreceiver.h:248
void WalReceiverMain(void) pg_attribute_noreturn()
Definition: walreceiver.c:167
void(* walrcv_endstreaming_fn)(WalReceiverConn *conn, TimeLineID *next_tli)
Definition: walreceiver.h:226
size_t Size
Definition: c.h:466
Size WalRcvShmemSize(void)
void tuplestore_end(Tuplestorestate *state)
Definition: tuplestore.c:453
walrcv_exec_fn walrcv_exec
Definition: walreceiver.h:258
struct WalReceiverFunctionsType WalReceiverFunctionsType
bool is_temp_slot
Definition: walreceiver.h:129
void FreeTupleDesc(TupleDesc tupdesc)
Definition: tupdesc.c:313
#define pg_attribute_noreturn()
Definition: c.h:145
static char * filename
Definition: pg_dumpall.c:90
walrcv_create_slot_fn walrcv_create_slot
Definition: walreceiver.h:256
bool WalRcvRunning(void)
CRSSnapshotAction
Definition: walsender.h:20
TimeLineID startpointTLI
Definition: walreceiver.h:166
walrcv_check_conninfo_fn walrcv_check_conninfo
Definition: walreceiver.h:246
XLogRecPtr receiveStart
Definition: walreceiver.h:73
char *(* walrcv_get_conninfo_fn)(WalReceiverConn *conn)
Definition: walreceiver.h:213
int(* walrcv_server_version_fn)(WalReceiverConn *conn)
Definition: walreceiver.h:219
walrcv_connect_fn walrcv_connect
Definition: walreceiver.h:245
Definition: pg_list.h:50
unsigned char bool
Definition: c.h:317
char *(* walrcv_identify_system_fn)(WalReceiverConn *conn, TimeLineID *primary_tli)
Definition: walreceiver.h:217