PostgreSQL Source Code  git master
 All Data Structures Namespaces Files Functions Variables Typedefs Enumerations Enumerator Macros
receivelog.c File Reference
#include "postgres_fe.h"
#include <sys/stat.h>
#include <unistd.h>
#include "receivelog.h"
#include "streamutil.h"
#include "libpq-fe.h"
#include "access/xlog_internal.h"
#include "common/file_utils.h"
Include dependency graph for receivelog.c:

Go to the source code of this file.

Functions

static PGresultHandleCopyStream (PGconn *conn, StreamCtl *stream, XLogRecPtr *stoppos)
 
static int CopyStreamPoll (PGconn *conn, long timeout_ms)
 
static int CopyStreamReceive (PGconn *conn, long timeout, char **buffer)
 
static bool ProcessKeepaliveMsg (PGconn *conn, StreamCtl *stream, char *copybuf, int len, XLogRecPtr blockpos, TimestampTz *last_status)
 
static bool ProcessXLogDataMsg (PGconn *conn, StreamCtl *stream, char *copybuf, int len, XLogRecPtr *blockpos)
 
static PGresultHandleEndOfCopyStream (PGconn *conn, StreamCtl *stream, char *copybuf, XLogRecPtr blockpos, XLogRecPtr *stoppos)
 
static bool CheckCopyStreamStop (PGconn *conn, StreamCtl *stream, XLogRecPtr blockpos, XLogRecPtr *stoppos)
 
static long CalculateCopyStreamSleeptime (TimestampTz now, int standby_message_timeout, TimestampTz last_status)
 
static bool ReadEndOfStreamingResult (PGresult *res, XLogRecPtr *startpos, uint32 *timeline)
 
static bool mark_file_as_archived (StreamCtl *stream, const char *fname)
 
static bool open_walfile (StreamCtl *stream, XLogRecPtr startpoint)
 
static bool close_walfile (StreamCtl *stream, XLogRecPtr pos)
 
static bool existsTimeLineHistoryFile (StreamCtl *stream)
 
static bool writeTimeLineHistoryFile (StreamCtl *stream, char *filename, char *content)
 
static bool sendFeedback (PGconn *conn, XLogRecPtr blockpos, TimestampTz now, bool replyRequested)
 
bool CheckServerVersionForStreaming (PGconn *conn)
 
bool ReceiveXlogStream (PGconn *conn, StreamCtl *stream)
 

Variables

static Walfilewalfile = NULL
 
static char current_walfile_name [MAXPGPATH] = ""
 
static bool reportFlushPosition = false
 
static XLogRecPtr lastFlushPosition = InvalidXLogRecPtr
 
static bool still_sending = true
 

Function Documentation

static long CalculateCopyStreamSleeptime ( TimestampTz  now,
int  standby_message_timeout,
TimestampTz  last_status 
)
static

Definition at line 1256 of file receivelog.c.

References feTimestampDifference(), sleeptime, and still_sending.

Referenced by HandleCopyStream().

1258 {
1259  TimestampTz status_targettime = 0;
1260  long sleeptime;
1261 
1263  status_targettime = last_status +
1264  (standby_message_timeout - 1) * ((int64) 1000);
1265 
1266  if (status_targettime > 0)
1267  {
1268  long secs;
1269  int usecs;
1270 
1272  status_targettime,
1273  &secs,
1274  &usecs);
1275  /* Always sleep at least 1 sec */
1276  if (secs <= 0)
1277  {
1278  secs = 1;
1279  usecs = 0;
1280  }
1281 
1282  sleeptime = secs * 1000 + usecs / 1000;
1283  }
1284  else
1285  sleeptime = -1;
1286 
1287  return sleeptime;
1288 }
static bool still_sending
Definition: receivelog.c:38
int64 TimestampTz
Definition: timestamp.h:39
int sleeptime
Definition: pg_standby.c:40
void feTimestampDifference(TimestampTz start_time, TimestampTz stop_time, long *secs, int *microsecs)
Definition: streamutil.c:459
static int standby_message_timeout
Definition: pg_basebackup.c:90
Datum now(PG_FUNCTION_ARGS)
Definition: timestamp.c:1533
static bool CheckCopyStreamStop ( PGconn conn,
StreamCtl stream,
XLogRecPtr  blockpos,
XLogRecPtr stoppos 
)
static

Definition at line 1230 of file receivelog.c.

References _, close_walfile(), NULL, PQerrorMessage(), PQflush(), PQputCopyEnd(), progname, still_sending, StreamCtl::stream_stop, and StreamCtl::timeline.

Referenced by HandleCopyStream().

1232 {
1233  if (still_sending && stream->stream_stop(blockpos, stream->timeline, false))
1234  {
1235  if (!close_walfile(stream, blockpos))
1236  {
1237  /* Potential error message is written by close_walfile */
1238  return false;
1239  }
1240  if (PQputCopyEnd(conn, NULL) <= 0 || PQflush(conn))
1241  {
1242  fprintf(stderr, _("%s: could not send copy-end packet: %s"),
1243  progname, PQerrorMessage(conn));
1244  return false;
1245  }
1246  still_sending = false;
1247  }
1248 
1249  return true;
1250 }
char * PQerrorMessage(const PGconn *conn)
Definition: fe-connect.c:6011
static bool still_sending
Definition: receivelog.c:38
int PQputCopyEnd(PGconn *conn, const char *errormsg)
Definition: fe-exec.c:2288
const char * progname
Definition: pg_standby.c:37
TimeLineID timeline
Definition: receivelog.h:33
int PQflush(PGconn *conn)
Definition: fe-exec.c:3187
static bool close_walfile(StreamCtl *stream, XLogRecPtr pos)
Definition: receivelog.c:178
stream_stop_callback stream_stop
Definition: receivelog.h:43
#define NULL
Definition: c.h:229
#define _(x)
Definition: elog.c:84
bool CheckServerVersionForStreaming ( PGconn conn)

Definition at line 360 of file receivelog.c.

References _, PQparameterStatus(), PQserverVersion(), and progname.

Referenced by BaseBackup(), ReceiveXlogStream(), and StreamLog().

361 {
362  int minServerMajor,
363  maxServerMajor;
364  int serverMajor;
365 
366  /*
367  * The message format used in streaming replication changed in 9.3, so we
368  * cannot stream from older servers. And we don't support servers newer
369  * than the client; it might work, but we don't know, so err on the safe
370  * side.
371  */
372  minServerMajor = 903;
373  maxServerMajor = PG_VERSION_NUM / 100;
374  serverMajor = PQserverVersion(conn) / 100;
375  if (serverMajor < minServerMajor)
376  {
377  const char *serverver = PQparameterStatus(conn, "server_version");
378 
379  fprintf(stderr, _("%s: incompatible server version %s; client does not support streaming from server versions older than %s\n"),
380  progname,
381  serverver ? serverver : "'unknown'",
382  "9.3");
383  return false;
384  }
385  else if (serverMajor > maxServerMajor)
386  {
387  const char *serverver = PQparameterStatus(conn, "server_version");
388 
389  fprintf(stderr, _("%s: incompatible server version %s; client does not support streaming from server versions newer than %s\n"),
390  progname,
391  serverver ? serverver : "'unknown'",
392  PG_VERSION);
393  return false;
394  }
395  return true;
396 }
const char * PQparameterStatus(const PGconn *conn, const char *paramName)
Definition: fe-connect.c:5976
int PQserverVersion(const PGconn *conn)
Definition: fe-connect.c:6001
const char * progname
Definition: pg_standby.c:37
#define _(x)
Definition: elog.c:84
static bool close_walfile ( StreamCtl stream,
XLogRecPtr  pos 
)
static

Definition at line 178 of file receivelog.c.

