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