PostgreSQL Source Code  git master
receivelog.c File Reference
#include "postgres_fe.h"
#include <sys/stat.h>
#include <unistd.h>
#include "access/xlog_internal.h"
#include "common/file_utils.h"
#include "common/logging.h"
#include "libpq-fe.h"
#include "receivelog.h"
#include "streamutil.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)
 
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

◆ CalculateCopyStreamSleeptime()

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

Definition at line 1250 of file receivelog.c.

1252 {
1253  TimestampTz status_targettime = 0;
1254  long sleeptime;
1255 
1257  status_targettime = last_status +
1258  (standby_message_timeout - 1) * ((int64) 1000);
1259 
1260  if (status_targettime > 0)
1261  {
1262  long secs;
1263  int usecs;
1264 
1266  status_targettime,
1267  &secs,
1268  &usecs);
1269  /* Always sleep at least 1 sec */
1270  if (secs <= 0)
1271  {
1272  secs = 1;
1273  usecs = 0;
1274  }
1275 
1276  sleeptime = secs * 1000 + usecs / 1000;
1277  }
1278  else
1279  sleeptime = -1;
1280 
1281  return sleeptime;
1282 }
Datum now(PG_FUNCTION_ARGS)
Definition: timestamp.c:1544
int64 TimestampTz
Definition: timestamp.h:39
static int standby_message_timeout
static bool still_sending
Definition: receivelog.c:36
void feTimestampDifference(TimestampTz start_time, TimestampTz stop_time, long *secs, int *microsecs)
Definition: streamutil.c:817

References feTimestampDifference(), now(), standby_message_timeout, and still_sending.

Referenced by HandleCopyStream().

◆ CheckCopyStreamStop()

static bool CheckCopyStreamStop ( PGconn conn,
StreamCtl stream,
XLogRecPtr  blockpos 
)
static

Definition at line 1225 of file receivelog.c.

1226 {
1227  if (still_sending && stream->stream_stop(blockpos, stream->timeline, false))
1228  {
1229  if (!close_walfile(stream, blockpos))
1230  {
1231  /* Potential error message is written by close_walfile */
1232  return false;
1233  }
1234  if (PQputCopyEnd(conn, NULL) <= 0 || PQflush(conn))
1235  {
1236  pg_log_error("could not send copy-end packet: %s",
1237  PQerrorMessage(conn));
1238  return false;
1239  }
1240  still_sending = false;
1241  }
1242 
1243  return true;
1244 }
char * PQerrorMessage(const PGconn *conn)
Definition: fe-connect.c:6915
int PQflush(PGconn *conn)
Definition: fe-exec.c:3766
int PQputCopyEnd(PGconn *conn, const char *errormsg)
Definition: fe-exec.c:2600
#define pg_log_error(...)
Definition: logging.h:80
static bool close_walfile(StreamCtl *stream, XLogRecPtr pos)
Definition: receivelog.c:191
PGconn * conn
Definition: streamutil.c:54
TimeLineID timeline
Definition: receivelog.h:32
stream_stop_callback stream_stop
Definition: receivelog.h:41

References close_walfile(), conn, pg_log_error, PQerrorMessage(), PQflush(), PQputCopyEnd(), still_sending, StreamCtl::stream_stop, and StreamCtl::timeline.

Referenced by HandleCopyStream().

◆ CheckServerVersionForStreaming()

bool CheckServerVersionForStreaming ( PGconn conn)

Definition at line 381 of file receivelog.c.

382 {
383  int minServerMajor,
384  maxServerMajor;
385  int serverMajor;
386 
387  /*
388  * The message format used in streaming replication changed in 9.3, so we
389  * cannot stream from older servers. And we don't support servers newer
390  * than the client; it might work, but we don't know, so err on the safe
391  * side.
392  */
393  minServerMajor = 903;
394  maxServerMajor = PG_VERSION_NUM / 100;
395  serverMajor = PQserverVersion(conn) / 100;
396  if (serverMajor < minServerMajor)
397  {
398  const char *serverver = PQparameterStatus(conn, "server_version");
399 
400  pg_log_error("incompatible server version %s; client does not support streaming from server versions older than %s",
401  serverver ? serverver : "'unknown'",
402  "9.3");
403  return false;
404  }
405  else if (serverMajor > maxServerMajor)
406  {
407  const char *serverver = PQparameterStatus(conn, "server_version");
408 
409  pg_log_error("incompatible server version %s; client does not support streaming from server versions newer than %s",
410  serverver ? serverver : "'unknown'",
411  PG_VERSION);
412  return false;
413  }
414  return true;
415 }
const char * PQparameterStatus(const PGconn *conn, const char *paramName)
Definition: fe-connect.c:6880
int PQserverVersion(const PGconn *conn)
Definition: fe-connect.c:6905

References conn, pg_log_error, PQparameterStatus(), and PQserverVersion().

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

◆ close_walfile()

static bool close_walfile ( StreamCtl stream,
XLogRecPtr  pos 
)
static

Definition at line 191 of file receivelog.c.

