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"
18 #include "replication/logicalproto.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 */
26 extern int wal_receiver_status_interval;
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;
63 WalRcvState walRcvState;
64 pg_time_t startTime;
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 */
72 XLogRecPtr receiveStart;
73 TimeLineID receiveStartTLI;
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 */
82 XLogRecPtr receivedUpto;
83 TimeLineID receivedTLI;
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 */
91 XLogRecPtr latestChunkStart;
92
93 /*
94 * Time of send and receive of any message received.
95 */
96 TimestampTz lastMsgSendTime;
97 TimestampTz lastMsgReceiptTime;
98
99 /*
100 * Latest reported end of WAL on the sender
101 */
102 XLogRecPtr latestWalEnd;
103 TimestampTz latestWalEndTime;
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) */
118 bool ready_to_display;
119
120 slock_t mutex; /* locks shared variables shown above */
121
122 /*
123 * force walreceiver reply? This doesn't need to be locked; memory
124 * barriers for ordering are sufficient. But we do need atomic fetch and
125 * store semantics, so use sig_atomic_t.
126 */
127 sig_atomic_t force_reply; /* used as a bool */
128
129 /*
130 * Latch used by startup process to wake up walreceiver after telling it
131 * where to start streaming (after setting receiveStart and
132 * receiveStartTLI), and also to tell it to send apply feedback to the
133 * primary whenever specially marked commit records are applied. This is
134 * normally mapped to procLatch when walreceiver is running.
135 */
136 Latch *latch;
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;
160 } WalRcvStreamOptions;
161
162 struct WalReceiverConn;
163 typedef struct WalReceiverConn 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. */
180 } WalRcvExecStatus;
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 {
188 WalRcvExecStatus status;
189 char *err;
190 Tuplestorestate *tuplestore;
191 TupleDesc tupledesc;
192 } WalRcvExecResult;
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);
203 typedef void (*walrcv_readtimelinehistoryfile_fn) (WalReceiverConn *conn,
204 TimeLineID tli,
205 char **filename,
206 char **content, int *size);
207 typedef bool (*walrcv_startstreaming_fn) (WalReceiverConn *conn,
208 const WalRcvStreamOptions *options);
209 typedef void (*walrcv_endstreaming_fn) (WalReceiverConn *conn,
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);
223 typedef void (*walrcv_disconnect_fn) (WalReceiverConn *conn);
224
225 typedef struct WalReceiverFunctionsType
226 {
227 walrcv_connect_fn walrcv_connect;
228 walrcv_check_conninfo_fn walrcv_check_conninfo;
229 walrcv_get_conninfo_fn walrcv_get_conninfo;
230 walrcv_identify_system_fn walrcv_identify_system;
231 walrcv_readtimelinehistoryfile_fn walrcv_readtimelinehistoryfile;
232 walrcv_startstreaming_fn walrcv_startstreaming;
233 walrcv_endstreaming_fn walrcv_endstreaming;
234 walrcv_receive_fn walrcv_receive;
235 walrcv_send_fn walrcv_send;
236 walrcv_create_slot_fn walrcv_create_slot;
237 walrcv_exec_fn walrcv_exec;
238 walrcv_disconnect_fn walrcv_disconnect;
239 } WalReceiverFunctionsType;
240
241 extern PGDLLIMPORT WalReceiverFunctionsType *WalReceiverFunctions;
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
walrcv_clear_result(WalRcvExecResult * walres)269 walrcv_clear_result(WalRcvExecResult *walres)
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 extern void ProcessWalRcvInterrupts(void);
289
290 /* prototypes for functions in walreceiverfuncs.c */
291 extern Size WalRcvShmemSize(void);
292 extern void WalRcvShmemInit(void);
293 extern void ShutdownWalRcv(void);
294 extern bool WalRcvStreaming(void);
295 extern bool WalRcvRunning(void);
296 extern void RequestXLogStreaming(TimeLineID tli, XLogRecPtr recptr,
297 const char *conninfo, const char *slotname);
298 extern XLogRecPtr GetWalRcvWriteRecPtr(XLogRecPtr *latestChunkStart, TimeLineID *receiveTLI);
299 extern int GetReplicationApplyDelay(void);
300 extern int GetReplicationTransferLatency(void);
301 extern void WalRcvForceReply(void);
302
303 #endif /* _WALRECEIVER_H */
304