Line data Source code
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.
125 : */
126 : bool force_reply;
127 :
128 : /*
129 : * Latch used by startup process to wake up walreceiver after telling it
130 : * where to start streaming (after setting receiveStart and
131 : * receiveStartTLI), and also to tell it to send apply feedback to the
132 : * primary whenever specially marked commit records are applied. This is
133 : * normally mapped to procLatch when walreceiver is running.
134 : */
135 : Latch *latch;
136 : } WalRcvData;
137 :
138 : extern WalRcvData *WalRcv;
139 :
140 : typedef struct
141 : {
142 : bool logical; /* True if this is logical replication stream,
143 : * false if physical stream. */
144 : char *slotname; /* Name of the replication slot or NULL. */
145 : XLogRecPtr startpoint; /* LSN of starting point. */
146 :
147 : union
148 : {
149 : struct
150 : {
151 : TimeLineID startpointTLI; /* Starting timeline */
152 : } physical;
153 : struct
154 : {
155 : uint32 proto_version; /* Logical protocol version */
156 : List *publication_names; /* String list of publications */
157 : } logical;
158 : } proto;
159 : } WalRcvStreamOptions;
160 :
161 : struct WalReceiverConn;
162 : typedef struct WalReceiverConn WalReceiverConn;
163 :
164 : /*
165 : * Status of walreceiver query execution.
166 : *
167 : * We only define statuses that are currently used.
168 : */
169 : typedef enum
170 : {
171 : WALRCV_ERROR, /* There was error when executing the query. */
172 : WALRCV_OK_COMMAND, /* Query executed utility or replication
173 : * command. */
174 : WALRCV_OK_TUPLES, /* Query returned tuples. */
175 : WALRCV_OK_COPY_IN, /* Query started COPY FROM. */
176 : WALRCV_OK_COPY_OUT, /* Query started COPY TO. */
177 : WALRCV_OK_COPY_BOTH /* Query started COPY BOTH replication
178 : * protocol. */
179 : } WalRcvExecStatus;
180 :
181 : /*
182 : * Return value for walrcv_query, returns the status of the execution and
183 : * tuples if any.
184 : */
185 : typedef struct WalRcvExecResult
186 : {
187 : WalRcvExecStatus status;
188 : char *err;
189 : Tuplestorestate *tuplestore;
190 : TupleDesc tupledesc;
191 : } WalRcvExecResult;
192 :
193 : /* libpqwalreceiver hooks */
194 : typedef WalReceiverConn *(*walrcv_connect_fn) (const char *conninfo, bool logical,
195 : const char *appname,
196 : char **err);
197 : typedef void (*walrcv_check_conninfo_fn) (const char *conninfo);
198 : typedef char *(*walrcv_get_conninfo_fn) (WalReceiverConn *conn);
199 : typedef char *(*walrcv_identify_system_fn) (WalReceiverConn *conn,
200 : TimeLineID *primary_tli,
201 : int *server_version);
202 : typedef void (*walrcv_readtimelinehistoryfile_fn) (WalReceiverConn *conn,
203 : TimeLineID tli,
204 : char **filename,
205 : char **content, int *size);
206 : typedef bool (*walrcv_startstreaming_fn) (WalReceiverConn *conn,
207 : const WalRcvStreamOptions *options);
208 : typedef void (*walrcv_endstreaming_fn) (WalReceiverConn *conn,
209 : TimeLineID *next_tli);
210 : typedef int (*walrcv_receive_fn) (WalReceiverConn *conn, char **buffer,
211 : pgsocket *wait_fd);
212 : typedef void (*walrcv_send_fn) (WalReceiverConn *conn, const char *buffer,
213 : int nbytes);
214 : typedef char *(*walrcv_create_slot_fn) (WalReceiverConn *conn,
215 : const char *slotname, bool temporary,
216 : CRSSnapshotAction snapshot_action,
217 : XLogRecPtr *lsn);
218 : typedef WalRcvExecResult *(*walrcv_exec_fn) (WalReceiverConn *conn,
219 : const char *query,
220 : const int nRetTypes,
221 : const Oid *retTypes);
222 : typedef void (*walrcv_disconnect_fn) (WalReceiverConn *conn);
223 :
224 : typedef struct WalReceiverFunctionsType
225 : {
226 : walrcv_connect_fn walrcv_connect;
227 : walrcv_check_conninfo_fn walrcv_check_conninfo;
228 : walrcv_get_conninfo_fn walrcv_get_conninfo;
229 : walrcv_identify_system_fn walrcv_identify_system;
230 : walrcv_readtimelinehistoryfile_fn walrcv_readtimelinehistoryfile;
231 : walrcv_startstreaming_fn walrcv_startstreaming;
232 : walrcv_endstreaming_fn walrcv_endstreaming;
233 : walrcv_receive_fn walrcv_receive;
234 : walrcv_send_fn walrcv_send;
235 : walrcv_create_slot_fn walrcv_create_slot;
236 : walrcv_exec_fn walrcv_exec;
237 : walrcv_disconnect_fn walrcv_disconnect;
238 : } WalReceiverFunctionsType;
239 :
240 : extern PGDLLIMPORT WalReceiverFunctionsType *WalReceiverFunctions;
241 :
242 : #define walrcv_connect(conninfo, logical, appname, err) \
243 : WalReceiverFunctions->walrcv_connect(conninfo, logical, appname, err)
244 : #define walrcv_check_conninfo(conninfo) \
245 : WalReceiverFunctions->walrcv_check_conninfo(conninfo)
246 : #define walrcv_get_conninfo(conn) \
247 : WalReceiverFunctions->walrcv_get_conninfo(conn)
248 : #define walrcv_identify_system(conn, primary_tli, server_version) \
249 : WalReceiverFunctions->walrcv_identify_system(conn, primary_tli, server_version)
250 : #define walrcv_readtimelinehistoryfile(conn, tli, filename, content, size) \
251 : WalReceiverFunctions->walrcv_readtimelinehistoryfile(conn, tli, filename, content, size)
252 : #define walrcv_startstreaming(conn, options) \
253 : WalReceiverFunctions->walrcv_startstreaming(conn, options)
254 : #define walrcv_endstreaming(conn, next_tli) \
255 : WalReceiverFunctions->walrcv_endstreaming(conn, next_tli)
256 : #define walrcv_receive(conn, buffer, wait_fd) \
257 : WalReceiverFunctions->walrcv_receive(conn, buffer, wait_fd)
258 : #define walrcv_send(conn, buffer, nbytes) \
259 : WalReceiverFunctions->walrcv_send(conn, buffer, nbytes)
260 : #define walrcv_create_slot(conn, slotname, temporary, snapshot_action, lsn) \
261 : WalReceiverFunctions->walrcv_create_slot(conn, slotname, temporary, snapshot_action, lsn)
262 : #define walrcv_exec(conn, exec, nRetTypes, retTypes) \
263 : WalReceiverFunctions->walrcv_exec(conn, exec, nRetTypes, retTypes)
264 : #define walrcv_disconnect(conn) \
265 : WalReceiverFunctions->walrcv_disconnect(conn)
266 :
267 : static inline void
268 0 : walrcv_clear_result(WalRcvExecResult *walres)
269 : {
270 0 : if (!walres)
271 0 : return;
272 :
273 0 : if (walres->err)
274 0 : pfree(walres->err);
275 :
276 0 : if (walres->tuplestore)
277 0 : tuplestore_end(walres->tuplestore);
278 :
279 0 : if (walres->tupledesc)
280 0 : FreeTupleDesc(walres->tupledesc);
281 :
282 0 : pfree(walres);
283 : }
284 :
285 : /* prototypes for functions in walreceiver.c */
286 : extern void WalReceiverMain(void) pg_attribute_noreturn();
287 :
288 : /* prototypes for functions in walreceiverfuncs.c */
289 : extern Size WalRcvShmemSize(void);
290 : extern void WalRcvShmemInit(void);
291 : extern void ShutdownWalRcv(void);
292 : extern bool WalRcvStreaming(void);
293 : extern bool WalRcvRunning(void);
294 : extern void RequestXLogStreaming(TimeLineID tli, XLogRecPtr recptr,
295 : const char *conninfo, const char *slotname);
296 : extern XLogRecPtr GetWalRcvWriteRecPtr(XLogRecPtr *latestChunkStart, TimeLineID *receiveTLI);
297 : extern int GetReplicationApplyDelay(void);
298 : extern int GetReplicationTransferLatency(void);
299 : extern void WalRcvForceReply(void);
300 :
301 : #endif /* _WALRECEIVER_H */
|