192 {
193  char *fn;
194  off_t currpos;
195  int r;
196 
197  if (walfile == NULL)
198  return true;
199 
200  /* Note that this considers the compression used if necessary */
202  stream->partial_suffix);
203 
204  currpos = stream->walmethod->get_current_pos(walfile);
205 
206  if (currpos == -1)
207  {
208  pg_log_error("could not determine seek position in file \"%s\": %s",
209  fn, stream->walmethod->getlasterror());
210  stream->walmethod->close(walfile, CLOSE_UNLINK);
211  walfile = NULL;
212 
213  pg_free(fn);
214  return false;
215  }
216 
217  if (stream->partial_suffix)
218  {
219  if (currpos == WalSegSz)
220  r = stream->walmethod->close(walfile, CLOSE_NORMAL);
221  else
222  {
223  pg_log_info("not renaming \"%s\", segment is not complete", fn);
224  r = stream->walmethod->close(walfile, CLOSE_NO_RENAME);
225  }
226  }
227  else
228  r = stream->walmethod->close(walfile, CLOSE_NORMAL);
229 
230  walfile = NULL;
231 
232  if (r != 0)
233  {
234  pg_log_error("could not close file \"%s\": %s",
235  fn, stream->walmethod->getlasterror());
236 
237  pg_free(fn);
238  return false;
239  }
240 
241  pg_free(fn);
242 
243  /*
244  * Mark file as archived if requested by the caller - pg_basebackup needs
245  * to do so as files can otherwise get archived again after promotion of a
246  * new node. This is in line with walreceiver.c always doing a
247  * XLogArchiveForceDone() after a complete segment.
248  */
249  if (currpos == WalSegSz && stream->mark_done)
250  {
251  /* writes error message if failed */
253  return false;
254  }
255 
256  lastFlushPosition = pos;
257  return true;
258 }
void pg_free(void *ptr)
Definition: fe_memutils.c:105
#define pg_log_info(...)
Definition: logging.h:88
static bool mark_file_as_archived(StreamCtl *stream, const char *fname)
Definition: receivelog.c:57
static char current_walfile_name[MAXPGPATH]
Definition: receivelog.c:32
static Walfile * walfile
Definition: receivelog.c:31
static XLogRecPtr lastFlushPosition
Definition: receivelog.c:34
uint32 WalSegSz
Definition: streamutil.c:34
WalWriteMethod * walmethod
Definition: receivelog.h:46
bool mark_done
Definition: receivelog.h:37
char * partial_suffix
Definition: receivelog.h:47
const char *(* getlasterror)(void)
Definition: walmethods.h:95
off_t(* get_current_pos)(Walfile f)
Definition: walmethods.h:79
int(* close)(Walfile f, WalCloseMethod method)
Definition: walmethods.h:55
char *(* get_file_name)(const char *pathname, const char *temp_suffix)
Definition: walmethods.h:67
static void * fn(void *arg)
@ CLOSE_UNLINK
Definition: walmethods.h:18
@ CLOSE_NO_RENAME
Definition: walmethods.h:19
@ CLOSE_NORMAL
Definition: walmethods.h:17

References WalWriteMethod::close, CLOSE_NO_RENAME, CLOSE_NORMAL, CLOSE_UNLINK, current_walfile_name, fn(), WalWriteMethod::get_current_pos, WalWriteMethod::get_file_name, WalWriteMethod::getlasterror, lastFlushPosition, StreamCtl::mark_done, mark_file_as_archived(), StreamCtl::partial_suffix, pg_free(), pg_log_error, pg_log_info, walfile, StreamCtl::walmethod, and WalSegSz.

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

◆ CopyStreamPoll()

static int CopyStreamPoll ( PGconn conn,
long  timeout_ms,
pgsocket  stop_socket 
)
static

Definition at line 880 of file receivelog.c.

881 {
882  int ret;
883  fd_set input_mask;
884  int connsocket;
885  int maxfd;
886  struct timeval timeout;
887  struct timeval *timeoutptr;
888 
889  connsocket = PQsocket(conn);
890  if (connsocket < 0)
891  {
892  pg_log_error("invalid socket: %s", PQerrorMessage(conn));
893  return -1;
894  }
895 
896  FD_ZERO(&input_mask);
897  FD_SET(connsocket, &input_mask);
898  maxfd = connsocket;
899  if (stop_socket != PGINVALID_SOCKET)
900  {
901  FD_SET(stop_socket, &input_mask);
902  maxfd = Max(maxfd, stop_socket);
903  }
904 
905  if (timeout_ms < 0)
906  timeoutptr = NULL;
907  else
908  {
909  timeout.tv_sec = timeout_ms / 1000L;
910  timeout.tv_usec = (timeout_ms % 1000L) * 1000L;
911  timeoutptr = &timeout;
912  }
913 
914  ret = select(maxfd + 1, &input_mask, NULL, NULL, timeoutptr);
915 
916  if (ret < 0)
917  {
918  if (errno == EINTR)
919  return 0; /* Got a signal, so not an error */
920  pg_log_error("%s() failed: %m", "select");
921  return -1;
922  }
923  if (ret > 0 && FD_ISSET(connsocket, &input_mask))
924  return 1; /* Got input on connection socket */
925 
926  return 0; /* Got timeout or input on stop_socket */
927 }
#define Max(x, y)
Definition: c.h:980
int PQsocket(const PGconn *conn)
Definition: fe-connect.c:6941
#define PGINVALID_SOCKET
Definition: port.h:31
#define EINTR
Definition: win32_port.h:351
#define select(n, r, w, e, timeout)
Definition: win32_port.h:474

References conn, EINTR, Max, pg_log_error, PGINVALID_SOCKET, PQerrorMessage(), PQsocket(), and select.

Referenced by CopyStreamReceive().

◆ CopyStreamReceive()

static int CopyStreamReceive ( PGconn conn,
long  timeout,
pgsocket  stop_socket,
char **  buffer 
)
static

Definition at line 942 of file receivelog.c.

