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, pgsocket stop_socket)
 
static int CopyStreamReceive (PGconn *conn, long timeout, pgsocket stop_socket, 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 1281 of file receivelog.c.

References feTimestampDifference(), sleeptime, and still_sending.

Referenced by HandleCopyStream().

1283 {
1284  TimestampTz status_targettime = 0;
1285  long sleeptime;
1286 
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 }
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:1534
static bool CheckCopyStreamStop ( PGconn conn,
StreamCtl stream,
XLogRecPtr  blockpos,
XLogRecPtr stoppos 
)
static

Definition at line 1255 of file receivelog.c.

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

Referenced by HandleCopyStream().

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 }
char * PQerrorMessage(const PGconn *conn)
Definition: fe-connect.c:6014
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:182
stream_stop_callback stream_stop
Definition: receivelog.h:42
#define NULL
Definition: c.h:229
#define _(x)
Definition: elog.c:84
bool CheckServerVersionForStreaming ( PGconn conn)

Definition at line 364 of file receivelog.c.

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

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

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 }
const char * PQparameterStatus(const PGconn *conn, const char *paramName)
Definition: fe-connect.c:5979
int PQserverVersion(const PGconn *conn)
Definition: fe-connect.c:6004
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 182 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().

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 }
char * partial_suffix
Definition: receivelog.h:48
const char * progname
Definition: pg_standby.c:37
static Walfile * walfile
Definition: receivelog.c:33
bool mark_done
Definition: receivelog.h:38
off_t(* get_current_pos)(Walfile f)
Definition: walmethods.h:62
WalWriteMethod * walmethod
Definition: receivelog.h:47
static XLogRecPtr lastFlushPosition
Definition: receivelog.c:36
#define NULL
Definition: c.h:229
const char *(* getlasterror)(void)
Definition: walmethods.h:78
static bool mark_file_as_archived(StreamCtl *stream, const char *fname)
Definition: receivelog.c:60
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:47
static int CopyStreamPoll ( PGconn conn,
long  timeout_ms,
pgsocket  stop_socket 
)
static

Definition at line 900 of file receivelog.c.

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

Referenced by CopyStreamReceive().

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 }
char * PQerrorMessage(const PGconn *conn)
Definition: fe-connect.c:6014
#define select(n, r, w, e, timeout)
Definition: win32.h:374
const char * progname
Definition: pg_standby.c:37
#define PGINVALID_SOCKET
Definition: port.h:24
#define EINTR
Definition: win32.h:285
#define Max(x, y)
Definition: c.h:800
#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:6032
static int CopyStreamReceive ( PGconn conn,
long  timeout,
pgsocket  stop_socket,
char **  buffer 
)
static

Definition at line 964 of file receivelog.c.

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

Referenced by HandleCopyStream().

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 }
char * PQerrorMessage(const PGconn *conn)
Definition: fe-connect.c:6014
static int CopyStreamPoll(PGconn *conn, long timeout_ms, pgsocket stop_socket)
Definition: receivelog.c:900
const char * progname
Definition: pg_standby.c:37
int PQgetCopyData(PGconn *conn, char **buffer, int async)
Definition: fe-exec.c:2377
int PQconsumeInput(PGconn *conn)
Definition: fe-exec.c:1631
StringInfo copybuf
Definition: tablesync.c:114
#define NULL
Definition: c.h:229
WalTimeSample buffer[LAG_TRACKER_BUFFER_SIZE]
Definition: walsender.c:214
#define _(x)
Definition: elog.c:84
void PQfreemem(void *ptr)
Definition: fe-exec.c:3200
static bool existsTimeLineHistoryFile ( StreamCtl stream)
static

Definition at line 248 of file receivelog.c.

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

Referenced by ReceiveXlogStream().

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 }
bool(* existsfile)(const char *pathname)
Definition: walmethods.h:50
TimeLineID timeline
Definition: receivelog.h:33
#define TLHistoryFileName(fname, tli)
WalWriteMethod * walmethod
Definition: receivelog.h:47
#define MAXFNAMELEN
static PGresult * HandleCopyStream ( PGconn conn,
StreamCtl stream,
XLogRecPtr stoppos 
)
static

Definition at line 771 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, StreamCtl::stop_socket, WalWriteMethod::sync, StreamCtl::synchronous, walfile, and StreamCtl::walmethod.

Referenced by ReceiveXlogStream().

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 }
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:1255
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
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:1021
static int CopyStreamReceive(PGconn *conn, long timeout, pgsocket stop_socket, char **buffer)
Definition: receivelog.c:964
static Walfile * walfile
Definition: receivelog.c:33
WalWriteMethod * walmethod
Definition: receivelog.h:47
int(* sync)(Walfile f)
Definition: walmethods.h:67
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
static bool ProcessXLogDataMsg(PGconn *conn, StreamCtl *stream, char *copybuf, int len, XLogRecPtr *blockpos)
Definition: receivelog.c:1079
StringInfo copybuf
Definition: tablesync.c:114
#define NULL
Definition: c.h:229
uint64 XLogRecPtr
Definition: xlogdefs.h:21
const char *(* getlasterror)(void)
Definition: walmethods.h:78
bool synchronous
Definition: receivelog.h:37
pgsocket stop_socket
Definition: receivelog.h:44
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:1534
static PGresult * HandleEndOfCopyStream ( PGconn conn,
StreamCtl stream,
char *  copybuf,
XLogRecPtr  blockpos,
XLogRecPtr stoppos 
)
static