References _, WalWriteMethod::close, CLOSE_NO_RENAME, CLOSE_NORMAL, CLOSE_UNLINK, current_walfile_name, WalWriteMethod::get_current_pos, WalWriteMethod::getlasterror, lastFlushPosition, StreamCtl::mark_done, mark_file_as_archived(), NULL, StreamCtl::partial_suffix, progname, walfile, and StreamCtl::walmethod.

Referenced by CheckCopyStreamStop(), HandleEndOfCopyStream(), and ProcessXLogDataMsg().

179 {
180  off_t currpos;
181  int r;
182 
183  if (walfile == NULL)
184  return true;
185 
186  currpos = stream->walmethod->get_current_pos(walfile);
187  if (currpos == -1)
188  {
189  fprintf(stderr,
190  _("%s: could not determine seek position in file \"%s\": %s\n"),
192  stream->walmethod->close(walfile, CLOSE_UNLINK);
193  walfile = NULL;
194 
195  return false;
196  }
197 
198  if (stream->partial_suffix)
199  {
200  if (currpos == XLOG_SEG_SIZE)
201  r = stream->walmethod->close(walfile, CLOSE_NORMAL);
202  else
203  {
204  fprintf(stderr,
205  _("%s: not renaming \"%s%s\", segment is not complete\n"),
207  r = stream->walmethod->close(walfile, CLOSE_NO_RENAME);
208  }
209  }
210  else
211  r = stream->walmethod->close(walfile, CLOSE_NORMAL);
212 
213  walfile = NULL;
214 
215  if (r != 0)
216  {
217  fprintf(stderr, _("%s: could not close file \"%s\": %s\n"),
219  return false;
220  }
221 
222  /*
223  * Mark file as archived if requested by the caller - pg_basebackup needs
224  * to do so as files can otherwise get archived again after promotion of a
225  * new node. This is in line with walreceiver.c always doing a
226  * XLogArchiveForceDone() after a complete segment.
227  */
228  if (currpos == XLOG_SEG_SIZE && stream->mark_done)
229  {
230  /* writes error message if failed */
232  return false;
233  }
234 
235  lastFlushPosition = pos;
236  return true;
237 }
char *(* getlasterror)(void)
Definition: walmethods.h:34
char * partial_suffix
Definition: receivelog.h:46
const char * progname
Definition: pg_standby.c:37
static Walfile * walfile
Definition: receivelog.c:33
bool mark_done
Definition: receivelog.h:39
off_t(* get_current_pos)(Walfile f)
Definition: walmethods.h:31
WalWriteMethod * walmethod
Definition: receivelog.h:45
static XLogRecPtr lastFlushPosition
Definition: receivelog.c:36
#define NULL
Definition: c.h:229
static bool mark_file_as_archived(StreamCtl *stream, const char *fname)
Definition: receivelog.c:59
static char current_walfile_name[MAXPGPATH]
Definition: receivelog.c:34
#define _(x)
Definition: elog.c:84
int(* close)(Walfile f, WalCloseMethod method)
Definition: walmethods.h:26
static int CopyStreamPoll ( PGconn conn,
long  timeout_ms 
)
static

Definition at line 887 of file receivelog.c.

References _, EINTR, NULL, PQerrorMessage(), PQsocket(), progname, select, and strerror().

Referenced by CopyStreamReceive().

888 {
889  int ret;
890  fd_set input_mask;
891  struct timeval timeout;
892  struct timeval *timeoutptr;
893 
894  if (PQsocket(conn) < 0)
895  {
896  fprintf(stderr, _("%s: invalid socket: %s"), progname,
897  PQerrorMessage(conn));
898  return -1;
899  }
900 
901  FD_ZERO(&input_mask);
902  FD_SET(PQsocket(conn), &input_mask);
903 
904  if (timeout_ms < 0)
905  timeoutptr = NULL;
906  else
907  {
908  timeout.tv_sec = timeout_ms / 1000L;
909  timeout.tv_usec = (timeout_ms % 1000L) * 1000L;
910  timeoutptr = &timeout;
911  }
912 
913  ret = select(PQsocket(conn) + 1, &input_mask, NULL, NULL, timeoutptr);
914  if (ret == 0 || (ret < 0 && errno == EINTR))
915  return 0; /* Got a timeout or signal */
916  else if (ret < 0)
917  {
918  fprintf(stderr, _("%s: select() failed: %s\n"),
919  progname, strerror(errno));
920  return -1;
921  }
922 
923  return 1;
924 }
char * PQerrorMessage(const PGconn *conn)
Definition: fe-connect.c:6011
#define select(n, r, w, e, timeout)
Definition: win32.h:384
const char * progname
Definition: pg_standby.c:37
#define EINTR
Definition: win32.h:295
#define NULL
Definition: c.h:229
const char * strerror(int errnum)
Definition: strerror.c:19
#define _(x)
Definition: elog.c:84
int PQsocket(const PGconn *conn)
Definition: fe-connect.c:6029
static int CopyStreamReceive ( PGconn conn,
long  timeout,
char **  buffer 
)
static

Definition at line 938 of file receivelog.c.

References _, copybuf, CopyStreamPoll(), NULL, PQconsumeInput(), PQerrorMessage(), PQfreemem(), PQgetCopyData(), and progname.

Referenced by HandleCopyStream().

939 {
940  char *copybuf = NULL;
941  int rawlen;
942 
943  if (*buffer != NULL)
944  PQfreemem(*buffer);
945  *buffer = NULL;
946 
947  /* Try to receive a CopyData message */
948  rawlen = PQgetCopyData(conn, &copybuf, 1);
949  if (rawlen == 0)
950  {
951  /*
952  * No data available. Wait for some to appear, but not longer than the
953  * specified timeout, so that we can ping the server.
954  */
955  if (timeout != 0)
956  {
957  int ret;
958 
959  ret = CopyStreamPoll(conn, timeout);
960  if (ret <= 0)
961  return ret;
962  }
963 
964  /* Else there is actually data on the socket */
965  if (PQconsumeInput(conn) == 0)
966  {
967  fprintf(stderr,
968  _("%s: could not receive data from WAL stream: %s"),
969  progname, PQerrorMessage(conn));
970  return -1;
971  }
972 
973  /* Now that we've consumed some input, try again */
974  rawlen = PQgetCopyData(conn, &copybuf, 1);
975  if (rawlen == 0)
976  return 0;
977  }
978  if (rawlen == -1) /* end-of-streaming or error */
979  return -2;
980  if (rawlen == -2)
981  {
982  fprintf(stderr, _("%s: could not read COPY data: %s"),
983  progname, PQerrorMessage(conn));
984  return -1;
985  }
986 
987  /* Return received messages to caller */
988  *buffer = copybuf;
989  return rawlen;
990 }
char * PQerrorMessage(const PGconn *conn)
Definition: fe-connect.c:6011
const char * progname
Definition: pg_standby.c:37
int PQgetCopyData(PGconn *conn, char **buffer, int async)
Definition: fe-exec.c:2377
static int CopyStreamPoll(PGconn *conn, long timeout_ms)
Definition: receivelog.c:887
int PQconsumeInput(PGconn *conn)
Definition: fe-exec.c:1631
StringInfo copybuf
Definition: tablesync.c:109
#define NULL
Definition: c.h:229
WalTimeSample buffer[LAG_TRACKER_BUFFER_SIZE]
Definition: walsender.c:207
#define _(x)
Definition: elog.c:84
void PQfreemem(void *ptr)
Definition: fe-exec.c:3200
static bool existsTimeLineHistoryFile ( StreamCtl stream)
static

Definition at line 244 of file receivelog.c.

References WalWriteMethod::existsfile, MAXFNAMELEN, StreamCtl::timeline, TLHistoryFileName, and StreamCtl::walmethod.

Referenced by ReceiveXlogStream().

245 {
246  char histfname[MAXFNAMELEN];
247 
248  /*
249  * Timeline 1 never has a history file. We treat that as if it existed,
250  * since we never need to stream it.
251  */
252  if (stream->timeline == 1)
253  return true;
254 
255  TLHistoryFileName(histfname, stream->timeline);
256 
257  return stream->walmethod->existsfile(histfname);
258 }
bool(* existsfile)(const char *pathname)
Definition: walmethods.h:27
TimeLineID timeline
Definition: receivelog.h:33
#define TLHistoryFileName(fname, tli)
WalWriteMethod * walmethod
Definition: receivelog.h:45
#define MAXFNAMELEN
static PGresult * HandleCopyStream ( PGconn conn,
StreamCtl stream,
XLogRecPtr stoppos 
)
static

Definition at line 760 of file receivelog.c.

References _, CalculateCopyStreamSleeptime(), CheckCopyStreamStop(), copybuf, CopyStreamReceive(), current_walfile_name, error(), feGetCurrentTimestamp(), feTimestampDifferenceExceeds(), WalWriteMethod::getlasterror, HandleEndOfCopyStream(), lastFlushPosition, now(), NULL, PQfreemem(), ProcessKeepaliveMsg(), ProcessXLogDataMsg(), progname, sendFeedback(), sleeptime, StreamCtl::standby_message_timeout, StreamCtl::startpos, still_sending, WalWriteMethod::sync, StreamCtl::synchronous, walfile, and StreamCtl::walmethod.

Referenced by ReceiveXlogStream().

762 {
763  char *copybuf = NULL;
764  TimestampTz last_status = -1;
765  XLogRecPtr blockpos = stream->startpos;
766 
767  still_sending = true;
768 
769  while (1)
770  {
771  int r;
773  long sleeptime;
774 
775  /*
776  * Check if we should continue streaming, or abort at this point.
777  */
778  if (!CheckCopyStreamStop(conn, stream, blockpos, stoppos))
779  goto error;
780 
781  now = feGetCurrentTimestamp();
782 
783  /*
784  * If synchronous option is true, issue sync command as soon as there
785  * are WAL data which has not been flushed yet.
786  */
787  if (stream->synchronous && lastFlushPosition < blockpos && walfile != NULL)
788  {
789  if (stream->walmethod->sync(walfile) != 0)
790  {
791  fprintf(stderr, _("%s: could not fsync file \"%s\": %s\n"),
793  goto error;
794  }
795  lastFlushPosition = blockpos;
796 
797  /*
798  * Send feedback so that the server sees the latest WAL locations
799  * immediately.
800  */
801  if (!sendFeedback(conn, blockpos, now, false))
802  goto error;
803  last_status = now;
804  }
805 
806  /*
807  * Potentially send a status message to the master
808  */
809  if (still_sending && stream->standby_message_timeout > 0 &&
810  feTimestampDifferenceExceeds(last_status, now,
811  stream->standby_message_timeout))
812  {
813  /* Time to send feedback! */
814  if (!sendFeedback(conn, blockpos, now, false))
815  goto error;
816  last_status = now;
817  }
818 
819  /*
820  * Calculate how long send/receive loops should sleep
821  */
822  sleeptime = CalculateCopyStreamSleeptime(now, stream->standby_message_timeout,
823  last_status);
824 
825  r = CopyStreamReceive(conn, sleeptime, &copybuf);
826  while (r != 0)
827  {
828  if (r == -1)
829  goto error;
830  if (r == -2)
831  {
832  PGresult *res = HandleEndOfCopyStream(conn, stream, copybuf, blockpos, stoppos);
833 
834  if (res == NULL)
835  goto error;
836  else
837  return res;
838  }
839 
840  /* Check the message type. */
841  if (copybuf[0] == 'k')
842  {
843  if (!ProcessKeepaliveMsg(conn, stream, copybuf, r, blockpos,
844  &last_status))
845  goto error;
846  }
847  else if (copybuf[0] == 'w')
848  {
849  if (!ProcessXLogDataMsg(conn, stream, copybuf, r, &blockpos))
850  goto error;
851 
852  /*
853  * Check if we should continue streaming, or abort at this
854  * point.
855  */
856  if (!CheckCopyStreamStop(conn, stream, blockpos, stoppos))
857  goto error;
858  }
859  else
860  {
861  fprintf(stderr, _("%s: unrecognized streaming header: \"%c\"\n"),
862  progname, copybuf[0]);
863  goto error;
864  }
865 
866  /*
867  * Process the received data, and any subsequent data we can read
868  * without blocking.
869  */
870  r = CopyStreamReceive(conn, 0, &copybuf);
871  }
872  }
873 
874 error:
875  if (copybuf != NULL)
876  PQfreemem(copybuf);
877  return NULL;
878 }
static int CopyStreamReceive(PGconn *conn, long timeout, char **buffer)
Definition: receivelog.c:938
char *(* getlasterror)(void)
Definition: walmethods.h:34
TimestampTz feGetCurrentTimestamp(void)
Definition: streamutil.c:440
static void error(void)
Definition: sql-dyntest.c:147
static bool still_sending
Definition: receivelog.c:38
int64 TimestampTz
Definition: timestamp.h:39
static bool CheckCopyStreamStop(PGconn *conn, StreamCtl *stream, XLogRecPtr blockpos, XLogRecPtr *stoppos)
Definition: receivelog.c:1230
XLogRecPtr startpos
Definition: receivelog.h:32
static long CalculateCopyStreamSleeptime(TimestampTz now, int standby_message_timeout, TimestampTz last_status)
Definition: receivelog.c:1256
int sleeptime
Definition: pg_standby.c:40
const char * progname
Definition: pg_standby.c:37
bool feTimestampDifferenceExceeds(TimestampTz start_time, TimestampTz stop_time, int msec)
Definition: streamutil.c:481
static bool ProcessKeepaliveMsg(PGconn *conn, StreamCtl *stream, char *copybuf, int len, XLogRecPtr blockpos, TimestampTz *last_status)
Definition: receivelog.c:996
static Walfile * walfile
Definition: receivelog.c:33
WalWriteMethod * walmethod
Definition: receivelog.h:45
int(* sync)(Walfile f)
Definition: walmethods.h:32
static bool sendFeedback(PGconn *conn, XLogRecPtr blockpos, TimestampTz now, bool replyRequested)
Definition: receivelog.c:322
static XLogRecPtr lastFlushPosition
Definition: receivelog.c:36
static PGresult * HandleEndOfCopyStream(PGconn *conn, StreamCtl *stream, char *copybuf, XLogRecPtr blockpos, XLogRecPtr *stoppos)
Definition: receivelog.c:1188
static bool ProcessXLogDataMsg(PGconn *conn, StreamCtl *stream, char *copybuf, int len, XLogRecPtr *blockpos)
Definition: receivelog.c:1054
StringInfo copybuf
Definition: tablesync.c:109
#define NULL
Definition: c.h:229
uint64 XLogRecPtr
Definition: xlogdefs.h:21
bool synchronous
Definition: receivelog.h:38
int standby_message_timeout
Definition: receivelog.h:36
static char current_walfile_name[MAXPGPATH]
Definition: receivelog.c:34
#define _(x)
Definition: elog.c:84
void PQfreemem(void *ptr)
Definition: fe-exec.c:3200
Datum now(PG_FUNCTION_ARGS)
Definition: timestamp.c:1533
static PGresult * HandleEndOfCopyStream ( PGconn conn,
StreamCtl stream,
char *  copybuf,
XLogRecPtr  blockpos,
XLogRecPtr stoppos 
)
static

Definition at line 1188 of file receivelog.c.

References _, close_walfile(), NULL, PGRES_COPY_IN, PQclear(), PQerrorMessage(), PQflush(), PQfreemem(), PQgetResult(), PQputCopyEnd(), PQresultStatus(), progname, and still_sending.

Referenced by HandleCopyStream().

1190 {
1191  PGresult *res = PQgetResult(conn);
1192 
1193  /*
1194  * The server closed its end of the copy stream. If we haven't closed
1195  * ours already, we need to do so now, unless the server threw an error,
1196  * in which case we don't.
1197  */
1198  if (still_sending)
1199  {
1200  if (!close_walfile(stream, blockpos))
1201  {
1202  /* Error message written in close_walfile() */
1203  PQclear(res);
1204  return NULL;
1205  }
1206  if (PQresultStatus(res) == PGRES_COPY_IN)
1207  {
1208  if (PQputCopyEnd(conn, NULL) <= 0 || PQflush(conn))
1209  {
1210  fprintf(stderr,
1211  _("%s: could not send copy-end packet: %s"),
1212  progname, PQerrorMessage(conn));
1213  PQclear(res);
1214  return NULL;
1215  }
1216  res = PQgetResult(conn);
1217  }
1218  still_sending = false;
1219  }
1220  if (copybuf != NULL)
1221  PQfreemem(copybuf);
1222  *stoppos = blockpos;
1223  return res;
1224 }
char * PQerrorMessage(const PGconn *conn)
Definition: fe-connect.c:6011
static bool still_sending
Definition: receivelog.c:38
int PQputCopyEnd(PGconn *conn, const char *errormsg)
Definition: fe-exec.c:2288
const char * progname
Definition: pg_standby.c:37
ExecStatusType PQresultStatus(const PGresult *res)
Definition: fe-exec.c:2596
int PQflush(PGconn *conn)
Definition: fe-exec.c:3187
static bool close_walfile(StreamCtl *stream, XLogRecPtr pos)
Definition: receivelog.c:178
StringInfo copybuf
Definition: tablesync.c:109
void PQclear(PGresult *res)
Definition: fe-exec.c:650
#define NULL
Definition: c.h:229
#define _(x)
Definition: elog.c:84
void PQfreemem(void *ptr)
Definition: fe-exec.c:3200
PGresult * PQgetResult(PGconn *conn)
Definition: fe-exec.c:1702
static bool mark_file_as_archived ( StreamCtl stream,
const char *  fname 
)
static

Definition at line 59 of file receivelog.c.

References _, WalWriteMethod::close, CLOSE_NORMAL, WalWriteMethod::getlasterror, MAXPGPATH, NULL, WalWriteMethod::open_for_write, progname, snprintf(), and StreamCtl::walmethod.

Referenced by close_walfile(), and writeTimeLineHistoryFile().

60 {
61  Walfile *f;
62  static char tmppath[MAXPGPATH];
63 
64  snprintf(tmppath, sizeof(tmppath), "archive_status/%s.done",
65  fname);
66 
67  f = stream->walmethod->open_for_write(tmppath, NULL, 0);
68  if (f == NULL)
69  {
70  fprintf(stderr, _("%s: could not create archive status file \"%s\": %s\n"),
71  progname, tmppath, stream->walmethod->getlasterror());
72  return false;
73  }
74 
75  stream->walmethod->close(f, CLOSE_NORMAL);
76 
77  return true;
78 }
char *(* getlasterror)(void)
Definition: walmethods.h:34
void * Walfile
Definition: walmethods.h:13
int snprintf(char *str, size_t count, const char *fmt,...) pg_attribute_printf(3
const char * progname
Definition: pg_standby.c:37
#define MAXPGPATH
WalWriteMethod * walmethod
Definition: receivelog.h:45
#define NULL
Definition: c.h:229
Walfile(* open_for_write)(const char *pathname, const char *temp_suffix, size_t pad_to_size)
Definition: walmethods.h:25
#define _(x)
Definition: elog.c:84
int(* close)(Walfile f, WalCloseMethod method)
Definition: walmethods.h:26
static bool open_walfile ( StreamCtl stream,
XLogRecPtr  startpoint 
)
static

Definition at line 90 of file receivelog.c.

References _, WalWriteMethod::close, CLOSE_UNLINK, current_walfile_name, WalWriteMethod::existsfile, fn(), WalWriteMethod::get_file_size, WalWriteMethod::getlasterror, MAXPGPATH, NULL, WalWriteMethod::open_for_write, StreamCtl::partial_suffix, progname, snprintf(), WalWriteMethod::sync, StreamCtl::timeline, walfile, StreamCtl::walmethod, XLByteToSeg, XLogFileName, and XLogSegSize.

Referenced by ProcessXLogDataMsg().

91 {
92  Walfile *f;
93  char fn[MAXPGPATH];
94  ssize_t size;
95  XLogSegNo segno;
96 
97  XLByteToSeg(startpoint, segno);
99 
100  snprintf(fn, sizeof(fn), "%s%s", current_walfile_name,
101  stream->partial_suffix ? stream->partial_suffix : "");
102 
103  /*
104  * When streaming to files, if an existing file exists we verify that it's
105  * either empty (just created), or a complete XLogSegSize segment (in
106  * which case it has been created and padded). Anything else indicates a
107  * corrupt file.
108  *
109  * When streaming to tar, no file with this name will exist before, so we
110  * never have to verify a size.
111  */
112  if (stream->walmethod->existsfile(fn))
113  {
114  size = stream->walmethod->get_file_size(fn);
115  if (size < 0)
116  {
117  fprintf(stderr,
118  _("%s: could not get size of transaction log file \"%s\": %s\n"),
119  progname, fn, stream->walmethod->getlasterror());
120  return false;
121  }
122  if (size == XLogSegSize)
123  {
124  /* Already padded file. Open it for use */
126  if (f == NULL)
127  {
128  fprintf(stderr,
129  _("%s: could not open existing transaction log file \"%s\": %s\n"),
130  progname, fn, stream->walmethod->getlasterror());
131  return false;
132  }
133 
134  /* fsync file in case of a previous crash */
135  if (!stream->walmethod->sync(f))
136  {
137  stream->walmethod->close(f, CLOSE_UNLINK);
138  return false;
139  }
140 
141  walfile = f;
142  return true;
143  }
144  if (size != 0)
145  {
146  /* if write didn't set errno, assume problem is no disk space */
147  if (errno == 0)
148  errno = ENOSPC;
149  fprintf(stderr,
150  _("%s: transaction log file \"%s\" has %d bytes, should be 0 or %d\n"),
151  progname, fn, (int) size, XLogSegSize);
152  return false;
153  }
154  /* File existed and was empty, so fall through and open */
155  }
156 
157  /* No file existed, so create one */
158 
160  if (f == NULL)
161  {
162  fprintf(stderr,
163  _("%s: could not open transaction log file \"%s\": %s\n"),
164  progname, fn, stream->walmethod->getlasterror());
165  return false;
166  }
167 
168  walfile = f;
169  return true;
170 }
#define XLogSegSize
Definition: xlog_internal.h:92
char *(* getlasterror)(void)
Definition: walmethods.h:34
void * Walfile
Definition: walmethods.h:13
bool(* existsfile)(const char *pathname)
Definition: walmethods.h:27
#define XLogFileName(fname, tli, logSegNo)
int snprintf(char *str, size_t count, const char *fmt,...) pg_attribute_printf(3
char * partial_suffix
Definition: receivelog.h:46
const char * progname
Definition: pg_standby.c:37
TimeLineID timeline
Definition: receivelog.h:33
ssize_t(* get_file_size)(const char *pathname)
Definition: walmethods.h:28
#define MAXPGPATH
static Walfile * walfile
Definition: receivelog.c:33
uint64 XLogSegNo
Definition: xlogdefs.h:34
WalWriteMethod * walmethod
Definition: receivelog.h:45
int(* sync)(Walfile f)
Definition: walmethods.h:32
static void * fn(void *arg)
#define XLByteToSeg(xlrp, logSegNo)
#define NULL
Definition: c.h:229
Walfile(* open_for_write)(const char *pathname, const char *temp_suffix, size_t pad_to_size)
Definition: walmethods.h:25
static char current_walfile_name[MAXPGPATH]
Definition: receivelog.c:34
#define _(x)
Definition: elog.c:84
int(* close)(Walfile f, WalCloseMethod method)
Definition: walmethods.h:26
static bool ProcessKeepaliveMsg ( PGconn conn,
StreamCtl stream,
char *  copybuf,
int  len,
XLogRecPtr  blockpos,
TimestampTz last_status 
)
static

Definition at line 996 of file receivelog.c.

References _, current_walfile_name, feGetCurrentTimestamp(), WalWriteMethod::getlasterror, lastFlushPosition, now(), NULL, progname, reportFlushPosition, sendFeedback(), still_sending, WalWriteMethod::sync, walfile, and StreamCtl::walmethod.

Referenced by HandleCopyStream().

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  */
1007  pos = 1; /* skip msgtype 'k' */
1008  pos += 8; /* skip walEnd */
1009  pos += 8; /* skip sendTime */
1010 
1011  if (len < pos + 1)
1012  {
1013  fprintf(stderr, _("%s: streaming header too small: %d\n"),
1014  progname, len);
1015  return false;
1016  }
1017  replyRequested = copybuf[pos];
1018 
1019  /* If the server requested an immediate reply, send one. */
1020  if (replyRequested && still_sending)
1021  {
1022  if (reportFlushPosition && lastFlushPosition < blockpos &&
1023  walfile != NULL)
1024  {
1025  /*
1026  * If a valid flush location needs to be reported, flush the
1027  * current WAL file so that the latest flush location is sent back
1028  * to the server. This is necessary to see whether the last WAL
1029  * data has been successfully replicated or not, at the normal
1030  * shutdown of the server.
1031  */
1032  if (stream->walmethod->sync(walfile) != 0)
1033  {
1034  fprintf(stderr, _("%s: could not fsync file \"%s\": %s\n"),
1036  return false;
1037  }
1038  lastFlushPosition = blockpos;
1039  }
1040 
1041  now = feGetCurrentTimestamp();
1042  if (!sendFeedback(conn, blockpos, now, false))
1043  return false;
1044  *last_status = now;
1045  }
1046 
1047  return true;
1048 }
char *(* getlasterror)(void)
Definition: walmethods.h:34
TimestampTz feGetCurrentTimestamp(void)
Definition: streamutil.c:440
static bool still_sending
Definition: receivelog.c:38
int64 TimestampTz
Definition: timestamp.h:39
const char * progname
Definition: pg_standby.c:37
static Walfile * walfile
Definition: receivelog.c:33
static bool reportFlushPosition
Definition: receivelog.c:35
WalWriteMethod * walmethod
Definition: receivelog.h:45
int(* sync)(Walfile f)
Definition: walmethods.h:32
static bool sendFeedback(PGconn *conn, XLogRecPtr blockpos, TimestampTz now, bool replyRequested)
Definition: receivelog.c:322
static XLogRecPtr lastFlushPosition
Definition: receivelog.c:36
StringInfo copybuf
Definition: tablesync.c:109
#define NULL
Definition: c.h:229
static char current_walfile_name[MAXPGPATH]
Definition: receivelog.c:34
#define _(x)
Definition: elog.c:84
Datum now(PG_FUNCTION_ARGS)
Definition: timestamp.c:1533
static bool ProcessXLogDataMsg ( PGconn conn,
StreamCtl stream,
char *  copybuf,
int  len,
XLogRecPtr blockpos 
)
static

Definition at line 1054 of file receivelog.c.

References _, close_walfile(), current_walfile_name, fe_recvint64(), WalWriteMethod::get_current_pos, WalWriteMethod::getlasterror, NULL, open_walfile(), PQerrorMessage(), PQflush(), PQputCopyEnd(), progname, still_sending, StreamCtl::stream_stop, StreamCtl::timeline, walfile, StreamCtl::walmethod, and WalWriteMethod::write.

Referenced by HandleCopyStream().

1056 {
1057  int xlogoff;
1058  int bytes_left;
1059  int bytes_written;
1060  int hdr_len;
1061 
1062  /*
1063  * Once we've decided we don't want to receive any more, just ignore any
1064  * subsequent XLogData messages.
1065  */
1066  if (!(still_sending))
1067  return true;
1068 
1069  /*
1070  * Read the header of the XLogData message, enclosed in the CopyData
1071  * message. We only need the WAL location field (dataStart), the rest of
1072  * the header is ignored.
1073  */
1074  hdr_len = 1; /* msgtype 'w' */
1075  hdr_len += 8; /* dataStart */
1076  hdr_len += 8; /* walEnd */
1077  hdr_len += 8; /* sendTime */
1078  if (len < hdr_len)
1079  {
1080  fprintf(stderr, _("%s: streaming header too small: %d\n"),
1081  progname, len);
1082  return false;
1083  }
1084  *blockpos = fe_recvint64(&copybuf[1]);
1085 
1086  /* Extract WAL location for this block */
1087  xlogoff = *blockpos % XLOG_SEG_SIZE;
1088 
1089  /*
1090  * Verify that the initial location in the stream matches where we think
1091  * we are.
1092  */
1093  if (walfile == NULL)
1094  {
1095  /* No file open yet */
1096  if (xlogoff != 0)
1097  {
1098  fprintf(stderr,
1099  _("%s: received transaction log record for offset %u with no file open\n"),
1100  progname, xlogoff);
1101  return false;
1102  }
1103  }
1104  else
1105  {
1106  /* More data in existing segment */
1107  if (stream->walmethod->get_current_pos(walfile) != xlogoff)
1108  {
1109  fprintf(stderr,
1110  _("%s: got WAL data offset %08x, expected %08x\n"),
1111  progname, xlogoff, (int) stream->walmethod->get_current_pos(walfile));
1112  return false;
1113  }
1114  }
1115 
1116  bytes_left = len - hdr_len;
1117  bytes_written = 0;
1118 
1119  while (bytes_left)
1120  {
1121  int bytes_to_write;
1122 
1123  /*
1124  * If crossing a WAL boundary, only write up until we reach
1125  * XLOG_SEG_SIZE.
1126  */
1127  if (xlogoff + bytes_left > XLOG_SEG_SIZE)
1128  bytes_to_write = XLOG_SEG_SIZE - xlogoff;
1129  else
1130  bytes_to_write = bytes_left;
1131 
1132  if (walfile == NULL)
1133  {
1134  if (!open_walfile(stream, *blockpos))
1135  {
1136  /* Error logged by open_walfile */
1137  return false;
1138  }
1139  }
1140 
1141  if (stream->walmethod->write(walfile, copybuf + hdr_len + bytes_written,
1142  bytes_to_write) != bytes_to_write)
1143  {
1144  fprintf(stderr,
1145  _("%s: could not write %u bytes to WAL file \"%s\": %s\n"),
1146  progname, bytes_to_write, current_walfile_name,
1147  stream->walmethod->getlasterror());
1148  return false;
1149  }
1150 
1151  /* Write was successful, advance our position */
1152  bytes_written += bytes_to_write;
1153  bytes_left -= bytes_to_write;
1154  *blockpos += bytes_to_write;
1155  xlogoff += bytes_to_write;
1156 
1157  /* Did we reach the end of a WAL segment? */
1158  if (*blockpos % XLOG_SEG_SIZE == 0)
1159  {
1160  if (!close_walfile(stream, *blockpos))
1161  /* Error message written in close_walfile() */
1162  return false;
1163 
1164  xlogoff = 0;
1165 
1166  if (still_sending && stream->stream_stop(*blockpos, stream->timeline, true))
1167  {
1168  if (PQputCopyEnd(conn, NULL) <= 0 || PQflush(conn))
1169  {
1170  fprintf(stderr, _("%s: could not send copy-end packet: %s"),
1171  progname, PQerrorMessage(conn));
1172  return false;
1173  }
1174  still_sending = false;
1175  return true; /* ignore the rest of this XLogData packet */
1176  }
1177  }
1178  }
1179  /* No more data left to write, receive next copy packet */
1180 
1181  return true;
1182 }
static bool open_walfile(StreamCtl *stream, XLogRecPtr startpoint)
Definition: receivelog.c:90
char * PQerrorMessage(const PGconn *conn)
Definition: fe-connect.c:6011
char *(* getlasterror)(void)
Definition: walmethods.h:34
static bool still_sending
Definition: receivelog.c:38
int PQputCopyEnd(PGconn *conn, const char *errormsg)
Definition: fe-exec.c:2288
const char * progname
Definition: pg_standby.c:37
TimeLineID timeline
Definition: receivelog.h:33
static Walfile * walfile
Definition: receivelog.c:33
int PQflush(PGconn *conn)
Definition: fe-exec.c:3187
off_t(* get_current_pos)(Walfile f)
Definition: walmethods.h:31
static bool close_walfile(StreamCtl *stream, XLogRecPtr pos)
Definition: receivelog.c:178
stream_stop_callback stream_stop
Definition: receivelog.h:43
WalWriteMethod * walmethod
Definition: receivelog.h:45
StringInfo copybuf
Definition: tablesync.c:109
ssize_t(* write)(Walfile f, const void *buf, size_t count)
Definition: walmethods.h:30
#define NULL
Definition: c.h:229
int64 fe_recvint64(char *buf)
Definition: streamutil.c:513
static char current_walfile_name[MAXPGPATH]
Definition: receivelog.c:34
#define _(x)
Definition: elog.c:84
static bool ReadEndOfStreamingResult ( PGresult res,
XLogRecPtr startpos,
uint32 timeline 
)
static

Definition at line 712 of file receivelog.c.

References _, PQgetvalue(), PQnfields(), PQntuples(), and progname.

Referenced by ReceiveXlogStream().

713 {
714  uint32 startpos_xlogid,
715  startpos_xrecoff;
716 
717  /*----------
718  * The result set consists of one row and two columns, e.g:
719  *
720  * next_tli | next_tli_startpos
721  * ----------+-------------------
722  * 4 | 0/9949AE0
723  *
724  * next_tli is the timeline ID of the next timeline after the one that
725  * just finished streaming. next_tli_startpos is the XLOG position where
726  * the server switched to it.
727  *----------
728  */
729  if (PQnfields(res) < 2 || PQntuples(res) != 1)
730  {
731  fprintf(stderr,
732  _("%s: unexpected result set after end-of-timeline: got %d rows and %d fields, expected %d rows and %d fields\n"),
733  progname, PQntuples(res), PQnfields(res), 1, 2);
734  return false;
735  }
736 
737  *timeline = atoi(PQgetvalue(res, 0, 0));
738  if (sscanf(PQgetvalue(res, 0, 1), "%X/%X", &startpos_xlogid,
739  &startpos_xrecoff) != 2)
740  {
741  fprintf(stderr,
742  _("%s: could not parse next timeline's starting point \"%s\"\n"),
743  progname, PQgetvalue(res, 0, 1));
744  return false;
745  }
746  *startpos = ((uint64) startpos_xlogid << 32) | startpos_xrecoff;
747 
748  return true;
749 }
int PQnfields(const PGresult *res)
Definition: fe-exec.c:2681
char * PQgetvalue(const PGresult *res, int tup_num, int field_num)
Definition: fe-exec.c:3067
const char * progname
Definition: pg_standby.c:37
int PQntuples(const PGresult *res)
Definition: fe-exec.c:2673
unsigned int uint32
Definition: c.h:268
static XLogRecPtr startpos
#define _(x)
Definition: elog.c:84
bool ReceiveXlogStream ( PGconn conn,
StreamCtl stream 
)

Definition at line 433 of file receivelog.c.

References _, CheckServerVersionForStreaming(), WalWriteMethod::close, CLOSE_NO_RENAME, current_walfile_name, error(), existsTimeLineHistoryFile(), WalWriteMethod::getlasterror, HandleCopyStream(), lastFlushPosition, NULL, PGRES_COMMAND_OK, PGRES_COPY_BOTH, PGRES_TUPLES_OK, PQclear(), PQerrorMessage(), PQexec(), PQgetResult(), PQgetvalue(), PQnfields(), PQntuples(), PQresultErrorMessage(), PQresultStatus(), progname, ReadEndOfStreamingResult(), StreamCtl::replication_slot, reportFlushPosition, snprintf(), StreamCtl::startpos, StreamCtl::stream_stop, StreamCtl::synchronous, StreamCtl::sysidentifier, StreamCtl::temp_slot, StreamCtl::timeline, walfile, StreamCtl::walmethod, and writeTimeLineHistoryFile().

Referenced by LogStreamerMain(), and StreamLog().

434 {
435  char query[128];
436  char slotcmd[128];
437  PGresult *res;
438  XLogRecPtr stoppos;
439 
440  /*
441  * The caller should've checked the server version already, but doesn't do
442  * any harm to check it here too.
443  */
445  return false;
446 
447  /*
448  * Decide whether we want to report the flush position. If we report
449  * the flush position, the primary will know what WAL we'll
450  * possibly re-request, and it can then remove older WAL safely.
451  * We must always do that when we are using slots.
452  *
453  * Reporting the flush position makes one eligible as a synchronous
454  * replica. People shouldn't include generic names in
455  * synchronous_standby_names, but we've protected them against it so
456  * far, so let's continue to do so unless specifically requested.
457  */
458  if (stream->replication_slot != NULL)
459  {
460  reportFlushPosition = true;
461  sprintf(slotcmd, "SLOT \"%s\" ", stream->replication_slot);
462  }
463  else
464  {
465  if (stream->synchronous)
466  reportFlushPosition = true;
467  else
468  reportFlushPosition = false;
469  slotcmd[0] = 0;
470  }
471 
472  if (stream->sysidentifier != NULL)
473  {
474  /* Validate system identifier hasn't changed */
475  res = PQexec(conn, "IDENTIFY_SYSTEM");
476  if (PQresultStatus(res) != PGRES_TUPLES_OK)
477  {
478  fprintf(stderr,
479  _("%s: could not send replication command \"%s\": %s"),
480  progname, "IDENTIFY_SYSTEM", PQerrorMessage(conn));
481  PQclear(res);
482  return false;
483  }
484  if (PQntuples(res) != 1 || PQnfields(res) < 3)
485  {
486  fprintf(stderr,
487  _("%s: could not identify system: got %d rows and %d fields, expected %d rows and %d or more fields\n"),
488  progname, PQntuples(res), PQnfields(res), 1, 3);
489  PQclear(res);
490  return false;
491  }
492  if (strcmp(stream->sysidentifier, PQgetvalue(res, 0, 0)) != 0)
493  {
494  fprintf(stderr,
495  _("%s: system identifier does not match between base backup and streaming connection\n"),
496  progname);
497  PQclear(res);
498  return false;
499  }
500  if (stream->timeline > atoi(PQgetvalue(res, 0, 1)))
501  {
502  fprintf(stderr,
503  _("%s: starting timeline %u is not present in the server\n"),
504  progname, stream->timeline);
505  PQclear(res);
506  return false;
507  }
508  PQclear(res);
509  }
510 
511  /*
512  * Create temporary replication slot if one is needed
513  */
514  if (stream->temp_slot)
515  {
516  snprintf(query, sizeof(query),
517  "CREATE_REPLICATION_SLOT \"%s\" TEMPORARY PHYSICAL RESERVE_WAL",
518  stream->replication_slot);
519  res = PQexec(conn, query);
520  if (PQresultStatus(res) != PGRES_TUPLES_OK)
521  {
522  fprintf(stderr, _("%s: could not create temporary replication slot \"%s\": %s"),
523  progname, stream->replication_slot, PQerrorMessage(conn));
524  PQclear(res);
525  return false;
526  }
527  }
528 
529  /*
530  * initialize flush position to starting point, it's the caller's
531  * responsibility that that's sane.
532  */
533  lastFlushPosition = stream->startpos;
534 
535  while (1)
536  {
537  /*
538  * Fetch the timeline history file for this timeline, if we don't have
539  * it already. When streaming log to tar, this will always return
540  * false, as we are never streaming into an existing file and
541  * therefore there can be no pre-existing timeline history file.
542  */
543  if (!existsTimeLineHistoryFile(stream))
544  {
545  snprintf(query, sizeof(query), "TIMELINE_HISTORY %u", stream->timeline);
546  res = PQexec(conn, query);
547  if (PQresultStatus(res) != PGRES_TUPLES_OK)
548  {
549  /* FIXME: we might send it ok, but get an error */
550  fprintf(stderr, _("%s: could not send replication command \"%s\": %s"),
551  progname, "TIMELINE_HISTORY", PQresultErrorMessage(res));
552  PQclear(res);
553  return false;
554  }
555 
556  /*
557  * The response to TIMELINE_HISTORY is a single row result set
558  * with two fields: filename and content
559  */
560  if (PQnfields(res) != 2 || PQntuples(res) != 1)
561  {
562  fprintf(stderr,
563  _("%s: unexpected response to TIMELINE_HISTORY command: got %d rows and %d fields, expected %d rows and %d fields\n"),
564  progname, PQntuples(res), PQnfields(res), 1, 2);
565  }
566 
567  /* Write the history file to disk */
569  PQgetvalue(res, 0, 0),
570  PQgetvalue(res, 0, 1));
571 
572  PQclear(res);
573  }
574 
575  /*
576  * Before we start streaming from the requested location, check if the
577  * callback tells us to stop here.
578  */
579  if (stream->stream_stop(stream->startpos, stream->timeline, false))
580  return true;
581 
582  /* Initiate the replication stream at specified location */
583  snprintf(query, sizeof(query), "START_REPLICATION %s%X/%X TIMELINE %u",
584  slotcmd,
585  (uint32) (stream->startpos >> 32), (uint32) stream->startpos,
586  stream->timeline);
587  res = PQexec(conn, query);
588  if (PQresultStatus(res) != PGRES_COPY_BOTH)
589  {
590  fprintf(stderr, _("%s: could not send replication command \"%s\": %s"),
591  progname, "START_REPLICATION", PQresultErrorMessage(res));
592  PQclear(res);
593  return false;
594  }
595  PQclear(res);
596 
597  /* Stream the WAL */
598  res = HandleCopyStream(conn, stream, &stoppos);
599  if (res == NULL)
600  goto error;
601 
602  /*
603  * Streaming finished.
604  *
605  * There are two possible reasons for that: a controlled shutdown, or
606  * we reached the end of the current timeline. In case of
607  * end-of-timeline, the server sends a result set after Copy has
608  * finished, containing information about the next timeline. Read
609  * that, and restart streaming from the next timeline. In case of
610  * controlled shutdown, stop here.
611  */
612  if (PQresultStatus(res) == PGRES_TUPLES_OK)
613  {
614  /*
615  * End-of-timeline. Read the next timeline's ID and starting
616  * position. Usually, the starting position will match the end of
617  * the previous timeline, but there are corner cases like if the
618  * server had sent us half of a WAL record, when it was promoted.
619  * The new timeline will begin at the end of the last complete
620  * record in that case, overlapping the partial WAL record on the
621  * the old timeline.
622  */
623  uint32 newtimeline;
624  bool parsed;
625 
626  parsed = ReadEndOfStreamingResult(res, &stream->startpos, &newtimeline);
627  PQclear(res);
628  if (!parsed)
629  goto error;
630 
631  /* Sanity check the values the server gave us */
632  if (newtimeline <= stream->timeline)
633  {
634  fprintf(stderr,
635  _("%s: server reported unexpected next timeline %u, following timeline %u\n"),
636  progname, newtimeline, stream->timeline);
637  goto error;
638  }
639  if (stream->startpos > stoppos)
640  {
641  fprintf(stderr,
642  _("%s: server stopped streaming timeline %u at %X/%X, but reported next timeline %u to begin at %X/%X\n"),
643  progname,
644  stream->timeline, (uint32) (stoppos >> 32), (uint32) stoppos,
645  newtimeline, (uint32) (stream->startpos >> 32), (uint32) stream->startpos);
646  goto error;
647  }
648 
649  /* Read the final result, which should be CommandComplete. */
650  res = PQgetResult(conn);
651  if (PQresultStatus(res) != PGRES_COMMAND_OK)
652  {
653  fprintf(stderr,
654  _("%s: unexpected termination of replication stream: %s"),
656  PQclear(res);
657  goto error;
658  }
659  PQclear(res);
660 
661  /*
662  * Loop back to start streaming from the new timeline. Always
663  * start streaming at the beginning of a segment.
664  */
665  stream->timeline = newtimeline;
666  stream->startpos = stream->startpos - (stream->startpos % XLOG_SEG_SIZE);
667  continue;
668  }
669  else if (PQresultStatus(res) == PGRES_COMMAND_OK)
670  {
671  PQclear(res);
672 
673  /*
674  * End of replication (ie. controlled shut down of the server).
675  *
676  * Check if the callback thinks it's OK to stop here. If not,
677  * complain.
678  */
679  if (stream->stream_stop(stoppos, stream->timeline, false))
680  return true;
681  else
682  {
683  fprintf(stderr, _("%s: replication stream was terminated before stop point\n"),
684  progname);
685  goto error;
686  }
687  }
688  else
689  {
690  /* Server returned an error. */
691  fprintf(stderr,
692  _("%s: unexpected termination of replication stream: %s"),
694  PQclear(res);
695  goto error;
696  }
697  }
698 
699 error:
700  if (walfile != NULL && stream->walmethod->close(walfile, CLOSE_NO_RENAME) != 0)
701  fprintf(stderr, _("%s: could not close file \"%s\": %s\n"),
703  walfile = NULL;
704  return false;
705 }
int PQnfields(const PGresult *res)
Definition: fe-exec.c:2681
char * PQerrorMessage(const PGconn *conn)
Definition: fe-connect.c:6011
char *(* getlasterror)(void)
Definition: walmethods.h:34
char * PQgetvalue(const PGresult *res, int tup_num, int field_num)
Definition: fe-exec.c:3067
static void error(void)
Definition: sql-dyntest.c:147
char * sysidentifier
Definition: receivelog.h:34
int snprintf(char *str, size_t count, const char *fmt,...) pg_attribute_printf(3
XLogRecPtr startpos
Definition: receivelog.h:32
const char * progname
Definition: pg_standby.c:37
int PQntuples(const PGresult *res)
Definition: fe-exec.c:2673
TimeLineID timeline
Definition: receivelog.h:33
ExecStatusType PQresultStatus(const PGresult *res)
Definition: fe-exec.c:2596
static bool ReadEndOfStreamingResult(PGresult *res, XLogRecPtr *startpos, uint32 *timeline)
Definition: receivelog.c:712
static Walfile * walfile
Definition: receivelog.c:33
char * replication_slot
Definition: receivelog.h:47
static bool reportFlushPosition
Definition: receivelog.c:35
unsigned int uint32
Definition: c.h:268
stream_stop_callback stream_stop
Definition: receivelog.h:43
WalWriteMethod * walmethod
Definition: receivelog.h:45
static XLogRecPtr lastFlushPosition
Definition: receivelog.c:36
void PQclear(PGresult *res)
Definition: fe-exec.c:650
static PGresult * HandleCopyStream(PGconn *conn, StreamCtl *stream, XLogRecPtr *stoppos)
Definition: receivelog.c:760
#define NULL
Definition: c.h:229
uint64 XLogRecPtr
Definition: xlogdefs.h:21
static bool existsTimeLineHistoryFile(StreamCtl *stream)
Definition: receivelog.c:244
bool synchronous
Definition: receivelog.h:38
char * PQresultErrorMessage(const PGresult *res)
Definition: fe-exec.c:2612
PGresult * PQexec(PGconn *conn, const char *query)
Definition: fe-exec.c:1846
static char current_walfile_name[MAXPGPATH]
Definition: receivelog.c:34
bool temp_slot
Definition: receivelog.h:48
#define _(x)
Definition: elog.c:84
PGresult * PQgetResult(PGconn *conn)
Definition: fe-exec.c:1702
static bool writeTimeLineHistoryFile(StreamCtl *stream, char *filename, char *content)
Definition: receivelog.c:261
bool CheckServerVersionForStreaming(PGconn *conn)
Definition: receivelog.c:360
int(* close)(Walfile f, WalCloseMethod method)
Definition: walmethods.h:26
static bool sendFeedback ( PGconn conn,
XLogRecPtr  blockpos,
TimestampTz  now,
bool  replyRequested 
)
static

Definition at line 322 of file receivelog.c.

References _, fe_sendint64(), InvalidXLogRecPtr, lastFlushPosition, PQerrorMessage(), PQflush(), PQputCopyData(), progname, and reportFlushPosition.

Referenced by HandleCopyStream(), and ProcessKeepaliveMsg().

323 {
324  char replybuf[1 + 8 + 8 + 8 + 8 + 1];
325  int len = 0;
326 
327  replybuf[len] = 'r';
328  len += 1;
329  fe_sendint64(blockpos, &replybuf[len]); /* write */
330  len += 8;
332  fe_sendint64(lastFlushPosition, &replybuf[len]); /* flush */
333  else
334  fe_sendint64(InvalidXLogRecPtr, &replybuf[len]); /* flush */
335  len += 8;
336  fe_sendint64(InvalidXLogRecPtr, &replybuf[len]); /* apply */
337  len += 8;
338  fe_sendint64(now, &replybuf[len]); /* sendTime */
339  len += 8;
340  replybuf[len] = replyRequested ? 1 : 0; /* replyRequested */
341  len += 1;
342 
343  if (PQputCopyData(conn, replybuf, len) <= 0 || PQflush(conn))
344  {
345  fprintf(stderr, _("%s: could not send feedback packet: %s"),
346  progname, PQerrorMessage(conn));
347  return false;
348  }
349 
350  return true;
351 }
int PQputCopyData(PGconn *conn, const char *buffer, int nbytes)
Definition: fe-exec.c:2221
char * PQerrorMessage(const PGconn *conn)
Definition: fe-connect.c:6011
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28
const char * progname
Definition: pg_standby.c:37
int PQflush(PGconn *conn)
Definition: fe-exec.c:3187
static bool reportFlushPosition
Definition: receivelog.c:35
static XLogRecPtr lastFlushPosition
Definition: receivelog.c:36
void fe_sendint64(int64 i, char *buf)
Definition: streamutil.c:494
#define _(x)
Definition: elog.c:84
Datum now(PG_FUNCTION_ARGS)
Definition: timestamp.c:1533
static bool writeTimeLineHistoryFile ( StreamCtl stream,
char *  filename,
char *  content 
)
static

Definition at line 261 of file receivelog.c.

References _, WalWriteMethod::close, CLOSE_NORMAL, CLOSE_UNLINK, WalWriteMethod::getlasterror, StreamCtl::mark_done, mark_file_as_archived(), MAXFNAMELEN, NULL, WalWriteMethod::open_for_write, progname, StreamCtl::timeline, TLHistoryFileName, StreamCtl::walmethod, and WalWriteMethod::write.

Referenced by ReceiveXlogStream().

262 {
263  int size = strlen(content);
264  char histfname[MAXFNAMELEN];
265  Walfile *f;
266 
267  /*
268  * Check that the server's idea of how timeline history files should be
269  * named matches ours.
270  */
271  TLHistoryFileName(histfname, stream->timeline);
272  if (strcmp(histfname, filename) != 0)
273  {
274  fprintf(stderr, _("%s: server reported unexpected history file name for timeline %u: %s\n"),
275  progname, stream->timeline, filename);
276  return false;
277  }
278 
279  f = stream->walmethod->open_for_write(histfname, ".tmp", 0);
280  if (f == NULL)
281  {
282  fprintf(stderr, _("%s: could not create timeline history file \"%s\": %s\n"),
283  progname, histfname, stream->walmethod->getlasterror());
284  return false;
285  }
286 
287  if ((int) stream->walmethod->write(f, content, size) != size)
288  {
289  fprintf(stderr, _("%s: could not write timeline history file \"%s\": %s\n"),
290  progname, histfname, stream->walmethod->getlasterror());
291 
292  /*
293  * If we fail to make the file, delete it to release disk space
294  */
295  stream->walmethod->close(f, CLOSE_UNLINK);
296 
297  return false;
298  }
299 
300  if (stream->walmethod->close(f, CLOSE_NORMAL) != 0)
301  {
302  fprintf(stderr, _("%s: could not close file \"%s\": %s\n"),
303  progname, histfname, stream->walmethod->getlasterror());
304  return false;
305  }
306 
307  /* Maintain archive_status, check close_walfile() for details. */
308  if (stream->mark_done)
309  {
310  /* writes error message if failed */
311  if (!mark_file_as_archived(stream, histfname))
312  return false;
313  }
314 
315  return true;
316 }
char *(* getlasterror)(void)
Definition: walmethods.h:34
void * Walfile
Definition: walmethods.h:13
const char * progname
Definition: pg_standby.c:37
TimeLineID timeline
Definition: receivelog.h:33
bool mark_done
Definition: receivelog.h:39
#define TLHistoryFileName(fname, tli)
WalWriteMethod * walmethod
Definition: receivelog.h:45
#define MAXFNAMELEN
ssize_t(* write)(Walfile f, const void *buf, size_t count)
Definition: walmethods.h:30
#define NULL
Definition: c.h:229
Walfile(* open_for_write)(const char *pathname, const char *temp_suffix, size_t pad_to_size)
Definition: walmethods.h:25
static char * filename
Definition: pg_dumpall.c:87
static bool mark_file_as_archived(StreamCtl *stream, const char *fname)
Definition: receivelog.c:59
#define _(x)
Definition: elog.c:84
int(* close)(Walfile f, WalCloseMethod method)
Definition: walmethods.h:26

Variable Documentation

char current_walfile_name[MAXPGPATH] = ""
static
XLogRecPtr lastFlushPosition = InvalidXLogRecPtr
static
bool reportFlushPosition = false
static

Definition at line 35 of file receivelog.c.

Referenced by ProcessKeepaliveMsg(), ReceiveXlogStream(), and sendFeedback().