944 {
945  char *copybuf = NULL;
946  int rawlen;
947 
948  if (*buffer != NULL)
949  PQfreemem(*buffer);
950  *buffer = NULL;
951 
952  /* Try to receive a CopyData message */
953  rawlen = PQgetCopyData(conn, &copybuf, 1);
954  if (rawlen == 0)
955  {
956  int ret;
957 
958  /*
959  * No data available. Wait for some to appear, but not longer than
960  * the specified timeout, so that we can ping the server. Also stop
961  * waiting if input appears on stop_socket.
962  */
963  ret = CopyStreamPoll(conn, timeout, stop_socket);
964  if (ret <= 0)
965  return ret;
966 
967  /* Now there is actually data on the socket */
968  if (PQconsumeInput(conn) == 0)
969  {
970  pg_log_error("could not receive data from WAL stream: %s",
972  return -1;
973  }
974 
975  /* Now that we've consumed some input, try again */
976  rawlen = PQgetCopyData(conn, &copybuf, 1);
977  if (rawlen == 0)
978  return 0;
979  }
980  if (rawlen == -1) /* end-of-streaming or error */
981  return -2;
982  if (rawlen == -2)
983  {
984  pg_log_error("could not read COPY data: %s", PQerrorMessage(conn));
985  return -1;
986  }
987 
988  /* Return received messages to caller */
989  *buffer = copybuf;
990  return rawlen;
991 }
void PQfreemem(void *ptr)
Definition: fe-exec.c:3796
int PQconsumeInput(PGconn *conn)
Definition: fe-exec.c:1904
int PQgetCopyData(PGconn *conn, char **buffer, int async)
Definition: fe-exec.c:2668
static int CopyStreamPoll(PGconn *conn, long timeout_ms, pgsocket stop_socket)
Definition: receivelog.c:880
StringInfo copybuf
Definition: tablesync.c:126

References conn, copybuf, CopyStreamPoll(), pg_log_error, PQconsumeInput(), PQerrorMessage(), PQfreemem(), and PQgetCopyData().

Referenced by HandleCopyStream().

◆ existsTimeLineHistoryFile()

static bool existsTimeLineHistoryFile ( StreamCtl stream)
static

Definition at line 265 of file receivelog.c.

266 {
267  char histfname[MAXFNAMELEN];
268 
269  /*
270  * Timeline 1 never has a history file. We treat that as if it existed,
271  * since we never need to stream it.
272  */
273  if (stream->timeline == 1)
274  return true;
275 
276  TLHistoryFileName(histfname, stream->timeline);
277 
278  return stream->walmethod->existsfile(histfname);
279 }
bool(* existsfile)(const char *pathname)
Definition: walmethods.h:58
#define TLHistoryFileName(fname, tli)
#define MAXFNAMELEN

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

Referenced by ReceiveXlogStream().

◆ HandleCopyStream()

static PGresult * HandleCopyStream ( PGconn conn,
StreamCtl stream,
XLogRecPtr stoppos 
)
static

Definition at line 751 of file receivelog.c.

753 {
754  char *copybuf = NULL;
755  TimestampTz last_status = -1;
756  XLogRecPtr blockpos = stream->startpos;
757 
758  still_sending = true;
759 
760  while (1)
761  {
762  int r;
764  long sleeptime;
765 
766  /*
767  * Check if we should continue streaming, or abort at this point.
768  */
769  if (!CheckCopyStreamStop(conn, stream, blockpos))
770  goto error;
771 
773 
774  /*
775  * If synchronous option is true, issue sync command as soon as there
776  * are WAL data which has not been flushed yet.
777  */
778  if (stream->synchronous && lastFlushPosition < blockpos && walfile != NULL)
779  {
780  if (stream->walmethod->sync(walfile) != 0)
781  {
782  pg_log_fatal("could not fsync file \"%s\": %s",
784  exit(1);
785  }
786  lastFlushPosition = blockpos;
787 
788  /*
789  * Send feedback so that the server sees the latest WAL locations
790  * immediately.
791  */
792  if (!sendFeedback(conn, blockpos, now, false))
793  goto error;
794  last_status = now;
795  }
796 
797  /*
798  * Potentially send a status message to the primary
799  */
800  if (still_sending && stream->standby_message_timeout > 0 &&
801  feTimestampDifferenceExceeds(last_status, now,
802  stream->standby_message_timeout))
803  {
804  /* Time to send feedback! */
805  if (!sendFeedback(conn, blockpos, now, false))
806  goto error;
807  last_status = now;
808  }
809 
810  /*
811  * Calculate how long send/receive loops should sleep
812  */
814  last_status);
815 
816  r = CopyStreamReceive(conn, sleeptime, stream->stop_socket, &copybuf);
817  while (r != 0)
818  {
819  if (r == -1)
820  goto error;
821  if (r == -2)
822  {
823  PGresult *res = HandleEndOfCopyStream(conn, stream, copybuf, blockpos, stoppos);
824 
825  if (res == NULL)
826  goto error;
827  else
828  return res;
829  }
830 
831  /* Check the message type. */
832  if (copybuf[0] == 'k')
833  {
834  if (!ProcessKeepaliveMsg(conn, stream, copybuf, r, blockpos,
835  &last_status))
836  goto error;
837  }
838  else if (copybuf[0] == 'w')
839  {
840  if (!ProcessXLogDataMsg(conn, stream, copybuf, r, &blockpos))
841  goto error;
842 
843  /*
844  * Check if we should continue streaming, or abort at this
845  * point.
846  */
847  if (!CheckCopyStreamStop(conn, stream, blockpos))
848  goto error;
849  }
850  else
851  {
852  pg_log_error("unrecognized streaming header: \"%c\"",
853  copybuf[0]);
854  goto error;
855  }
856 
857  /*
858  * Process the received data, and any subsequent data we can read
859  * without blocking.
860  */
861  r = CopyStreamReceive(conn, 0, stream->stop_socket, &copybuf);
862  }
863  }
864 
865 error:
866  if (copybuf != NULL)
868  return NULL;
869 }
exit(1)
#define pg_log_fatal(...)
Definition: logging.h:76
static int CopyStreamReceive(PGconn *conn, long timeout, pgsocket stop_socket, char **buffer)
Definition: receivelog.c:942
static bool CheckCopyStreamStop(PGconn *conn, StreamCtl *stream, XLogRecPtr blockpos)
Definition: receivelog.c:1225
static bool ProcessKeepaliveMsg(PGconn *conn, StreamCtl *stream, char *copybuf, int len, XLogRecPtr blockpos, TimestampTz *last_status)
Definition: receivelog.c:997
static bool ProcessXLogDataMsg(PGconn *conn, StreamCtl *stream, char *copybuf, int len, XLogRecPtr *blockpos)
Definition: receivelog.c:1054
static PGresult * HandleEndOfCopyStream(PGconn *conn, StreamCtl *stream, char *copybuf, XLogRecPtr blockpos, XLogRecPtr *stoppos)
Definition: receivelog.c:1184
static long CalculateCopyStreamSleeptime(TimestampTz now, int standby_message_timeout, TimestampTz last_status)
Definition: receivelog.c:1250
static bool sendFeedback(PGconn *conn, XLogRecPtr blockpos, TimestampTz now, bool replyRequested)
Definition: receivelog.c:343
static void error(void)
Definition: sql-dyntest.c:147
TimestampTz feGetCurrentTimestamp(void)
Definition: streamutil.c:798
bool feTimestampDifferenceExceeds(TimestampTz start_time, TimestampTz stop_time, int msec)
Definition: streamutil.c:839
XLogRecPtr startpos
Definition: receivelog.h:31
pgsocket stop_socket
Definition: receivelog.h:43
int standby_message_timeout
Definition: receivelog.h:35
bool synchronous
Definition: receivelog.h:36
int(* sync)(Walfile f)
Definition: walmethods.h:84
uint64 XLogRecPtr
Definition: xlogdefs.h:21

References CalculateCopyStreamSleeptime(), CheckCopyStreamStop(), conn, copybuf, CopyStreamReceive(), current_walfile_name, error(), exit(), feGetCurrentTimestamp(), feTimestampDifferenceExceeds(), WalWriteMethod::getlasterror, HandleEndOfCopyStream(), lastFlushPosition, now(), pg_log_error, pg_log_fatal, PQfreemem(), ProcessKeepaliveMsg(), ProcessXLogDataMsg(), res, sendFeedback(), StreamCtl::standby_message_timeout, StreamCtl::startpos, still_sending, StreamCtl::stop_socket, WalWriteMethod::sync, StreamCtl::synchronous, walfile, and StreamCtl::walmethod.

Referenced by ReceiveXlogStream().

◆ HandleEndOfCopyStream()

static PGresult * HandleEndOfCopyStream ( PGconn conn,
StreamCtl stream,
char *  copybuf,
XLogRecPtr  blockpos,
XLogRecPtr stoppos 
)
static

Definition at line 1184 of file receivelog.c.

1186 {
1188 
1189  /*
1190  * The server closed its end of the copy stream. If we haven't closed
1191  * ours already, we need to do so now, unless the server threw an error,
1192  * in which case we don't.
1193  */
1194  if (still_sending)
1195  {
1196  if (!close_walfile(stream, blockpos))
1197  {
1198  /* Error message written in close_walfile() */
1199  PQclear(res);
1200  return NULL;
1201  }
1203  {
1204  if (PQputCopyEnd(conn, NULL) <= 0 || PQflush(conn))
1205  {
1206  pg_log_error("could not send copy-end packet: %s",
1207  PQerrorMessage(conn));
1208  PQclear(res);
1209  return NULL;
1210  }
1211  res = PQgetResult(conn);
1212  }
1213  still_sending = false;
1214  }
1215  if (copybuf != NULL)
1216  PQfreemem(copybuf);
1217  *stoppos = blockpos;
1218  return res;
1219 }
ExecStatusType PQresultStatus(const PGresult *res)
Definition: fe-exec.c:3178
void PQclear(PGresult *res)
Definition: fe-exec.c:694
PGresult * PQgetResult(PGconn *conn)
Definition: fe-exec.c:1978
@ PGRES_COPY_IN
Definition: libpq-fe.h:102

References close_walfile(), conn, copybuf, pg_log_error, PGRES_COPY_IN, PQclear(), PQerrorMessage(), PQflush(), PQfreemem(), PQgetResult(), PQputCopyEnd(), PQresultStatus(), res, and still_sending.

Referenced by HandleCopyStream().

◆ mark_file_as_archived()

static bool mark_file_as_archived ( StreamCtl stream,
const char *  fname 
)
static

Definition at line 57 of file receivelog.c.

58 {
59  Walfile *f;
60  static char tmppath[MAXPGPATH];
61 
62  snprintf(tmppath, sizeof(tmppath), "archive_status/%s.done",
63  fname);
64 
65  f = stream->walmethod->open_for_write(tmppath, NULL, 0);
66  if (f == NULL)
67  {
68  pg_log_error("could not create archive status file \"%s\": %s",
69  tmppath, stream->walmethod->getlasterror());
70  return false;
71  }
72 
73  if (stream->walmethod->close(f, CLOSE_NORMAL) != 0)
74  {
75  pg_log_error("could not close archive status file \"%s\": %s",
76  tmppath, stream->walmethod->getlasterror());
77  return false;
78  }
79 
80  return true;
81 }
#define MAXPGPATH
#define snprintf
Definition: port.h:225
Walfile(* open_for_write)(const char *pathname, const char *temp_suffix, size_t pad_to_size)
Definition: walmethods.h:49
void * Walfile
Definition: walmethods.h:13

References WalWriteMethod::close, CLOSE_NORMAL, WalWriteMethod::getlasterror, MAXPGPATH, WalWriteMethod::open_for_write, pg_log_error, snprintf, and StreamCtl::walmethod.

Referenced by close_walfile(), and writeTimeLineHistoryFile().

◆ open_walfile()

static bool open_walfile ( StreamCtl stream,
XLogRecPtr  startpoint 
)
static

Definition at line 93 of file receivelog.c.

94 {
95  Walfile *f;
96  char *fn;
97  ssize_t size;
98  XLogSegNo segno;
99 
100  XLByteToSeg(startpoint, segno, WalSegSz);
102 
103  /* Note that this considers the compression used if necessary */
105  stream->partial_suffix);
106 
107  /*
108  * When streaming to files, if an existing file exists we verify that it's
109  * either empty (just created), or a complete WalSegSz segment (in which
110  * case it has been created and padded). Anything else indicates a corrupt
111  * file. Compressed files have no need for padding, so just ignore this
112  * case.
113  *
114  * When streaming to tar, no file with this name will exist before, so we
115  * never have to verify a size.
116  */
117  if (stream->walmethod->compression_method() == COMPRESSION_NONE &&
118  stream->walmethod->existsfile(fn))
119  {
120  size = stream->walmethod->get_file_size(fn);
121  if (size < 0)
122  {
123  pg_log_error("could not get size of write-ahead log file \"%s\": %s",
124  fn, stream->walmethod->getlasterror());
125  pg_free(fn);
126  return false;
127  }
128  if (size == WalSegSz)
129  {
130  /* Already padded file. Open it for use */
132  if (f == NULL)
133  {
134  pg_log_error("could not open existing write-ahead log file \"%s\": %s",
135  fn, stream->walmethod->getlasterror());
136  pg_free(fn);
137  return false;
138  }
139 
140  /* fsync file in case of a previous crash */
141  if (stream->walmethod->sync(f) != 0)
142  {
143  pg_log_fatal("could not fsync existing write-ahead log file \"%s\": %s",
144  fn, stream->walmethod->getlasterror());
145  stream->walmethod->close(f, CLOSE_UNLINK);
146  exit(1);
147  }
148 
149  walfile = f;
150  pg_free(fn);
151  return true;
152  }
153  if (size != 0)
154  {
155  /* if write didn't set errno, assume problem is no disk space */
156  if (errno == 0)
157  errno = ENOSPC;
158  pg_log_error(ngettext("write-ahead log file \"%s\" has %zd byte, should be 0 or %d",
159  "write-ahead log file \"%s\" has %zd bytes, should be 0 or %d",
160  size),
161  fn, size, WalSegSz);
162  pg_free(fn);
163  return false;
164  }
165  /* File existed and was empty, so fall through and open */
166  }
167 
168  /* No file existed, so create one */
169 
171  stream->partial_suffix, WalSegSz);
172  if (f == NULL)
173  {
174  pg_log_error("could not open write-ahead log file \"%s\": %s",
175  fn, stream->walmethod->getlasterror());
176  pg_free(fn);
177  return false;
178  }
179 
180  pg_free(fn);
181  walfile = f;
182  return true;
183 }
#define ngettext(s, p, n)
Definition: c.h:1179
ssize_t(* get_file_size)(const char *pathname)
Definition: walmethods.h:61
WalCompressionMethod(* compression_method)(void)
Definition: walmethods.h:70
@ COMPRESSION_NONE
Definition: walmethods.h:27
#define XLogFileName(fname, tli, logSegNo, wal_segsz_bytes)
#define XLByteToSeg(xlrp, logSegNo, wal_segsz_bytes)
uint64 XLogSegNo
Definition: xlogdefs.h:48