Definition at line 1213 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().

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 }
char * PQerrorMessage(const PGconn *conn)
Definition: fe-connect.c:6014
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:182
StringInfo copybuf
Definition: tablesync.c:114
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 60 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().

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 }
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:47
#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:41
const char *(* getlasterror)(void)
Definition: walmethods.h:78
#define _(x)
Definition: elog.c:84
int(* close)(Walfile f, WalCloseMethod method)
Definition: walmethods.h:47
static bool open_walfile ( StreamCtl stream,
XLogRecPtr  startpoint 
)
static

Definition at line 91 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().

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 }
#define XLogSegSize
Definition: xlog_internal.h:92
void * Walfile
Definition: walmethods.h:13
bool(* existsfile)(const char *pathname)
Definition: walmethods.h:50
#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:48
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:53
#define MAXPGPATH
static Walfile * walfile
Definition: receivelog.c:33
uint64 XLogSegNo
Definition: xlogdefs.h:34
WalWriteMethod * walmethod
Definition: receivelog.h:47
int(* sync)(Walfile f)
Definition: walmethods.h:67
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:41
const char *(* getlasterror)(void)
Definition: walmethods.h:78
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:47
static bool ProcessKeepaliveMsg ( PGconn conn,
StreamCtl stream,
char *  copybuf,
int  len,
XLogRecPtr  blockpos,
TimestampTz last_status 
)
static

Definition at line 1021 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().

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 }
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:47
int(* sync)(Walfile f)
Definition: walmethods.h:67
static bool sendFeedback(PGconn *conn, XLogRecPtr blockpos, TimestampTz now, bool replyRequested)
Definition: receivelog.c:326
static XLogRecPtr lastFlushPosition
Definition: receivelog.c:36
StringInfo copybuf
Definition: tablesync.c:114
#define NULL
Definition: c.h:229
const char *(* getlasterror)(void)
Definition: walmethods.h:78
static char current_walfile_name[MAXPGPATH]
Definition: receivelog.c:34
#define _(x)
Definition: elog.c:84
Datum now(PG_FUNCTION_ARGS)
Definition: timestamp.c:1534
static bool ProcessXLogDataMsg ( PGconn conn,
StreamCtl stream,
char *  copybuf,
int  len,
XLogRecPtr blockpos 
)
static

Definition at line 1079 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().

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 }
static bool open_walfile(StreamCtl *stream, XLogRecPtr startpoint)
Definition: receivelog.c:91
char * PQerrorMessage(const PGconn *conn)
Definition: fe-connect.c:6014
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:62
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
StringInfo copybuf
Definition: tablesync.c:114
ssize_t(* write)(Walfile f, const void *buf, size_t count)
Definition: walmethods.h:59
#define NULL
Definition: c.h:229
const char *(* getlasterror)(void)
Definition: walmethods.h:78
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 723 of file receivelog.c.

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

Referenced by ReceiveXlogStream().

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 }
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 444 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().

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 }
int PQnfields(const PGresult *res)
Definition: fe-exec.c:2681
char * PQerrorMessage(const PGconn *conn)
Definition: fe-connect.c:6014
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:723
static Walfile * walfile
Definition: receivelog.c:33
char * replication_slot
Definition: receivelog.h:49
static bool reportFlushPosition
Definition: receivelog.c:35
unsigned int uint32
Definition: c.h:268
stream_stop_callback stream_stop
Definition: receivelog.h:42
WalWriteMethod * walmethod
Definition: receivelog.h:47
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:771
#define NULL
Definition: c.h:229
uint64 XLogRecPtr
Definition: xlogdefs.h:21
static bool existsTimeLineHistoryFile(StreamCtl *stream)
Definition: receivelog.c:248
const char *(* getlasterror)(void)
Definition: walmethods.h:78
bool synchronous
Definition: receivelog.h:37
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:50
#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:265
bool CheckServerVersionForStreaming(PGconn *conn)
Definition: receivelog.c:364
int(* close)(Walfile f, WalCloseMethod method)
Definition: walmethods.h:47
static bool sendFeedback ( PGconn conn,
XLogRecPtr  blockpos,
TimestampTz  now,
bool  replyRequested 
)
static

Definition at line 326 of file receivelog.c.

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

Referenced by HandleCopyStream(), and ProcessKeepaliveMsg().

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 }
int PQputCopyData(PGconn *conn, const char *buffer, int nbytes)
Definition: fe-exec.c:2221
char * PQerrorMessage(const PGconn *conn)
Definition: fe-connect.c:6014
#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:1534
static bool writeTimeLineHistoryFile ( StreamCtl stream,
char *  filename,
char *  content 
)
static

Definition at line 265 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().

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 }
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:38
#define TLHistoryFileName(fname, tli)
WalWriteMethod * walmethod
Definition: receivelog.h:47
#define MAXFNAMELEN
ssize_t(* write)(Walfile f, const void *buf, size_t count)
Definition: walmethods.h:59
#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:41
const char *(* getlasterror)(void)
Definition: walmethods.h:78
static char * filename
Definition: pg_dumpall.c:89
static bool mark_file_as_archived(StreamCtl *stream, const char *fname)
Definition: receivelog.c:60
#define _(x)
Definition: elog.c:84
int(* close)(Walfile f, WalCloseMethod method)
Definition: walmethods.h:47

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().