Age Owner Branch data TLA Line data Source code
1 : : /*-------------------------------------------------------------------------
2 : : *
3 : : * receivelog.c - receive WAL files using the streaming
4 : : * replication protocol.
5 : : *
6 : : * Author: Magnus Hagander <magnus@hagander.net>
7 : : *
8 : : * Portions Copyright (c) 1996-2026, PostgreSQL Global Development Group
9 : : *
10 : : * IDENTIFICATION
11 : : * src/bin/pg_basebackup/receivelog.c
12 : : *-------------------------------------------------------------------------
13 : : */
14 : :
15 : : #include "postgres_fe.h"
16 : :
17 : : #include <sys/select.h>
18 : : #include <sys/stat.h>
19 : : #include <unistd.h>
20 : :
21 : : #include "access/xlog_internal.h"
22 : : #include "common/logging.h"
23 : : #include "common/pg_parse_lsn.h"
24 : : #include "libpq-fe.h"
25 : : #include "libpq/protocol.h"
26 : : #include "receivelog.h"
27 : : #include "streamutil.h"
28 : :
29 : : /* currently open WAL file */
30 : : static Walfile *walfile = NULL;
31 : : static bool reportFlushPosition = false;
32 : : static XLogRecPtr lastFlushPosition = InvalidXLogRecPtr;
33 : :
34 : : static bool still_sending = true; /* feedback still needs to be sent? */
35 : :
36 : : static PGresult *HandleCopyStream(PGconn *conn, StreamCtl *stream,
37 : : XLogRecPtr *stoppos);
38 : : static int CopyStreamPoll(PGconn *conn, long timeout_ms, pgsocket stop_socket);
39 : : static int CopyStreamReceive(PGconn *conn, long timeout, pgsocket stop_socket,
40 : : char **buffer);
41 : : static bool ProcessKeepaliveMsg(PGconn *conn, StreamCtl *stream, char *copybuf,
42 : : int len, XLogRecPtr blockpos, TimestampTz *last_status);
43 : : static bool ProcessWALDataMsg(PGconn *conn, StreamCtl *stream, char *copybuf, int len,
44 : : XLogRecPtr *blockpos);
45 : : static PGresult *HandleEndOfCopyStream(PGconn *conn, StreamCtl *stream, char *copybuf,
46 : : XLogRecPtr blockpos, XLogRecPtr *stoppos);
47 : : static bool CheckCopyStreamStop(PGconn *conn, StreamCtl *stream, XLogRecPtr blockpos);
48 : : static long CalculateCopyStreamSleeptime(TimestampTz now, int standby_message_timeout,
49 : : TimestampTz last_status);
50 : :
51 : : static bool ReadEndOfStreamingResult(PGresult *res, XLogRecPtr *startpos,
52 : : uint32 *timeline);
53 : :
54 : : static bool
3595 magnus@hagander.net 55 :CBC 12 : mark_file_as_archived(StreamCtl *stream, const char *fname)
56 : : {
57 : : Walfile *f;
58 : : static char tmppath[MAXPGPATH];
59 : :
60 : 12 : snprintf(tmppath, sizeof(tmppath), "archive_status/%s.done",
61 : : fname);
62 : :
1438 rhaas@postgresql.org 63 : 12 : f = stream->walmethod->ops->open_for_write(stream->walmethod, tmppath,
64 : : NULL, 0);
3595 magnus@hagander.net 65 [ - + ]: 12 : if (f == NULL)
66 : : {
2705 peter@eisentraut.org 67 :UBC 0 : pg_log_error("could not create archive status file \"%s\": %s",
68 : : tmppath, GetLastWalMethodError(stream->walmethod));
4254 andres@anarazel.de 69 : 0 : return false;
70 : : }
71 : :
1438 rhaas@postgresql.org 72 [ - + ]:CBC 12 : if (stream->walmethod->ops->close(f, CLOSE_NORMAL) != 0)
73 : : {
1744 tgl@sss.pgh.pa.us 74 :UBC 0 : pg_log_error("could not close archive status file \"%s\": %s",
75 : : tmppath, GetLastWalMethodError(stream->walmethod));
76 : 0 : return false;
77 : : }
78 : :
4254 andres@anarazel.de 79 :CBC 12 : return true;
80 : : }
81 : :
82 : : /*
83 : : * Open a new WAL file in the specified directory.
84 : : *
85 : : * Returns true if OK; on failure, returns false after printing an error msg.
86 : : * On success, 'walfile' is set to the opened WAL file.
87 : : *
88 : : * The file will be padded to 16Mb with zeroes.
89 : : */
90 : : static bool
3821 magnus@hagander.net 91 : 171 : open_walfile(StreamCtl *stream, XLogRecPtr startpoint)
92 : : {
93 : : Walfile *f;
94 : : char *fn;
95 : : ssize_t size;
96 : : XLogSegNo segno;
97 : : char walfile_name[MAXPGPATH];
98 : :
3264 andres@anarazel.de 99 : 171 : XLByteToSeg(startpoint, segno, WalSegSz);
1438 rhaas@postgresql.org 100 : 171 : XLogFileName(walfile_name, stream->timeline, segno, WalSegSz);
101 : :
102 : : /* Note that this considers the compression used if necessary */
103 : 171 : fn = stream->walmethod->ops->get_file_name(stream->walmethod,
104 : : walfile_name,
105 : 171 : stream->partial_suffix);
106 : :
107 : : /*
108 : : * When streaming to files, if an existing file exists we verify that it's
109 : : * either empty (just created), or a complete WalSegSz segment (in which
110 : : * case it has been created and padded). Anything else indicates a corrupt
111 : : * file. Compressed files have no need for padding, so just ignore this
112 : : * case.
113 : : *
114 : : * When streaming to tar, no file with this name will exist before, so we
115 : : * never have to verify a size.
116 : : */
117 [ + + - + ]: 335 : if (stream->walmethod->compression_algorithm == PG_COMPRESSION_NONE &&
118 : 164 : stream->walmethod->ops->existsfile(stream->walmethod, fn))
119 : : {
1438 rhaas@postgresql.org 120 :UBC 0 : size = stream->walmethod->ops->get_file_size(stream->walmethod, fn);
3595 magnus@hagander.net 121 [ # # ]: 0 : if (size < 0)
122 : : {
2705 peter@eisentraut.org 123 : 0 : pg_log_error("could not get size of write-ahead log file \"%s\": %s",
124 : : fn, GetLastWalMethodError(stream->walmethod));
1858 michael@paquier.xyz 125 : 0 : pg_free(fn);
3595 magnus@hagander.net 126 : 0 : return false;
127 : : }
3264 andres@anarazel.de 128 [ # # ]: 0 : if (size == WalSegSz)
129 : : {
130 : : /* Already padded file. Open it for use */
1438 rhaas@postgresql.org 131 : 0 : f = stream->walmethod->ops->open_for_write(stream->walmethod, walfile_name, stream->partial_suffix, 0);
3595 magnus@hagander.net 132 [ # # ]: 0 : if (f == NULL)
133 : : {
2705 peter@eisentraut.org 134 : 0 : pg_log_error("could not open existing write-ahead log file \"%s\": %s",
135 : : fn, GetLastWalMethodError(stream->walmethod));
1858 michael@paquier.xyz 136 : 0 : pg_free(fn);
3616 tgl@sss.pgh.pa.us 137 : 0 : return false;
138 : : }
139 : :
140 : : /* fsync file in case of a previous crash */
1438 rhaas@postgresql.org 141 [ # # ]: 0 : if (stream->walmethod->ops->sync(f) != 0)
142 : : {
1602 tgl@sss.pgh.pa.us 143 : 0 : pg_log_error("could not fsync existing write-ahead log file \"%s\": %s",
144 : : fn, GetLastWalMethodError(stream->walmethod));
1438 rhaas@postgresql.org 145 : 0 : stream->walmethod->ops->close(f, CLOSE_UNLINK);
2586 peter@eisentraut.org 146 : 0 : exit(1);
147 : : }
148 : :
3595 magnus@hagander.net 149 : 0 : walfile = f;
1858 michael@paquier.xyz 150 : 0 : pg_free(fn);
3595 magnus@hagander.net 151 : 0 : return true;
152 : : }
153 [ # # ]: 0 : if (size != 0)
154 : : {
155 : : /* if write didn't set errno, assume problem is no disk space */
3616 tgl@sss.pgh.pa.us 156 [ # # ]: 0 : if (errno == 0)
157 : 0 : errno = ENOSPC;
1845 peter@eisentraut.org 158 : 0 : pg_log_error(ngettext("write-ahead log file \"%s\" has %zd byte, should be 0 or %d",
159 : : "write-ahead log file \"%s\" has %zd bytes, should be 0 or %d",
160 : : size),
161 : : fn, size, WalSegSz);
1858 michael@paquier.xyz 162 : 0 : pg_free(fn);
4970 heikki.linnakangas@i 163 : 0 : return false;
164 : : }
165 : : /* File existed and was empty, so fall through and open */
166 : : }
167 : :
168 : : /* No file existed, so create one */
169 : :
1438 rhaas@postgresql.org 170 :CBC 171 : f = stream->walmethod->ops->open_for_write(stream->walmethod,
171 : : walfile_name,
172 : 171 : stream->partial_suffix,
173 : : WalSegSz);
3595 magnus@hagander.net 174 [ - + ]: 171 : if (f == NULL)
175 : : {
2705 peter@eisentraut.org 176 :UBC 0 : pg_log_error("could not open write-ahead log file \"%s\": %s",
177 : : fn, GetLastWalMethodError(stream->walmethod));
1858 michael@paquier.xyz 178 : 0 : pg_free(fn);
4970 heikki.linnakangas@i 179 : 0 : return false;
180 : : }
181 : :
1858 michael@paquier.xyz 182 :CBC 171 : pg_free(fn);
4970 heikki.linnakangas@i 183 : 171 : walfile = f;
184 : 171 : return true;
185 : : }
186 : :
187 : : /*
188 : : * Close the current WAL file (if open), and rename it to the correct
189 : : * filename if it's complete. On failure, prints an error message to stderr
190 : : * and returns false, otherwise returns true.
191 : : */
192 : : static bool
3821 magnus@hagander.net 193 : 178 : close_walfile(StreamCtl *stream, XLogRecPtr pos)
194 : : {
195 : : char *fn;
196 : : pgoff_t currpos;
197 : : int r;
198 : : char walfile_name[MAXPGPATH];
199 : :
3595 200 [ + + ]: 178 : if (walfile == NULL)
4970 heikki.linnakangas@i 201 : 7 : return true;
202 : :
1438 rhaas@postgresql.org 203 : 171 : strlcpy(walfile_name, walfile->pathname, MAXPGPATH);
204 : 171 : currpos = walfile->currpos;
205 : :
206 : : /* Note that this considers the compression used if necessary */
207 : 171 : fn = stream->walmethod->ops->get_file_name(stream->walmethod,
208 : : walfile_name,
209 : 171 : stream->partial_suffix);
210 : :
3595 magnus@hagander.net 211 [ + + ]: 171 : if (stream->partial_suffix)
212 : : {
3264 andres@anarazel.de 213 [ + + ]: 12 : if (currpos == WalSegSz)
1438 rhaas@postgresql.org 214 : 6 : r = stream->walmethod->ops->close(walfile, CLOSE_NORMAL);
215 : : else
216 : : {
1805 michael@paquier.xyz 217 : 6 : pg_log_info("not renaming \"%s\", segment is not complete", fn);
1438 rhaas@postgresql.org 218 : 6 : r = stream->walmethod->ops->close(walfile, CLOSE_NO_RENAME);
219 : : }
220 : : }
221 : : else
222 : 159 : r = stream->walmethod->ops->close(walfile, CLOSE_NORMAL);
223 : :
3595 magnus@hagander.net 224 : 171 : walfile = NULL;
225 : :
226 [ - + ]: 171 : if (r != 0)
227 : : {
2705 peter@eisentraut.org 228 :UBC 0 : pg_log_error("could not close file \"%s\": %s",
229 : : fn, GetLastWalMethodError(stream->walmethod));
230 : :
1805 michael@paquier.xyz 231 : 0 : pg_free(fn);
5411 magnus@hagander.net 232 : 0 : return false;
233 : : }
234 : :
1805 michael@paquier.xyz 235 :CBC 171 : pg_free(fn);
236 : :
237 : : /*
238 : : * Mark file as archived if requested by the caller - pg_basebackup needs
239 : : * to do so as files can otherwise get archived again after promotion of a
240 : : * new node. This is in line with walreceiver.c always doing a
241 : : * XLogArchiveForceDone() after a complete segment.
242 : : */
3264 andres@anarazel.de 243 [ + + + + ]: 171 : if (currpos == WalSegSz && stream->mark_done)
244 : : {
245 : : /* writes error message if failed */
1438 rhaas@postgresql.org 246 [ - + ]: 8 : if (!mark_file_as_archived(stream, walfile_name))
4254 andres@anarazel.de 247 :UBC 0 : return false;
248 : : }
249 : :
4591 rhaas@postgresql.org 250 :CBC 171 : lastFlushPosition = pos;
5411 magnus@hagander.net 251 : 171 : return true;
252 : : }
253 : :
254 : :
255 : : /*
256 : : * Check if a timeline history file exists.
257 : : */
258 : : static bool
3821 259 : 165 : existsTimeLineHistoryFile(StreamCtl *stream)
260 : : {
261 : : char histfname[MAXFNAMELEN];
262 : :
263 : : /*
264 : : * Timeline 1 never has a history file. We treat that as if it existed,
265 : : * since we never need to stream it.
266 : : */
267 [ + + ]: 165 : if (stream->timeline == 1)
4970 heikki.linnakangas@i 268 : 160 : return true;
269 : :
3821 magnus@hagander.net 270 : 5 : TLHistoryFileName(histfname, stream->timeline);
271 : :
1438 rhaas@postgresql.org 272 : 5 : return stream->walmethod->ops->existsfile(stream->walmethod, histfname);
273 : : }
274 : :
275 : : static bool
45 peter@eisentraut.org 276 :GNC 5 : writeTimeLineHistoryFile(StreamCtl *stream, const char *filename, const char *content)
277 : : {
43 278 : 5 : size_t size = strlen(content);
279 : : char histfname[MAXFNAMELEN];
280 : : Walfile *f;
281 : :
282 : : /*
283 : : * Check that the server's idea of how timeline history files should be
284 : : * named matches ours.
285 : : */
3821 magnus@hagander.net 286 :CBC 5 : TLHistoryFileName(histfname, stream->timeline);
4970 heikki.linnakangas@i 287 [ - + ]: 5 : if (strcmp(histfname, filename) != 0)
288 : : {
2705 peter@eisentraut.org 289 :UBC 0 : pg_log_error("server reported unexpected history file name for timeline %u: %s",
290 : : stream->timeline, filename);
4970 heikki.linnakangas@i 291 : 0 : return false;
292 : : }
293 : :
1438 rhaas@postgresql.org 294 :CBC 5 : f = stream->walmethod->ops->open_for_write(stream->walmethod,
295 : : histfname, ".tmp", 0);
3595 magnus@hagander.net 296 [ - + ]: 5 : if (f == NULL)
297 : : {
2705 peter@eisentraut.org 298 :UBC 0 : pg_log_error("could not create timeline history file \"%s\": %s",
299 : : histfname, GetLastWalMethodError(stream->walmethod));
4970 heikki.linnakangas@i 300 : 0 : return false;
301 : : }
302 : :
43 peter@eisentraut.org 303 [ - + ]:GNC 5 : if (stream->walmethod->ops->write(f, content, size) != size)
304 : : {
2705 peter@eisentraut.org 305 :UBC 0 : pg_log_error("could not write timeline history file \"%s\": %s",
306 : : histfname, GetLastWalMethodError(stream->walmethod));
307 : :
308 : : /*
309 : : * If we fail to make the file, delete it to release disk space
310 : : */
1438 rhaas@postgresql.org 311 : 0 : stream->walmethod->ops->close(f, CLOSE_UNLINK);
312 : :
4970 heikki.linnakangas@i 313 : 0 : return false;
314 : : }
315 : :
1438 rhaas@postgresql.org 316 [ - + ]:CBC 5 : if (stream->walmethod->ops->close(f, CLOSE_NORMAL) != 0)
317 : : {
2705 peter@eisentraut.org 318 :UBC 0 : pg_log_error("could not close file \"%s\": %s",
319 : : histfname, GetLastWalMethodError(stream->walmethod));
4970 heikki.linnakangas@i 320 : 0 : return false;
321 : : }
322 : :
323 : : /* Maintain archive_status, check close_walfile() for details. */
3821 magnus@hagander.net 324 [ + + ]:CBC 5 : if (stream->mark_done)
325 : : {
326 : : /* writes error message if failed */
3595 327 [ - + ]: 4 : if (!mark_file_as_archived(stream, histfname))
4254 andres@anarazel.de 328 :UBC 0 : return false;
329 : : }
330 : :
4970 heikki.linnakangas@i 331 :CBC 5 : return true;
332 : : }
333 : :
334 : : /*
335 : : * Send a Standby Status Update message to server.
336 : : */
337 : : static bool
3472 tgl@sss.pgh.pa.us 338 : 164 : sendFeedback(PGconn *conn, XLogRecPtr blockpos, TimestampTz now, bool replyRequested)
339 : : {
340 : : char replybuf[1 + 8 + 8 + 8 + 8 + 1];
4838 bruce@momjian.us 341 : 164 : int len = 0;
342 : :
386 nathan@postgresql.or 343 : 164 : replybuf[len] = PqReplMsg_StandbyStatusUpdate;
5041 heikki.linnakangas@i 344 : 164 : len += 1;
3354 tgl@sss.pgh.pa.us 345 : 164 : fe_sendint64(blockpos, &replybuf[len]); /* write */
5041 heikki.linnakangas@i 346 : 164 : len += 8;
4591 rhaas@postgresql.org 347 [ + + ]: 164 : if (reportFlushPosition)
3354 tgl@sss.pgh.pa.us 348 : 160 : fe_sendint64(lastFlushPosition, &replybuf[len]); /* flush */
349 : : else
350 : 4 : fe_sendint64(InvalidXLogRecPtr, &replybuf[len]); /* flush */
5041 heikki.linnakangas@i 351 : 164 : len += 8;
4496 bruce@momjian.us 352 : 164 : fe_sendint64(InvalidXLogRecPtr, &replybuf[len]); /* apply */
5041 heikki.linnakangas@i 353 : 164 : len += 8;
4496 bruce@momjian.us 354 : 164 : fe_sendint64(now, &replybuf[len]); /* sendTime */
5041 heikki.linnakangas@i 355 : 164 : len += 8;
3354 tgl@sss.pgh.pa.us 356 : 164 : replybuf[len] = replyRequested ? 1 : 0; /* replyRequested */
5041 heikki.linnakangas@i 357 : 164 : len += 1;
358 : :
359 [ + - - + ]: 164 : if (PQputCopyData(conn, replybuf, len) <= 0 || PQflush(conn))
360 : : {
2705 peter@eisentraut.org 361 :UBC 0 : pg_log_error("could not send feedback packet: %s",
362 : : PQerrorMessage(conn));
5041 heikki.linnakangas@i 363 : 0 : return false;
364 : : }
365 : :
5041 heikki.linnakangas@i 366 :CBC 164 : return true;
367 : : }
368 : :
369 : : /*
370 : : * Check that the server version we're connected to is supported by
371 : : * ReceiveXlogStream().
372 : : *
373 : : * If it's not, an error message is printed to stderr, and false is returned.
374 : : */
375 : : bool
4906 376 : 349 : CheckServerVersionForStreaming(PGconn *conn)
377 : : {
378 : : int minServerMajor,
379 : : maxServerMajor;
380 : : int serverMajor;
381 : :
382 : : /*
383 : : * The message format used in streaming replication changed in 9.3, so we
384 : : * cannot stream from older servers. And we don't support servers newer
385 : : * than the client; it might work, but we don't know, so err on the safe
386 : : * side.
387 : : */
388 : 349 : minServerMajor = 903;
389 : 349 : maxServerMajor = PG_VERSION_NUM / 100;
390 : 349 : serverMajor = PQserverVersion(conn) / 100;
4496 simon@2ndQuadrant.co 391 [ - + ]: 349 : if (serverMajor < minServerMajor)
392 : : {
4906 heikki.linnakangas@i 393 :UBC 0 : const char *serverver = PQparameterStatus(conn, "server_version");
394 : :
2705 peter@eisentraut.org 395 [ # # ]: 0 : pg_log_error("incompatible server version %s; client does not support streaming from server versions older than %s",
396 : : serverver ? serverver : "'unknown'",
397 : : "9.3");
4496 simon@2ndQuadrant.co 398 : 0 : return false;
399 : : }
4496 simon@2ndQuadrant.co 400 [ - + ]:CBC 349 : else if (serverMajor > maxServerMajor)
401 : : {
4496 simon@2ndQuadrant.co 402 :UBC 0 : const char *serverver = PQparameterStatus(conn, "server_version");
403 : :
2705 peter@eisentraut.org 404 [ # # ]: 0 : pg_log_error("incompatible server version %s; client does not support streaming from server versions newer than %s",
405 : : serverver ? serverver : "'unknown'",
406 : : PG_VERSION);
4906 heikki.linnakangas@i 407 : 0 : return false;
408 : : }
4906 heikki.linnakangas@i 409 :CBC 349 : return true;
410 : : }
411 : :
412 : : /*
413 : : * Receive a log stream starting at the specified position.
414 : : *
415 : : * Individual parameters are passed through the StreamCtl structure.
416 : : *
417 : : * If sysidentifier is specified, validate that both the system
418 : : * identifier and the timeline matches the specified ones
419 : : * (by sending an extra IDENTIFY_SYSTEM command)
420 : : *
421 : : * All received segments will be written to the directory
422 : : * specified by basedir. This will also fetch any missing timeline history
423 : : * files.
424 : : *
425 : : * The stream_stop callback will be called every time data
426 : : * is received, and whenever a segment is completed. If it returns
427 : : * true, the streaming will stop and the function
428 : : * return. As long as it returns false, streaming will continue
429 : : * indefinitely.
430 : : *
431 : : * If stream_stop() checks for external input, stop_socket should be set to
432 : : * the FD it checks. This will allow such input to be detected promptly
433 : : * rather than after standby_message_timeout (which might be indefinite).
434 : : * Note that signals will interrupt waits for input as well, but that is
435 : : * race-y since a signal received while busy won't interrupt the wait.
436 : : *
437 : : * standby_message_timeout controls how often we send a message
438 : : * back to the primary letting it know our progress, in milliseconds.
439 : : * Zero means no messages are sent.
440 : : * This message will only contain the write location, and never
441 : : * flush or replay.
442 : : *
443 : : * If 'partial_suffix' is not NULL, files are initially created with the
444 : : * given suffix, and the suffix is removed once the file is finished. That
445 : : * allows you to tell the difference between partial and completed files,
446 : : * so that you can continue later where you left.
447 : : *
448 : : * If 'synchronous' is true, the received WAL is flushed as soon as written,
449 : : * otherwise only when the WAL file is closed.
450 : : *
451 : : * Note: The WAL location *must* be at a log segment start!
452 : : */
453 : : bool
3821 magnus@hagander.net 454 : 164 : ReceiveXlogStream(PGconn *conn, StreamCtl *stream)
455 : : {
456 : : PQExpBuffer query;
457 : : PGresult *res;
458 : : XLogRecPtr stoppos;
459 : :
460 : : /*
461 : : * The caller should've checked the server version already, but doesn't do
462 : : * any harm to check it here too.
463 : : */
4906 heikki.linnakangas@i 464 [ - + ]: 164 : if (!CheckServerVersionForStreaming(conn))
4972 heikki.linnakangas@i 465 :UBC 0 : return false;
466 : :
467 : : /*
468 : : * Decide whether we want to report the flush position. If we report the
469 : : * flush position, the primary will know what WAL we'll possibly
470 : : * re-request, and it can then remove older WAL safely. We must always do
471 : : * that when we are using slots.
472 : : *
473 : : * Reporting the flush position makes one eligible as a synchronous
474 : : * replica. People shouldn't include generic names in
475 : : * synchronous_standby_names, but we've protected them against it so far,
476 : : * so let's continue to do so unless specifically requested.
477 : : */
3510 magnus@hagander.net 478 [ + + ]:CBC 164 : if (stream->replication_slot != NULL)
479 : : {
4591 rhaas@postgresql.org 480 : 159 : reportFlushPosition = true;
481 : : }
482 : : else
483 : : {
3650 simon@2ndQuadrant.co 484 [ + + ]: 5 : if (stream->synchronous)
485 : 1 : reportFlushPosition = true;
486 : : else
487 : 4 : reportFlushPosition = false;
488 : : }
489 : :
3821 magnus@hagander.net 490 [ + - ]: 164 : if (stream->sysidentifier != NULL)
491 : : {
1822 michael@paquier.xyz 492 : 164 : char *sysidentifier = NULL;
493 : : TimeLineID servertli;
494 : :
495 : : /*
496 : : * Get the server system identifier and timeline, and validate them.
497 : : */
498 [ - + ]: 164 : if (!RunIdentifySystem(conn, &sysidentifier, &servertli, NULL, NULL))
499 : : {
1822 michael@paquier.xyz 500 :UBC 0 : pg_free(sysidentifier);
5419 magnus@hagander.net 501 : 0 : return false;
502 : : }
503 : :
1822 michael@paquier.xyz 504 [ - + ]:CBC 164 : if (strcmp(stream->sysidentifier, sysidentifier) != 0)
505 : : {
2705 peter@eisentraut.org 506 :UBC 0 : pg_log_error("system identifier does not match between base backup and streaming connection");
1822 michael@paquier.xyz 507 : 0 : pg_free(sysidentifier);
5419 magnus@hagander.net 508 : 0 : return false;
509 : : }
1822 michael@paquier.xyz 510 :CBC 164 : pg_free(sysidentifier);
511 : :
512 [ - + ]: 164 : if (stream->timeline > servertli)
513 : : {
2705 peter@eisentraut.org 514 :UBC 0 : pg_log_error("starting timeline %u is not present in the server",
515 : : stream->timeline);
5419 magnus@hagander.net 516 : 0 : return false;
517 : : }
518 : : }
519 : :
520 : : /*
521 : : * initialize flush position to starting point, it's the caller's
522 : : * responsibility that that's sane.
523 : : */
3821 magnus@hagander.net 524 :CBC 164 : lastFlushPosition = stream->startpos;
525 : :
526 : : while (1)
5419 527 : 1 : {
528 : : /*
529 : : * Fetch the timeline history file for this timeline, if we don't have
530 : : * it already. When streaming log to tar, this will always return
531 : : * false, as we are never streaming into an existing file and
532 : : * therefore there can be no pre-existing timeline history file.
533 : : */
3821 534 [ + + ]: 165 : if (!existsTimeLineHistoryFile(stream))
535 : : {
73 tgl@sss.pgh.pa.us 536 : 5 : query = createPQExpBuffer();
537 : 5 : appendPQExpBuffer(query, "TIMELINE_HISTORY %u", stream->timeline);
538 : 5 : res = PQexec(conn, query->data);
539 : 5 : destroyPQExpBuffer(query);
4970 heikki.linnakangas@i 540 [ - + ]: 5 : if (PQresultStatus(res) != PGRES_TUPLES_OK)
541 : : {
542 : : /* FIXME: we might send it ok, but get an error */
2705 peter@eisentraut.org 543 :UBC 0 : pg_log_error("could not send replication command \"%s\": %s",
544 : : "TIMELINE_HISTORY", PQresultErrorMessage(res));
4970 heikki.linnakangas@i 545 : 0 : PQclear(res);
546 : 0 : return false;
547 : : }
548 : :
549 : : /*
550 : : * The response to TIMELINE_HISTORY is a single row result set
551 : : * with two fields: filename and content
552 : : */
4970 heikki.linnakangas@i 553 [ + - - + ]:CBC 5 : if (PQnfields(res) != 2 || PQntuples(res) != 1)
554 : : {
2705 peter@eisentraut.org 555 :UBC 0 : pg_log_warning("unexpected response to TIMELINE_HISTORY command: got %d rows and %d fields, expected %d rows and %d fields",
556 : : PQntuples(res), PQnfields(res), 1, 2);
557 : : }
558 : :
559 : : /* Write the history file to disk */
3821 magnus@hagander.net 560 :CBC 5 : writeTimeLineHistoryFile(stream,
4970 heikki.linnakangas@i 561 :GIC 5 : PQgetvalue(res, 0, 0),
3821 magnus@hagander.net 562 : 5 : PQgetvalue(res, 0, 1));
563 : :
4970 heikki.linnakangas@i 564 :CBC 5 : PQclear(res);
565 : : }
566 : :
567 : : /*
568 : : * Before we start streaming from the requested location, check if the
569 : : * callback tells us to stop here.
570 : : */
3821 magnus@hagander.net 571 [ - + ]: 165 : if (stream->stream_stop(stream->startpos, stream->timeline, false))
4970 heikki.linnakangas@i 572 :UBC 0 : return true;
573 : :
574 : : /* Initiate the replication stream at specified location */
73 tgl@sss.pgh.pa.us 575 :CBC 165 : query = createPQExpBuffer();
576 : 165 : appendPQExpBufferStr(query, "START_REPLICATION");
577 [ + + ]: 165 : if (stream->replication_slot != NULL)
578 : : {
579 : 160 : appendPQExpBufferStr(query, " SLOT ");
580 : 160 : AppendQuotedIdentifier(query, stream->replication_slot);
581 : : }
582 : 165 : appendPQExpBuffer(query, " %X/%08X TIMELINE %u",
583 : 165 : LSN_FORMAT_ARGS(stream->startpos),
584 : : stream->timeline);
585 : 165 : res = PQexec(conn, query->data);
586 : 165 : destroyPQExpBuffer(query);
4970 heikki.linnakangas@i 587 [ + + ]: 165 : if (PQresultStatus(res) != PGRES_COPY_BOTH)
588 : : {
2705 peter@eisentraut.org 589 : 1 : pg_log_error("could not send replication command \"%s\": %s",
590 : : "START_REPLICATION", PQresultErrorMessage(res));
4970 heikki.linnakangas@i 591 : 1 : PQclear(res);
592 : 1 : return false;
593 : : }
5159 magnus@hagander.net 594 : 164 : PQclear(res);
595 : :
596 : : /* Stream the WAL */
3821 597 : 164 : res = HandleCopyStream(conn, stream, &stoppos);
4868 rhaas@postgresql.org 598 [ - + ]: 164 : if (res == NULL)
4970 heikki.linnakangas@i 599 :UBC 0 : goto error;
600 : :
601 : : /*
602 : : * Streaming finished.
603 : : *
604 : : * There are two possible reasons for that: a controlled shutdown, or
605 : : * we reached the end of the current timeline. In case of
606 : : * end-of-timeline, the server sends a result set after Copy has
607 : : * finished, containing information about the next timeline. Read
608 : : * that, and restart streaming from the next timeline. In case of
609 : : * controlled shutdown, stop here.
610 : : */
4970 heikki.linnakangas@i 611 [ + + ]:CBC 164 : if (PQresultStatus(res) == PGRES_TUPLES_OK)
612 : 1 : {
613 : : /*
614 : : * End-of-timeline. Read the next timeline's ID and starting
615 : : * position. Usually, the starting position will match the end of
616 : : * the previous timeline, but there are corner cases like if the
617 : : * server had sent us half of a WAL record, when it was promoted.
618 : : * The new timeline will begin at the end of the last complete
619 : : * record in that case, overlapping the partial WAL record on the
620 : : * old timeline.
621 : : */
622 : : uint32 newtimeline;
623 : : bool parsed;
624 : :
3821 magnus@hagander.net 625 : 1 : parsed = ReadEndOfStreamingResult(res, &stream->startpos, &newtimeline);
4970 heikki.linnakangas@i 626 : 1 : PQclear(res);
4859 627 [ - + ]: 1 : if (!parsed)
4859 heikki.linnakangas@i 628 :UBC 0 : goto error;
629 : :
630 : : /* Sanity check the values the server gave us */
3821 magnus@hagander.net 631 [ - + ]:CBC 1 : if (newtimeline <= stream->timeline)
632 : : {
2705 peter@eisentraut.org 633 :UBC 0 : pg_log_error("server reported unexpected next timeline %u, following timeline %u",
634 : : newtimeline, stream->timeline);
4859 heikki.linnakangas@i 635 : 0 : goto error;
636 : : }
3821 magnus@hagander.net 637 [ - + ]:CBC 1 : if (stream->startpos > stoppos)
638 : : {
416 alvherre@kurilemu.de 639 :UBC 0 : pg_log_error("server stopped streaming timeline %u at %X/%08X, but reported next timeline %u to begin at %X/%08X",
640 : : stream->timeline, LSN_FORMAT_ARGS(stoppos),
641 : : newtimeline, LSN_FORMAT_ARGS(stream->startpos));
4970 heikki.linnakangas@i 642 : 0 : goto error;
643 : : }
644 : :
645 : : /* Read the final result, which should be CommandComplete. */
4970 heikki.linnakangas@i 646 :CBC 1 : res = PQgetResult(conn);
647 [ - + ]: 1 : if (PQresultStatus(res) != PGRES_COMMAND_OK)
648 : : {
2705 peter@eisentraut.org 649 :UBC 0 : pg_log_error("unexpected termination of replication stream: %s",
650 : : PQresultErrorMessage(res));
4408 fujii@postgresql.org 651 : 0 : PQclear(res);
4970 heikki.linnakangas@i 652 : 0 : goto error;
653 : : }
4970 heikki.linnakangas@i 654 :CBC 1 : PQclear(res);
655 : :
656 : : /*
657 : : * Loop back to start streaming from the new timeline. Always
658 : : * start streaming at the beginning of a segment.
659 : : */
3821 magnus@hagander.net 660 : 1 : stream->timeline = newtimeline;
3264 andres@anarazel.de 661 : 1 : stream->startpos = stream->startpos -
662 : 1 : XLogSegmentOffset(stream->startpos, WalSegSz);
4970 heikki.linnakangas@i 663 : 1 : continue;
664 : : }
665 [ + + ]: 163 : else if (PQresultStatus(res) == PGRES_COMMAND_OK)
666 : : {
4408 fujii@postgresql.org 667 : 162 : PQclear(res);
668 : :
669 : : /*
670 : : * End of replication (ie. controlled shut down of the server).
671 : : *
672 : : * Check if the callback thinks it's OK to stop here. If not,
673 : : * complain.
674 : : */
3821 magnus@hagander.net 675 [ + - ]: 162 : if (stream->stream_stop(stoppos, stream->timeline, false))
4970 heikki.linnakangas@i 676 : 162 : return true;
677 : : else
678 : : {
2705 peter@eisentraut.org 679 :UBC 0 : pg_log_error("replication stream was terminated before stop point");
4970 heikki.linnakangas@i 680 : 0 : goto error;
681 : : }
682 : : }
683 : : else
684 : : {
685 : : /* Server returned an error. */
2705 peter@eisentraut.org 686 :CBC 1 : pg_log_error("unexpected termination of replication stream: %s",
687 : : PQresultErrorMessage(res));
4408 fujii@postgresql.org 688 : 1 : PQclear(res);
4970 heikki.linnakangas@i 689 : 1 : goto error;
690 : : }
691 : : }
692 : :
693 : 1 : error:
1438 rhaas@postgresql.org 694 [ - + - - ]: 1 : if (walfile != NULL && stream->walmethod->ops->close(walfile, CLOSE_NO_RENAME) != 0)
2705 peter@eisentraut.org 695 :UBC 0 : pg_log_error("could not close file \"%s\": %s",
696 : : walfile->pathname, GetLastWalMethodError(stream->walmethod));
3595 magnus@hagander.net 697 :CBC 1 : walfile = NULL;
4970 heikki.linnakangas@i 698 : 1 : return false;
699 : : }
700 : :
701 : : /*
702 : : * Helper function to parse the result set returned by server after streaming
703 : : * has finished. On failure, prints an error to stderr and returns false.
704 : : */
705 : : static bool
4859 706 : 1 : ReadEndOfStreamingResult(PGresult *res, XLogRecPtr *startpos, uint32 *timeline)
707 : : {
708 : : /*----------
709 : : * The result set consists of one row and two columns, e.g:
710 : : *
711 : : * next_tli | next_tli_startpos
712 : : * ----------+-------------------
713 : : * 4 | 0/9949AE0
714 : : *
715 : : * next_tli is the timeline ID of the next timeline after the one that
716 : : * just finished streaming. next_tli_startpos is the WAL location where
717 : : * the server switched to it.
718 : : *----------
719 : : */
720 [ + - - + ]: 1 : if (PQnfields(res) < 2 || PQntuples(res) != 1)
721 : : {
2705 peter@eisentraut.org 722 :UBC 0 : pg_log_error("unexpected result set after end-of-timeline: got %d rows and %d fields, expected %d rows and %d fields",
723 : : PQntuples(res), PQnfields(res), 1, 2);
4859 heikki.linnakangas@i 724 : 0 : return false;
725 : : }
726 : :
4859 heikki.linnakangas@i 727 :CBC 1 : *timeline = atoi(PQgetvalue(res, 0, 0));
6 fujii@postgresql.org 728 [ - + ]:GNC 1 : if (!pg_parse_lsn(PQgetvalue(res, 0, 1), startpos))
729 : : {
2705 peter@eisentraut.org 730 :UBC 0 : pg_log_error("could not parse next timeline's starting point \"%s\"",
731 : : PQgetvalue(res, 0, 1));
4859 heikki.linnakangas@i 732 : 0 : return false;
733 : : }
734 : :
4859 heikki.linnakangas@i 735 :CBC 1 : return true;
736 : : }
737 : :
738 : : /*
739 : : * The main loop of ReceiveXlogStream. Handles the COPY stream after
740 : : * initiating streaming with the START_REPLICATION command.
741 : : *
742 : : * If the COPY ends (not necessarily successfully) due a message from the
743 : : * server, returns a PGresult and sets *stoppos to the last byte written.
744 : : * On any other sort of error, returns NULL.
745 : : */
746 : : static PGresult *
3821 magnus@hagander.net 747 : 164 : HandleCopyStream(PGconn *conn, StreamCtl *stream,
748 : : XLogRecPtr *stoppos)
749 : : {
4970 heikki.linnakangas@i 750 : 164 : char *copybuf = NULL;
3472 tgl@sss.pgh.pa.us 751 : 164 : TimestampTz last_status = -1;
3821 magnus@hagander.net 752 : 164 : XLogRecPtr blockpos = stream->startpos;
753 : :
4404 fujii@postgresql.org 754 : 164 : still_sending = true;
755 : :
756 : : while (1)
5419 magnus@hagander.net 757 : 1413 : {
758 : : int r;
759 : : TimestampTz now;
760 : : long sleeptime;
761 : :
762 : : /*
763 : : * Check if we should continue streaming, or abort at this point.
764 : : */
2177 peter@eisentraut.org 765 [ - + ]: 1577 : if (!CheckCopyStreamStop(conn, stream, blockpos))
4402 fujii@postgresql.org 766 :UBC 0 : goto error;
767 : :
4402 fujii@postgresql.org 768 :CBC 1577 : now = feGetCurrentTimestamp();
769 : :
770 : : /*
771 : : * If synchronous option is true, issue sync command as soon as there
772 : : * are WAL data which has not been flushed yet.
773 : : */
3595 magnus@hagander.net 774 [ + + - + : 1577 : if (stream->synchronous && lastFlushPosition < blockpos && walfile != NULL)
- - ]
775 : : {
1438 rhaas@postgresql.org 776 [ # # ]:UBC 0 : if (stream->walmethod->ops->sync(walfile) != 0)
1602 tgl@sss.pgh.pa.us 777 : 0 : pg_fatal("could not fsync file \"%s\": %s",
778 : : walfile->pathname, GetLastWalMethodError(stream->walmethod));
4402 fujii@postgresql.org 779 : 0 : lastFlushPosition = blockpos;
780 : :
781 : : /*
782 : : * Send feedback so that the server sees the latest WAL locations
783 : : * immediately.
784 : : */
4300 785 [ # # ]: 0 : if (!sendFeedback(conn, blockpos, now, false))
786 : 0 : goto error;
787 : 0 : last_status = now;
788 : : }
789 : :
790 : : /*
791 : : * Potentially send a status message to the primary
792 : : */
3821 magnus@hagander.net 793 [ + + + - :CBC 3011 : if (still_sending && stream->standby_message_timeout > 0 &&
+ + ]
4545 rhaas@postgresql.org 794 : 1434 : feTimestampDifferenceExceeds(last_status, now,
795 : : stream->standby_message_timeout))
796 : : {
797 : : /* Time to send feedback! */
5040 heikki.linnakangas@i 798 [ - + ]: 164 : if (!sendFeedback(conn, blockpos, now, false))
5159 magnus@hagander.net 799 :UBC 0 : goto error;
5419 magnus@hagander.net 800 :CBC 164 : last_status = now;
801 : : }
802 : :
803 : : /*
804 : : * Calculate how long send/receive loops should sleep
805 : : */
3821 806 : 1577 : sleeptime = CalculateCopyStreamSleeptime(now, stream->standby_message_timeout,
807 : : last_status);
808 : :
809 : : /* Done with any prior message */
561 tgl@sss.pgh.pa.us 810 : 1577 : PQfreemem(copybuf);
811 : 1577 : copybuf = NULL;
812 : :
3409 813 : 1577 : r = CopyStreamReceive(conn, sleeptime, stream->stop_socket, ©buf);
4402 fujii@postgresql.org 814 [ + + ]: 3804 : while (r != 0)
815 : : {
816 [ - + ]: 2391 : if (r == -1)
4404 fujii@postgresql.org 817 :UBC 0 : goto error;
4402 fujii@postgresql.org 818 [ + + ]:CBC 2391 : if (r == -2)
819 : : {
3821 magnus@hagander.net 820 : 164 : PGresult *res = HandleEndOfCopyStream(conn, stream, copybuf, blockpos, stoppos);
821 : :
4402 fujii@postgresql.org 822 [ - + ]: 164 : if (res == NULL)
4402 fujii@postgresql.org 823 :UBC 0 : goto error;
561 tgl@sss.pgh.pa.us 824 :CBC 164 : PQfreemem(copybuf);
825 : 164 : return res;
826 : : }
827 : :
828 : : /* Check the message type. */
386 nathan@postgresql.or 829 [ - + ]: 2227 : if (copybuf[0] == PqReplMsg_Keepalive)
830 : : {
3619 peter_e@gmx.net 831 [ # # ]:UBC 0 : if (!ProcessKeepaliveMsg(conn, stream, copybuf, r, blockpos,
832 : : &last_status))
4402 fujii@postgresql.org 833 : 0 : goto error;
834 : : }
386 nathan@postgresql.or 835 [ + - ]:CBC 2227 : else if (copybuf[0] == PqReplMsg_WALData)
836 : : {
388 alvherre@kurilemu.de 837 [ - + ]: 2227 : if (!ProcessWALDataMsg(conn, stream, copybuf, r, &blockpos))
4402 fujii@postgresql.org 838 :UBC 0 : goto error;
839 : :
840 : : /*
841 : : * Check if we should continue streaming, or abort at this
842 : : * point.
843 : : */
2177 peter@eisentraut.org 844 [ - + ]:CBC 2227 : if (!CheckCopyStreamStop(conn, stream, blockpos))
4402 fujii@postgresql.org 845 :UBC 0 : goto error;
846 : : }
847 : : else
848 : : {
2705 peter@eisentraut.org 849 : 0 : pg_log_error("unrecognized streaming header: \"%c\"",
850 : : copybuf[0]);
5159 magnus@hagander.net 851 : 0 : goto error;
852 : : }
853 : :
854 : : /* Done with that message */
561 tgl@sss.pgh.pa.us 855 :CBC 2227 : PQfreemem(copybuf);
856 : 2227 : copybuf = NULL;
857 : :
858 : : /*
859 : : * Process the received data, and any subsequent data we can read
860 : : * without blocking.
861 : : */
3409 862 : 2227 : r = CopyStreamReceive(conn, 0, stream->stop_socket, ©buf);
863 : : }
864 : : }
865 : :
5159 magnus@hagander.net 866 :UBC 0 : error:
1462 peter@eisentraut.org 867 : 0 : PQfreemem(copybuf);
4868 rhaas@postgresql.org 868 : 0 : return NULL;
869 : : }
870 : :
871 : : /*
872 : : * Wait until we can read a CopyData message,
873 : : * or timeout, or occurrence of a signal or input on the stop_socket.
874 : : * (timeout_ms < 0 means wait indefinitely; 0 means don't wait.)
875 : : *
876 : : * Returns 1 if data has become available for reading, 0 if timed out
877 : : * or interrupted by signal or stop_socket input, and -1 on an error.
878 : : */
879 : : static int
3409 tgl@sss.pgh.pa.us 880 :CBC 3595 : CopyStreamPoll(PGconn *conn, long timeout_ms, pgsocket stop_socket)
881 : : {
882 : : int ret;
883 : : fd_set input_mask;
884 : : int connsocket;
885 : : int maxfd;
886 : : struct timeval timeout;
887 : : struct timeval *timeoutptr;
888 : :
889 : 3595 : connsocket = PQsocket(conn);
890 [ - + ]: 3595 : if (connsocket < 0)
891 : : {
2705 peter@eisentraut.org 892 :UBC 0 : pg_log_error("invalid socket: %s", PQerrorMessage(conn));
4437 fujii@postgresql.org 893 : 0 : return -1;
894 : : }
895 : :
4437 fujii@postgresql.org 896 [ + + ]:CBC 61115 : FD_ZERO(&input_mask);
3409 tgl@sss.pgh.pa.us 897 : 3595 : FD_SET(connsocket, &input_mask);
898 : 3595 : maxfd = connsocket;
899 [ + + ]: 3595 : if (stop_socket != PGINVALID_SOCKET)
900 : : {
901 : 3485 : FD_SET(stop_socket, &input_mask);
902 : 3485 : maxfd = Max(maxfd, stop_socket);
903 : : }
904 : :
4437 fujii@postgresql.org 905 [ + + ]: 3595 : if (timeout_ms < 0)
906 : 143 : timeoutptr = NULL;
907 : : else
908 : : {
909 : 3452 : timeout.tv_sec = timeout_ms / 1000L;
910 : 3452 : timeout.tv_usec = (timeout_ms % 1000L) * 1000L;
911 : 3452 : timeoutptr = &timeout;
912 : : }
913 : :
3409 tgl@sss.pgh.pa.us 914 : 3595 : ret = select(maxfd + 1, &input_mask, NULL, NULL, timeoutptr);
915 : :
916 [ - + ]: 3595 : if (ret < 0)
917 : : {
3409 tgl@sss.pgh.pa.us 918 [ # # ]:UBC 0 : if (errno == EINTR)
919 : 0 : return 0; /* Got a signal, so not an error */
1952 peter@eisentraut.org 920 : 0 : pg_log_error("%s() failed: %m", "select");
4437 fujii@postgresql.org 921 : 0 : return -1;
922 : : }
3409 tgl@sss.pgh.pa.us 923 [ + + + + ]:CBC 3595 : if (ret > 0 && FD_ISSET(connsocket, &input_mask))
924 : 2792 : return 1; /* Got input on connection socket */
925 : :
926 : 803 : return 0; /* Got timeout or input on stop_socket */
927 : : }
928 : :
929 : : /*
930 : : * Receive CopyData message available from XLOG stream, blocking for
931 : : * maximum of 'timeout' ms.
932 : : *
933 : : * If data was received, returns the length of the data. *buffer is set to
934 : : * point to a buffer holding the received message. The caller must eventually
935 : : * free the buffer with PQfreemem().
936 : : *
937 : : * Returns 0 if no data was available within timeout, or if wait was
938 : : * interrupted by signal or stop_socket input.
939 : : * -1 on error. -2 if the server ended the COPY.
940 : : */
941 : : static int
942 : 3804 : CopyStreamReceive(PGconn *conn, long timeout, pgsocket stop_socket,
943 : : char **buffer)
944 : : {
4437 fujii@postgresql.org 945 : 3804 : char *copybuf = NULL;
946 : : int rawlen;
947 : :
948 : : /* Caller should have cleared any prior buffer */
561 tgl@sss.pgh.pa.us 949 [ - + ]: 3804 : Assert(*buffer == NULL);
950 : :
951 : : /* Try to receive a CopyData message */
4437 fujii@postgresql.org 952 : 3804 : rawlen = PQgetCopyData(conn, ©buf, 1);
953 [ + + ]: 3804 : if (rawlen == 0)
954 : : {
955 : : int ret;
956 : :
957 : : /*
958 : : * No data available. Wait for some to appear, but not longer than
959 : : * the specified timeout, so that we can ping the server. Also stop
960 : : * waiting if input appears on stop_socket.
961 : : */
3409 tgl@sss.pgh.pa.us 962 : 3595 : ret = CopyStreamPoll(conn, timeout, stop_socket);
963 [ + + ]: 3595 : if (ret <= 0)
964 : 803 : return ret;
965 : :
966 : : /* Now there is actually data on the socket */
4437 fujii@postgresql.org 967 [ - + ]: 2792 : if (PQconsumeInput(conn) == 0)
968 : : {
2705 peter@eisentraut.org 969 :UBC 0 : pg_log_error("could not receive data from WAL stream: %s",
970 : : PQerrorMessage(conn));
4437 fujii@postgresql.org 971 : 0 : return -1;
972 : : }
973 : :
974 : : /* Now that we've consumed some input, try again */
4437 fujii@postgresql.org 975 :CBC 2792 : rawlen = PQgetCopyData(conn, ©buf, 1);
976 [ + + ]: 2792 : if (rawlen == 0)
977 : 610 : return 0;
978 : : }
979 [ + + ]: 2391 : if (rawlen == -1) /* end-of-streaming or error */
980 : 164 : return -2;
981 [ - + ]: 2227 : if (rawlen == -2)
982 : : {
2705 peter@eisentraut.org 983 :UBC 0 : pg_log_error("could not read COPY data: %s", PQerrorMessage(conn));
4437 fujii@postgresql.org 984 : 0 : return -1;
985 : : }
986 : :
987 : : /* Return received messages to caller */
4437 fujii@postgresql.org 988 :CBC 2227 : *buffer = copybuf;
989 : 2227 : return rawlen;
990 : : }
991 : :
992 : : /*
993 : : * Process the keepalive message.
994 : : */
995 : : static bool
3619 peter_e@gmx.net 996 :UBC 0 : ProcessKeepaliveMsg(PGconn *conn, StreamCtl *stream, char *copybuf, int len,
997 : : XLogRecPtr blockpos, TimestampTz *last_status)
998 : : {
999 : : int pos;
1000 : : bool replyRequested;
1001 : : TimestampTz now;
1002 : :
1003 : : /*
1004 : : * Parse the keepalive message, enclosed in the CopyData message. We just
1005 : : * check if the server requested a reply, and ignore the rest.
1006 : : */
386 nathan@postgresql.or 1007 : 0 : pos = 1; /* skip msgtype PqReplMsg_Keepalive */
4114 bruce@momjian.us 1008 : 0 : pos += 8; /* skip walEnd */
1009 : 0 : pos += 8; /* skip sendTime */
1010 : :
4404 fujii@postgresql.org 1011 [ # # ]: 0 : if (len < pos + 1)
1012 : : {
2705 peter@eisentraut.org 1013 : 0 : pg_log_error("streaming header too small: %d", len);
4404 fujii@postgresql.org 1014 : 0 : return false;
1015 : : }
1016 : 0 : replyRequested = copybuf[pos];
1017 : :
1018 : : /* If the server requested an immediate reply, send one. */
1019 [ # # # # ]: 0 : if (replyRequested && still_sending)
1020 : : {
4299 1021 [ # # # # ]: 0 : if (reportFlushPosition && lastFlushPosition < blockpos &&
3595 magnus@hagander.net 1022 [ # # ]: 0 : walfile != NULL)
1023 : : {
1024 : : /*
1025 : : * If a valid flush location needs to be reported, flush the
1026 : : * current WAL file so that the latest flush location is sent back
1027 : : * to the server. This is necessary to see whether the last WAL
1028 : : * data has been successfully replicated or not, at the normal
1029 : : * shutdown of the server.
1030 : : */
1438 rhaas@postgresql.org 1031 [ # # ]: 0 : if (stream->walmethod->ops->sync(walfile) != 0)
1602 tgl@sss.pgh.pa.us 1032 : 0 : pg_fatal("could not fsync file \"%s\": %s",
1033 : : walfile->pathname, GetLastWalMethodError(stream->walmethod));
4299 fujii@postgresql.org 1034 : 0 : lastFlushPosition = blockpos;
1035 : : }
1036 : :
4404 1037 : 0 : now = feGetCurrentTimestamp();
1038 [ # # ]: 0 : if (!sendFeedback(conn, blockpos, now, false))
1039 : 0 : return false;
1040 : 0 : *last_status = now;
1041 : : }
1042 : :
1043 : 0 : return true;
1044 : : }
1045 : :
1046 : : /*
1047 : : * Process WALData message.
1048 : : */
1049 : : static bool
388 alvherre@kurilemu.de 1050 :CBC 2227 : ProcessWALDataMsg(PGconn *conn, StreamCtl *stream, char *copybuf, int len,
1051 : : XLogRecPtr *blockpos)
1052 : : {
1053 : : int xlogoff;
1054 : : int bytes_left;
1055 : : int bytes_written;
1056 : : int hdr_len;
1057 : :
1058 : : /*
1059 : : * Once we've decided we don't want to receive any more, just ignore any
1060 : : * subsequent WALData messages.
1061 : : */
4404 fujii@postgresql.org 1062 [ + + ]: 2227 : if (!(still_sending))
1063 : 185 : return true;
1064 : :
1065 : : /*
1066 : : * Read the header of the WALData message, enclosed in the CopyData
1067 : : * message. We only need the WAL location field (dataStart), the rest of
1068 : : * the header is ignored.
1069 : : */
386 nathan@postgresql.or 1070 : 2042 : hdr_len = 1; /* msgtype PqReplMsg_WALData */
4114 bruce@momjian.us 1071 : 2042 : hdr_len += 8; /* dataStart */
1072 : 2042 : hdr_len += 8; /* walEnd */
1073 : 2042 : hdr_len += 8; /* sendTime */
4404 fujii@postgresql.org 1074 [ - + ]: 2042 : if (len < hdr_len)
1075 : : {
2705 peter@eisentraut.org 1076 :UBC 0 : pg_log_error("streaming header too small: %d", len);
4404 fujii@postgresql.org 1077 : 0 : return false;
1078 : : }
4404 fujii@postgresql.org 1079 :CBC 2042 : *blockpos = fe_recvint64(©buf[1]);
1080 : :
1081 : : /* Extract WAL location for this block */
3264 andres@anarazel.de 1082 : 2042 : xlogoff = XLogSegmentOffset(*blockpos, WalSegSz);
1083 : :
1084 : : /*
1085 : : * Verify that the initial location in the stream matches where we think
1086 : : * we are.
1087 : : */
3595 magnus@hagander.net 1088 [ + + ]: 2042 : if (walfile == NULL)
1089 : : {
1090 : : /* No file open yet */
4404 fujii@postgresql.org 1091 [ - + ]: 171 : if (xlogoff != 0)
1092 : : {
2705 peter@eisentraut.org 1093 :UBC 0 : pg_log_error("received write-ahead log record for offset %u with no file open",
1094 : : xlogoff);
4404 fujii@postgresql.org 1095 : 0 : return false;
1096 : : }
1097 : : }
1098 : : else
1099 : : {
1100 : : /* More data in existing segment */
1438 rhaas@postgresql.org 1101 [ - + ]:CBC 1871 : if (walfile->currpos != xlogoff)
1102 : : {
2705 peter@eisentraut.org 1103 :UBC 0 : pg_log_error("got WAL data offset %08x, expected %08x",
1104 : : xlogoff, (int) walfile->currpos);
4404 fujii@postgresql.org 1105 : 0 : return false;
1106 : : }
1107 : : }
1108 : :
4404 fujii@postgresql.org 1109 :CBC 2042 : bytes_left = len - hdr_len;
1110 : 2042 : bytes_written = 0;
1111 : :
1112 [ + + ]: 4084 : while (bytes_left)
1113 : : {
1114 : : int bytes_to_write;
1115 : :
1116 : : /*
1117 : : * If crossing a WAL boundary, only write up until we reach wal
1118 : : * segment size.
1119 : : */
3264 andres@anarazel.de 1120 [ - + ]: 2042 : if (xlogoff + bytes_left > WalSegSz)
3264 andres@anarazel.de 1121 :UBC 0 : bytes_to_write = WalSegSz - xlogoff;
1122 : : else
4404 fujii@postgresql.org 1123 :CBC 2042 : bytes_to_write = bytes_left;
1124 : :
3595 magnus@hagander.net 1125 [ + + ]: 2042 : if (walfile == NULL)
1126 : : {
3821 1127 [ - + ]: 171 : if (!open_walfile(stream, *blockpos))
1128 : : {
1129 : : /* Error logged by open_walfile */
4404 fujii@postgresql.org 1130 :UBC 0 : return false;
1131 : : }
1132 : : }
1133 : :
1438 rhaas@postgresql.org 1134 :CBC 4084 : if (stream->walmethod->ops->write(walfile,
1135 : 2042 : copybuf + hdr_len + bytes_written,
1136 [ - + ]: 2042 : bytes_to_write) != bytes_to_write)
1137 : : {
1744 peter@eisentraut.org 1138 :UBC 0 : pg_log_error("could not write %d bytes to WAL file \"%s\": %s",
1139 : : bytes_to_write, walfile->pathname,
1140 : : GetLastWalMethodError(stream->walmethod));
4404 fujii@postgresql.org 1141 : 0 : return false;
1142 : : }
1143 : :
1144 : : /* Write was successful, advance our position */
4404 fujii@postgresql.org 1145 :CBC 2042 : bytes_written += bytes_to_write;
1146 : 2042 : bytes_left -= bytes_to_write;
1147 : 2042 : *blockpos += bytes_to_write;
1148 : 2042 : xlogoff += bytes_to_write;
1149 : :
1150 : : /* Did we reach the end of a WAL segment? */
3264 andres@anarazel.de 1151 [ + + ]: 2042 : if (XLogSegmentOffset(*blockpos, WalSegSz) == 0)
1152 : : {
3821 magnus@hagander.net 1153 [ - + ]: 14 : if (!close_walfile(stream, *blockpos))
1154 : : /* Error message written in close_walfile() */
4404 fujii@postgresql.org 1155 :UBC 0 : return false;
1156 : :
4404 fujii@postgresql.org 1157 :CBC 14 : xlogoff = 0;
1158 : :
3821 magnus@hagander.net 1159 [ + - - + ]: 14 : if (still_sending && stream->stream_stop(*blockpos, stream->timeline, true))
1160 : : {
4404 fujii@postgresql.org 1161 [ # # # # ]:UBC 0 : if (PQputCopyEnd(conn, NULL) <= 0 || PQflush(conn))
1162 : : {
2705 peter@eisentraut.org 1163 : 0 : pg_log_error("could not send copy-end packet: %s",
1164 : : PQerrorMessage(conn));
4404 fujii@postgresql.org 1165 : 0 : return false;
1166 : : }
1167 : 0 : still_sending = false;
388 alvherre@kurilemu.de 1168 : 0 : return true; /* ignore the rest of this WALData packet */
1169 : : }
1170 : : }
1171 : : }
1172 : : /* No more data left to write, receive next copy packet */
1173 : :
4404 fujii@postgresql.org 1174 :CBC 2042 : return true;
1175 : : }
1176 : :
1177 : : /*
1178 : : * Handle end of the copy stream.
1179 : : */
1180 : : static PGresult *
3821 magnus@hagander.net 1181 : 164 : HandleEndOfCopyStream(PGconn *conn, StreamCtl *stream, char *copybuf,
1182 : : XLogRecPtr blockpos, XLogRecPtr *stoppos)
1183 : : {
4404 fujii@postgresql.org 1184 : 164 : PGresult *res = PQgetResult(conn);
1185 : :
1186 : : /*
1187 : : * The server closed its end of the copy stream. If we haven't closed
1188 : : * ours already, we need to do so now, unless the server threw an error,
1189 : : * in which case we don't.
1190 : : */
1191 [ + + ]: 164 : if (still_sending)
1192 : : {
3821 magnus@hagander.net 1193 [ - + ]: 2 : if (!close_walfile(stream, blockpos))
1194 : : {
1195 : : /* Error message written in close_walfile() */
4404 fujii@postgresql.org 1196 :UBC 0 : PQclear(res);
1197 : 0 : return NULL;
1198 : : }
4404 fujii@postgresql.org 1199 [ + + ]:CBC 2 : if (PQresultStatus(res) == PGRES_COPY_IN)
1200 : : {
1201 [ + - - + ]: 1 : if (PQputCopyEnd(conn, NULL) <= 0 || PQflush(conn))
1202 : : {
2705 peter@eisentraut.org 1203 :UBC 0 : pg_log_error("could not send copy-end packet: %s",
1204 : : PQerrorMessage(conn));
4404 fujii@postgresql.org 1205 : 0 : PQclear(res);
1206 : 0 : return NULL;
1207 : : }
4404 fujii@postgresql.org 1208 :CBC 1 : res = PQgetResult(conn);
1209 : : }
1210 : 2 : still_sending = false;
1211 : : }
1212 : 164 : *stoppos = blockpos;
1213 : 164 : return res;
1214 : : }
1215 : :
1216 : : /*
1217 : : * Check if we should continue streaming, or abort at this point.
1218 : : */
1219 : : static bool
2177 peter@eisentraut.org 1220 : 3804 : CheckCopyStreamStop(PGconn *conn, StreamCtl *stream, XLogRecPtr blockpos)
1221 : : {
3821 magnus@hagander.net 1222 [ + + + + ]: 3804 : if (still_sending && stream->stream_stop(blockpos, stream->timeline, false))
1223 : : {
1224 [ - + ]: 162 : if (!close_walfile(stream, blockpos))
1225 : : {
1226 : : /* Potential error message is written by close_walfile */
4402 fujii@postgresql.org 1227 :UBC 0 : return false;
1228 : : }
4402 fujii@postgresql.org 1229 [ + - - + ]:CBC 162 : if (PQputCopyEnd(conn, NULL) <= 0 || PQflush(conn))
1230 : : {
2705 peter@eisentraut.org 1231 :UBC 0 : pg_log_error("could not send copy-end packet: %s",
1232 : : PQerrorMessage(conn));
4402 fujii@postgresql.org 1233 : 0 : return false;
1234 : : }
4402 fujii@postgresql.org 1235 :CBC 162 : still_sending = false;
1236 : : }
1237 : :
1238 : 3804 : return true;
1239 : : }
1240 : :
1241 : : /*
1242 : : * Calculate how long send/receive loops should sleep
1243 : : */
1244 : : static long
3472 tgl@sss.pgh.pa.us 1245 : 1577 : CalculateCopyStreamSleeptime(TimestampTz now, int standby_message_timeout,
1246 : : TimestampTz last_status)
1247 : : {
1248 : 1577 : TimestampTz status_targettime = 0;
1249 : : long sleeptime;
1250 : :
4402 fujii@postgresql.org 1251 [ + - + + ]: 1577 : if (standby_message_timeout && still_sending)
1252 : 1434 : status_targettime = last_status +
1253 : 1434 : (standby_message_timeout - 1) * ((int64) 1000);
1254 : :
4300 1255 [ + + ]: 1577 : if (status_targettime > 0)
1256 : : {
1257 : : long secs;
1258 : : int usecs;
1259 : :
4402 1260 : 1434 : feTimestampDifference(now,
1261 : : status_targettime,
1262 : : &secs,
1263 : : &usecs);
1264 : : /* Always sleep at least 1 sec */
1265 [ - + ]: 1434 : if (secs <= 0)
1266 : : {
4402 fujii@postgresql.org 1267 :UBC 0 : secs = 1;
1268 : 0 : usecs = 0;
1269 : : }
1270 : :
4402 fujii@postgresql.org 1271 :CBC 1434 : sleeptime = secs * 1000 + usecs / 1000;
1272 : : }
1273 : : else
1274 : 143 : sleeptime = -1;
1275 : :
1276 : 1577 : return sleeptime;
1277 : : }
|