References WalWriteMethod::close, CLOSE_UNLINK, WalWriteMethod::compression_method, COMPRESSION_NONE, current_walfile_name, WalWriteMethod::existsfile, exit(), fn(), WalWriteMethod::get_file_name, WalWriteMethod::get_file_size, WalWriteMethod::getlasterror, ngettext, WalWriteMethod::open_for_write, StreamCtl::partial_suffix, pg_free(), pg_log_error, pg_log_fatal, WalWriteMethod::sync, StreamCtl::timeline, walfile, StreamCtl::walmethod, WalSegSz, XLByteToSeg, and XLogFileName.

Referenced by ProcessXLogDataMsg().

◆ ProcessKeepaliveMsg()

static bool ProcessKeepaliveMsg ( PGconn conn,
StreamCtl stream,
char *  copybuf,
int  len,
XLogRecPtr  blockpos,
TimestampTz last_status 
)
static

Definition at line 997 of file receivelog.c.

999 {
1000  int pos;
1001  bool replyRequested;
1002  TimestampTz now;
1003 
1004  /*
1005  * Parse the keepalive message, enclosed in the CopyData message. We just
1006  * check if the server requested a reply, and ignore the rest.
1007  */
1008  pos = 1; /* skip msgtype 'k' */
1009  pos += 8; /* skip walEnd */
1010  pos += 8; /* skip sendTime */
1011 
1012  if (len < pos + 1)
1013  {
1014  pg_log_error("streaming header too small: %d", 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  pg_log_fatal("could not fsync file \"%s\": %s",
1036  exit(1);
1037  }
1038  lastFlushPosition = blockpos;
1039  }
1040 
1042  if (!sendFeedback(conn, blockpos, now, false))
1043  return false;
1044  *last_status = now;
1045  }
1046 
1047  return true;
1048 }
const void size_t len
static bool reportFlushPosition
Definition: receivelog.c:33

