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