References conn, copybuf, current_walfile_name, exit(), feGetCurrentTimestamp(), WalWriteMethod::getlasterror, lastFlushPosition, len, now(), pg_log_error, pg_log_fatal, reportFlushPosition, sendFeedback(), still_sending, WalWriteMethod::sync, walfile, and StreamCtl::walmethod.

Referenced by HandleCopyStream().

◆ ProcessXLogDataMsg()

static bool ProcessXLogDataMsg ( PGconn conn,
StreamCtl stream,
char *  copybuf,
int  len,
XLogRecPtr blockpos 
)
static

Definition at line 1054 of file receivelog.c.

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  pg_log_error("streaming header too small: %d", len);
1081  return false;
1082  }
1083  *blockpos = fe_recvint64(&copybuf[1]);
1084 
1085  /* Extract WAL location for this block */
1086  xlogoff = XLogSegmentOffset(*blockpos, WalSegSz);
1087 
1088  /*
1089  * Verify that the initial location in the stream matches where we think
1090  * we are.
1091  */
1092  if (walfile == NULL)
1093  {
1094  /* No file open yet */
1095  if (xlogoff != 0)
1096  {
1097  pg_log_error("received write-ahead log record for offset %u with no file open",
1098  xlogoff);
1099  return false;
1100  }
1101  }
1102  else
1103  {
1104  /* More data in existing segment */
1105  if (stream->walmethod->get_current_pos(walfile) != xlogoff)
1106  {
1107  pg_log_error("got WAL data offset %08x, expected %08x",
1108  xlogoff, (int) stream->walmethod->get_current_pos(walfile));
1109  return false;
1110  }
1111  }
1112 
1113  bytes_left = len - hdr_len;
1114  bytes_written = 0;
1115 
1116  while (bytes_left)
1117  {
1118  int bytes_to_write;
1119 
1120  /*
1121  * If crossing a WAL boundary, only write up until we reach wal
1122  * segment size.
1123  */
1124  if (xlogoff + bytes_left > WalSegSz)
1125  bytes_to_write = WalSegSz - xlogoff;
1126  else
1127  bytes_to_write = bytes_left;
1128 
1129  if (walfile == NULL)
1130  {
1131  if (!open_walfile(stream, *blockpos))
1132  {
1133  /* Error logged by open_walfile */
1134  return false;
1135  }
1136  }
1137 
1138  if (stream->walmethod->write(walfile, copybuf + hdr_len + bytes_written,
1139  bytes_to_write) != bytes_to_write)
1140  {
1141  pg_log_error("could not write %d bytes to WAL file \"%s\": %s",
1142  bytes_to_write, current_walfile_name,
1143  stream->walmethod->getlasterror());
1144  return false;
1145  }
1146 
1147  /* Write was successful, advance our position */
1148  bytes_written += bytes_to_write;
1149  bytes_left -= bytes_to_write;
1150  *blockpos += bytes_to_write;
1151  xlogoff += bytes_to_write;
1152 
1153  /* Did we reach the end of a WAL segment? */
1154  if (XLogSegmentOffset(*blockpos, WalSegSz) == 0)
1155  {
1156  if (!close_walfile(stream, *blockpos))
1157  /* Error message written in close_walfile() */
1158  return false;
1159 
1160  xlogoff = 0;
1161 
1162  if (still_sending && stream->stream_stop(*blockpos, stream->timeline, true))
1163  {
1164  if (PQputCopyEnd(conn, NULL) <= 0 || PQflush(conn))
1165  {
1166  pg_log_error("could not send copy-end packet: %s",
1167  PQerrorMessage(conn));
1168  return false;
1169  }
1170  still_sending = false;
1171  return true; /* ignore the rest of this XLogData packet */
1172  }
1173  }
1174  }
1175  /* No more data left to write, receive next copy packet */
1176 
1177  return true;
1178 }
static bool open_walfile(StreamCtl *stream, XLogRecPtr startpoint)
Definition: receivelog.c:93
int64 fe_recvint64(char *buf)
Definition: streamutil.c:863
ssize_t(* write)(Walfile f, const void *buf, size_t count)
Definition: walmethods.h:76
#define XLogSegmentOffset(xlogptr, wal_segsz_bytes)

References close_walfile(), conn, copybuf, current_walfile_name, fe_recvint64(), WalWriteMethod::get_current_pos, WalWriteMethod::getlasterror, len, open_walfile(), pg_log_error, PQerrorMessage(), PQflush(), PQputCopyEnd(), still_sending, StreamCtl::stream_stop, StreamCtl::timeline, walfile, StreamCtl::walmethod, WalSegSz, WalWriteMethod::write, and XLogSegmentOffset.

Referenced by HandleCopyStream().

◆ ReadEndOfStreamingResult()

static bool ReadEndOfStreamingResult ( PGresult res,
XLogRecPtr startpos,
uint32 timeline 
)
static

Definition at line 705 of file receivelog.c.

706 {
707  uint32 startpos_xlogid,
708  startpos_xrecoff;
709 
710  /*----------
711  * The result set consists of one row and two columns, e.g:
712  *
713  * next_tli | next_tli_startpos
714  * ----------+-------------------
715  * 4 | 0/9949AE0
716  *
717  * next_tli is the timeline ID of the next timeline after the one that
718  * just finished streaming. next_tli_startpos is the WAL location where
719  * the server switched to it.
720  *----------
721  */
722  if (PQnfields(res) < 2 || PQntuples(res) != 1)
723  {
724  pg_log_error("unexpected result set after end-of-timeline: got %d rows and %d fields, expected %d rows and %d fields",
725  PQntuples(res), PQnfields(res), 1, 2);
726  return false;
727  }
728 
729  *timeline = atoi(PQgetvalue(res, 0, 0));
730  if (sscanf(PQgetvalue(res, 0, 1), "%X/%X", &startpos_xlogid,
731  &startpos_xrecoff) != 2)
732  {
733  pg_log_error("could not parse next timeline's starting point \"%s\"",
734  PQgetvalue(res, 0, 1));
735  return false;
736  }
737  *startpos = ((uint64) startpos_xlogid << 32) | startpos_xrecoff;
738 
739  return true;
740 }
unsigned int uint32
Definition: c.h:441
int PQntuples(const PGresult *res)
Definition: fe-exec.c:3248
char * PQgetvalue(const PGresult *res, int tup_num, int field_num)
Definition: fe-exec.c:3642
int PQnfields(const PGresult *res)
Definition: fe-exec.c:3256
static XLogRecPtr startpos

References pg_log_error, PQgetvalue(), PQnfields(), PQntuples(), res, and startpos.

Referenced by ReceiveXlogStream().

◆ ReceiveXlogStream()

bool ReceiveXlogStream ( PGconn conn,
StreamCtl stream 
)

Definition at line 459 of file receivelog.c.

460 {
461  char query[128];
462  char slotcmd[128];
463  PGresult *res;
464  XLogRecPtr stoppos;
465 
466  /*
467  * The caller should've checked the server version already, but doesn't do
468  * any harm to check it here too.
469  */
471  return false;
472 
473  /*
474  * Decide whether we want to report the flush position. If we report the
475  * flush position, the primary will know what WAL we'll possibly
476  * re-request, and it can then remove older WAL safely. We must always do
477  * that when we are using slots.
478  *
479  * Reporting the flush position makes one eligible as a synchronous
480  * replica. People shouldn't include generic names in
481  * synchronous_standby_names, but we've protected them against it so far,
482  * so let's continue to do so unless specifically requested.
483  */
484  if (stream->replication_slot != NULL)
485  {
486  reportFlushPosition = true;
487  sprintf(slotcmd, "SLOT \"%s\" ", stream->replication_slot);
488  }
489  else
490  {
491  if (stream->synchronous)
492  reportFlushPosition = true;
493  else
494  reportFlushPosition = false;
495  slotcmd[0] = 0;
496  }
497 
498  if (stream->sysidentifier != NULL)
499  {
500  char *sysidentifier = NULL;
501  TimeLineID servertli;
502 
503  /*
504  * Get the server system identifier and timeline, and validate them.
505  */
506  if (!RunIdentifySystem(conn, &sysidentifier, &servertli, NULL, NULL))
507  {
508  pg_free(sysidentifier);
509  return false;
510  }
511 
512  if (strcmp(stream->sysidentifier, sysidentifier) != 0)
513  {
514  pg_log_error("system identifier does not match between base backup and streaming connection");
515  pg_free(sysidentifier);
516  return false;
517  }
518  pg_free(sysidentifier);
519 
520  if (stream->timeline > servertli)
521  {
522  pg_log_error("starting timeline %u is not present in the server",
523  stream->timeline);
524  return false;
525  }
526  }
527 
528  /*
529  * initialize flush position to starting point, it's the caller's
530  * responsibility that that's sane.
531  */
532  lastFlushPosition = stream->startpos;
533 
534  while (1)
535  {
536  /*
537  * Fetch the timeline history file for this timeline, if we don't have
538  * it already. When streaming log to tar, this will always return
539  * false, as we are never streaming into an existing file and
540  * therefore there can be no pre-existing timeline history file.
541  */
542  if (!existsTimeLineHistoryFile(stream))
543  {
544  snprintf(query, sizeof(query), "TIMELINE_HISTORY %u", stream->timeline);
545  res = PQexec(conn, query);
547  {
548  /* FIXME: we might send it ok, but get an error */
549  pg_log_error("could not send replication command \"%s\": %s",
550  "TIMELINE_HISTORY", PQresultErrorMessage(res));
551  PQclear(res);
552  return false;
553  }
554 
555  /*
556  * The response to TIMELINE_HISTORY is a single row result set
557  * with two fields: filename and content
558  */
559  if (PQnfields(res) != 2 || PQntuples(res) != 1)
560  {
561  pg_log_warning("unexpected response to TIMELINE_HISTORY command: got %d rows and %d fields, expected %d rows and %d fields",
562  PQntuples(res), PQnfields(res), 1, 2);
563  }
564 
565  /* Write the history file to disk */
567  PQgetvalue(res, 0, 0),
568  PQgetvalue(res, 0, 1));
569 
570  PQclear(res);
571  }
572 
573  /*
574  * Before we start streaming from the requested location, check if the
575  * callback tells us to stop here.
576  */
577  if (stream->stream_stop(stream->startpos, stream->timeline, false))
578  return true;
579 
580  /* Initiate the replication stream at specified location */
581  snprintf(query, sizeof(query), "START_REPLICATION %s%X/%X TIMELINE %u",
582  slotcmd,
583  LSN_FORMAT_ARGS(stream->startpos),
584  stream->timeline);
585  res = PQexec(conn, query);
587  {
588  pg_log_error("could not send replication command \"%s\": %s",
589  "START_REPLICATION", PQresultErrorMessage(res));
590  PQclear(res);
591  return false;
592  }
593  PQclear(res);
594 
595  /* Stream the WAL */
596  res = HandleCopyStream(conn, stream, &stoppos);
597  if (res == NULL)
598  goto error;
599 
600  /*
601  * Streaming finished.
602  *
603  * There are two possible reasons for that: a controlled shutdown, or
604  * we reached the end of the current timeline. In case of
605  * end-of-timeline, the server sends a result set after Copy has
606  * finished, containing information about the next timeline. Read
607  * that, and restart streaming from the next timeline. In case of
608  * controlled shutdown, stop here.
609  */
611  {
612  /*
613  * End-of-timeline. Read the next timeline's ID and starting
614  * position. Usually, the starting position will match the end of
615  * the previous timeline, but there are corner cases like if the
616  * server had sent us half of a WAL record, when it was promoted.
617  * The new timeline will begin at the end of the last complete
618  * record in that case, overlapping the partial WAL record on the
619  * old timeline.
620  */
621  uint32 newtimeline;
622  bool parsed;
623 
624  parsed = ReadEndOfStreamingResult(res, &stream->startpos, &newtimeline);
625  PQclear(res);
626  if (!parsed)
627  goto error;
628 
629  /* Sanity check the values the server gave us */
630  if (newtimeline <= stream->timeline)
631  {
632  pg_log_error("server reported unexpected next timeline %u, following timeline %u",
633  newtimeline, stream->timeline);
634  goto error;
635  }
636  if (stream->startpos > stoppos)
637  {
638  pg_log_error("server stopped streaming timeline %u at %X/%X, but reported next timeline %u to begin at %X/%X",
639  stream->timeline, LSN_FORMAT_ARGS(stoppos),
640  newtimeline, LSN_FORMAT_ARGS(stream->startpos));
641  goto error;
642  }
643 
644  /* Read the final result, which should be CommandComplete. */
645  res = PQgetResult(conn);
647  {
648  pg_log_error("unexpected termination of replication stream: %s",
650  PQclear(res);
651  goto error;
652  }
653  PQclear(res);
654 
655  /*
656  * Loop back to start streaming from the new timeline. Always
657  * start streaming at the beginning of a segment.
658  */
659  stream->timeline = newtimeline;
660  stream->startpos = stream->startpos -
662  continue;
663  }
664  else if (PQresultStatus(res) == PGRES_COMMAND_OK)
665  {
666  PQclear(res);
667 
668  /*
669  * End of replication (ie. controlled shut down of the server).
670  *
671  * Check if the callback thinks it's OK to stop here. If not,
672  * complain.
673  */
674  if (stream->stream_stop(stoppos, stream->timeline, false))
675  return true;
676  else
677  {
678  pg_log_error("replication stream was terminated before stop point");
679  goto error;
680  }
681  }
682  else
683  {
684  /* Server returned an error. */
685  pg_log_error("unexpected termination of replication stream: %s",
687  PQclear(res);
688  goto error;
689  }
690  }
691 
692 error:
693  if (walfile != NULL && stream->walmethod->close(walfile, CLOSE_NO_RENAME) != 0)
694  pg_log_error("could not close file \"%s\": %s",
696  walfile = NULL;
697  return false;
698 }
char * PQresultErrorMessage(const PGresult *res)
Definition: fe-exec.c:3194
PGresult * PQexec(PGconn *conn, const char *query)
Definition: fe-exec.c:2193
@ PGRES_COPY_BOTH
Definition: libpq-fe.h:107
@ PGRES_COMMAND_OK
Definition: libpq-fe.h:95
@ PGRES_TUPLES_OK
Definition: libpq-fe.h:98
#define pg_log_warning(...)
Definition: pgfnames.c:24
#define sprintf
Definition: port.h:227
static PGresult * HandleCopyStream(PGconn *conn, StreamCtl *stream, XLogRecPtr *stoppos)
Definition: receivelog.c:751
static bool existsTimeLineHistoryFile(StreamCtl *stream)
Definition: receivelog.c:265
static bool writeTimeLineHistoryFile(StreamCtl *stream, char *filename, char *content)
Definition: receivelog.c:282
bool CheckServerVersionForStreaming(PGconn *conn)
Definition: receivelog.c:381
static bool ReadEndOfStreamingResult(PGresult *res, XLogRecPtr *startpos, uint32 *timeline)
Definition: receivelog.c:705
bool RunIdentifySystem(PGconn *conn, char **sysid, TimeLineID *starttli, XLogRecPtr *startpos, char **db_name)
Definition: streamutil.c:409
char * sysidentifier
Definition: receivelog.h:33
char * replication_slot
Definition: receivelog.h:48
#define LSN_FORMAT_ARGS(lsn)
Definition: xlogdefs.h:43
uint32 TimeLineID
Definition: xlogdefs.h:59

References CheckServerVersionForStreaming(), WalWriteMethod::close, CLOSE_NO_RENAME, conn, current_walfile_name, error(), existsTimeLineHistoryFile(), WalWriteMethod::getlasterror, HandleCopyStream(), lastFlushPosition, LSN_FORMAT_ARGS, pg_free(), pg_log_error, pg_log_warning, PGRES_COMMAND_OK, PGRES_COPY_BOTH, PGRES_TUPLES_OK, PQclear(), PQexec(), PQgetResult(), PQgetvalue(), PQnfields(), PQntuples(), PQresultErrorMessage(), PQresultStatus(), ReadEndOfStreamingResult(), StreamCtl::replication_slot, reportFlushPosition, res, RunIdentifySystem(), snprintf, sprintf, StreamCtl::startpos, StreamCtl::stream_stop, StreamCtl::synchronous, StreamCtl::sysidentifier, StreamCtl::timeline, walfile, StreamCtl::walmethod, WalSegSz, writeTimeLineHistoryFile(), and XLogSegmentOffset.

Referenced by LogStreamerMain(), and StreamLog().

◆ sendFeedback()

static bool sendFeedback ( PGconn conn,
XLogRecPtr  blockpos,
TimestampTz  now,
bool  replyRequested 
)
static

Definition at line 343 of file receivelog.c.

344 {
345  char replybuf[1 + 8 + 8 + 8 + 8 + 1];
346  int len = 0;
347 
348  replybuf[len] = 'r';
349  len += 1;
350  fe_sendint64(blockpos, &replybuf[len]); /* write */
351  len += 8;
353  fe_sendint64(lastFlushPosition, &replybuf[len]); /* flush */
354  else
355  fe_sendint64(InvalidXLogRecPtr, &replybuf[len]); /* flush */
356  len += 8;
357  fe_sendint64(InvalidXLogRecPtr, &replybuf[len]); /* apply */
358  len += 8;
359  fe_sendint64(now, &replybuf[len]); /* sendTime */
360  len += 8;
361  replybuf[len] = replyRequested ? 1 : 0; /* replyRequested */
362  len += 1;
363 
364  if (PQputCopyData(conn, replybuf, len) <= 0 || PQflush(conn))
365  {
366  pg_log_error("could not send feedback packet: %s",
368  return false;
369  }
370 
371  return true;
372 }
int PQputCopyData(PGconn *conn, const char *buffer, int nbytes)
Definition: fe-exec.c:2544
void fe_sendint64(int64 i, char *buf)
Definition: streamutil.c:852
#define InvalidXLogRecPtr
Definition: xlogdefs.h:28

References conn, fe_sendint64(), InvalidXLogRecPtr, lastFlushPosition, len, now(), pg_log_error, PQerrorMessage(), PQflush(), PQputCopyData(), and reportFlushPosition.

Referenced by HandleCopyStream(), and ProcessKeepaliveMsg().

◆ writeTimeLineHistoryFile()

static bool writeTimeLineHistoryFile ( StreamCtl stream,
char *  filename,
char *  content 
)
static

Definition at line 282 of file receivelog.c.

283 {
284  int size = strlen(content);
285  char histfname[MAXFNAMELEN];
286  Walfile *f;
287 
288  /*
289  * Check that the server's idea of how timeline history files should be
290  * named matches ours.
291  */
292  TLHistoryFileName(histfname, stream->timeline);
293  if (strcmp(histfname, filename) != 0)
294  {
295  pg_log_error("server reported unexpected history file name for timeline %u: %s",
296  stream->timeline, filename);
297  return false;
298  }
299 
300  f = stream->walmethod->open_for_write(histfname, ".tmp", 0);
301  if (f == NULL)
302  {
303  pg_log_error("could not create timeline history file \"%s\": %s",
304  histfname, stream->walmethod->getlasterror());
305  return false;
306  }
307 
308  if ((int) stream->walmethod->write(f, content, size) != size)
309  {
310  pg_log_error("could not write timeline history file \"%s\": %s",
311  histfname, stream->walmethod->getlasterror());
312 
313  /*
314  * If we fail to make the file, delete it to release disk space
315  */
316  stream->walmethod->close(f, CLOSE_UNLINK);
317 
318  return false;
319  }
320 
321  if (stream->walmethod->close(f, CLOSE_NORMAL) != 0)
322  {
323  pg_log_error("could not close file \"%s\": %s",
324  histfname, stream->walmethod->getlasterror());
325  return false;
326  }
327 
328  /* Maintain archive_status, check close_walfile() for details. */
329  if (stream->mark_done)
330  {
331  /* writes error message if failed */
332  if (!mark_file_as_archived(stream, histfname))
333  return false;
334  }
335 
336  return true;
337 }
static char * filename
Definition: pg_dumpall.c:92

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

Referenced by ReceiveXlogStream().

Variable Documentation

◆ current_walfile_name

char current_walfile_name[MAXPGPATH] = ""
static

◆ lastFlushPosition

XLogRecPtr lastFlushPosition = InvalidXLogRecPtr
static

◆ reportFlushPosition

bool reportFlushPosition = false
static

Definition at line 33 of file receivelog.c.

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

◆ still_sending

◆ walfile