Header And Logo
|
#include "postgres_fe.h"#include <sys/stat.h>#include <sys/time.h>#include <sys/types.h>#include <unistd.h>#include <netinet/in.h>#include <arpa/inet.h>#include "libpq-fe.h"#include "access/xlog_internal.h"#include "receivelog.h"#include "streamutil.h"
Go to the source code of this file.
Functions | |
| static PGresult * | HandleCopyStream (PGconn *conn, XLogRecPtr startpos, uint32 timeline, char *basedir, stream_stop_callback stream_stop, int standby_message_timeout, char *partial_suffix, XLogRecPtr *stoppos) |
| static bool | open_walfile (XLogRecPtr startpoint, uint32 timeline, char *basedir, char *partial_suffix) |
| static bool | close_walfile (char *basedir, char *partial_suffix) |
| static int64 | localGetCurrentTimestamp (void) |
| static void | localTimestampDifference (int64 start_time, int64 stop_time, long *secs, int *microsecs) |
| static bool | localTimestampDifferenceExceeds (int64 start_time, int64 stop_time, int msec) |
| static bool | existsTimeLineHistoryFile (char *basedir, TimeLineID tli) |
| static bool | writeTimeLineHistoryFile (char *basedir, TimeLineID tli, char *filename, char *content) |
| static void | sendint64 (int64 i, char *buf) |
| static int64 | recvint64 (char *buf) |
| static bool | sendFeedback (PGconn *conn, XLogRecPtr blockpos, int64 now, bool replyRequested) |
| bool | CheckServerVersionForStreaming (PGconn *conn) |
| bool | ReceiveXlogStream (PGconn *conn, XLogRecPtr startpos, uint32 timeline, char *sysidentifier, char *basedir, stream_stop_callback stream_stop, int standby_message_timeout, char *partial_suffix) |
Variables | |
| static int | walfile = -1 |
| static char | current_walfile_name [MAXPGPATH] = "" |
Definition at line 446 of file receivelog.c.
References _, PQparameterStatus(), PQserverVersion(), and progname.
Referenced by BaseBackup(), ReceiveXlogStream(), and StreamLog().
{
int minServerMajor,
maxServerMajor;
int serverMajor;
/*
* The message format used in streaming replication changed in 9.3, so we
* cannot stream from older servers. And we don't support servers newer
* than the client; it might work, but we don't know, so err on the safe
* side.
*/
minServerMajor = 903;
maxServerMajor = PG_VERSION_NUM / 100;
serverMajor = PQserverVersion(conn) / 100;
if (serverMajor < minServerMajor || serverMajor > maxServerMajor)
{
const char *serverver = PQparameterStatus(conn, "server_version");
fprintf(stderr, _("%s: incompatible server version %s; streaming is only supported with server version %s\n"),
progname,
serverver ? serverver : "'unknown'",
"9.3");
return false;
}
return true;
}
| static bool close_walfile | ( | char * | basedir, | |
| char * | partial_suffix | |||
| ) | [static] |
Definition at line 133 of file receivelog.c.
References _, close, current_walfile_name, fsync, progname, snprintf(), strerror(), and walfile.
Referenced by HandleCopyStream().
{
off_t currpos;
if (walfile == -1)
return true;
currpos = lseek(walfile, 0, SEEK_CUR);
if (currpos == -1)
{
fprintf(stderr,
_("%s: could not determine seek position in file \"%s\": %s\n"),
progname, current_walfile_name, strerror(errno));
return false;
}
if (fsync(walfile) != 0)
{
fprintf(stderr, _("%s: could not fsync file \"%s\": %s\n"),
progname, current_walfile_name, strerror(errno));
return false;
}
if (close(walfile) != 0)
{
fprintf(stderr, _("%s: could not close file \"%s\": %s\n"),
progname, current_walfile_name, strerror(errno));
walfile = -1;
return false;
}
walfile = -1;
/*
* Rename the .partial file only if we've completed writing the whole
* segment or segment_complete is true.
*/
if (currpos == XLOG_SEG_SIZE && partial_suffix)
{
char oldfn[MAXPGPATH];
char newfn[MAXPGPATH];
snprintf(oldfn, sizeof(oldfn), "%s/%s%s", basedir, current_walfile_name, partial_suffix);
snprintf(newfn, sizeof(newfn), "%s/%s", basedir, current_walfile_name);
if (rename(oldfn, newfn) != 0)
{
fprintf(stderr, _("%s: could not rename file \"%s\": %s\n"),
progname, current_walfile_name, strerror(errno));
return false;
}
}
else if (partial_suffix)
fprintf(stderr,
_("%s: not renaming \"%s%s\", segment is not complete\n"),
progname, current_walfile_name, partial_suffix);
return true;
}
| static bool existsTimeLineHistoryFile | ( | char * | basedir, | |
| TimeLineID | tli | |||
| ) | [static] |
Definition at line 253 of file receivelog.c.
References _, close, PG_BINARY, progname, snprintf(), strerror(), and TLHistoryFileName.
Referenced by ReceiveXlogStream().
{
char path[MAXPGPATH];
char histfname[MAXFNAMELEN];
int fd;
/*
* Timeline 1 never has a history file. We treat that as if it existed,
* since we never need to stream it.
*/
if (tli == 1)
return true;
TLHistoryFileName(histfname, tli);
snprintf(path, sizeof(path), "%s/%s", basedir, histfname);
fd = open(path, O_RDONLY | PG_BINARY, 0);
if (fd < 0)
{
if (errno != ENOENT)
fprintf(stderr, _("%s: could not open timeline history file \"%s\": %s\n"),
progname, path, strerror(errno));
return false;
}
else
{
close(fd);
return true;
}
}
| static PGresult * HandleCopyStream | ( | PGconn * | conn, | |
| XLogRecPtr | startpos, | |||
| uint32 | timeline, | |||
| char * | basedir, | |||
| stream_stop_callback | stream_stop, | |||
| int | standby_message_timeout, | |||
| char * | partial_suffix, | |||
| XLogRecPtr * | stoppos | |||
| ) | [static] |
Definition at line 716 of file receivelog.c.
References _, close_walfile(), current_walfile_name, EINTR, error(), localGetCurrentTimestamp(), localTimestampDifference(), localTimestampDifferenceExceeds(), NULL, open_walfile(), PGRES_COPY_IN, PQconsumeInput(), PQerrorMessage(), PQflush(), PQfreemem(), PQgetCopyData(), PQgetResult(), PQputCopyEnd(), PQresultStatus(), PQsocket(), progname, recvint64(), select, sendFeedback(), strerror(), walfile, and write.
Referenced by ReceiveXlogStream().
{
char *copybuf = NULL;
int64 last_status = -1;
XLogRecPtr blockpos = startpos;
bool still_sending = true;
while (1)
{
int r;
int xlogoff;
int bytes_left;
int bytes_written;
int64 now;
int hdr_len;
if (copybuf != NULL)
{
PQfreemem(copybuf);
copybuf = NULL;
}
/*
* Check if we should continue streaming, or abort at this point.
*/
if (still_sending && stream_stop(blockpos, timeline, false))
{
if (!close_walfile(basedir, partial_suffix))
{
/* Potential error message is written by close_walfile */
goto error;
}
if (PQputCopyEnd(conn, NULL) <= 0 || PQflush(conn))
{
fprintf(stderr, _("%s: could not send copy-end packet: %s"),
progname, PQerrorMessage(conn));
goto error;
}
still_sending = false;
}
/*
* Potentially send a status message to the master
*/
now = localGetCurrentTimestamp();
if (still_sending && standby_message_timeout > 0 &&
localTimestampDifferenceExceeds(last_status, now,
standby_message_timeout))
{
/* Time to send feedback! */
if (!sendFeedback(conn, blockpos, now, false))
goto error;
last_status = now;
}
r = PQgetCopyData(conn, ©buf, 1);
if (r == 0)
{
/*
* No data available. Wait for some to appear, but not longer
* than the specified timeout, so that we can ping the server.
*/
fd_set input_mask;
struct timeval timeout;
struct timeval *timeoutptr;
FD_ZERO(&input_mask);
FD_SET(PQsocket(conn), &input_mask);
if (standby_message_timeout && still_sending)
{
int64 targettime;
long secs;
int usecs;
targettime = last_status + (standby_message_timeout - 1) * ((int64) 1000);
localTimestampDifference(now,
targettime,
&secs,
&usecs);
if (secs <= 0)
timeout.tv_sec = 1; /* Always sleep at least 1 sec */
else
timeout.tv_sec = secs;
timeout.tv_usec = usecs;
timeoutptr = &timeout;
}
else
timeoutptr = NULL;
r = select(PQsocket(conn) + 1, &input_mask, NULL, NULL, timeoutptr);
if (r == 0 || (r < 0 && errno == EINTR))
{
/*
* Got a timeout or signal. Continue the loop and either
* deliver a status packet to the server or just go back
* into blocking.
*/
continue;
}
else if (r < 0)
{
fprintf(stderr, _("%s: select() failed: %s\n"),
progname, strerror(errno));
goto error;
}
/* Else there is actually data on the socket */
if (PQconsumeInput(conn) == 0)
{
fprintf(stderr,
_("%s: could not receive data from WAL stream: %s"),
progname, PQerrorMessage(conn));
goto error;
}
continue;
}
if (r == -1)
{
PGresult *res = PQgetResult(conn);
/*
* The server closed its end of the copy stream. If we haven't
* closed ours already, we need to do so now, unless the server
* threw an error, in which case we don't.
*/
if (still_sending)
{
if (!close_walfile(basedir, partial_suffix))
{
/* Error message written in close_walfile() */
goto error;
}
if (PQresultStatus(res) == PGRES_COPY_IN)
{
if (PQputCopyEnd(conn, NULL) <= 0 || PQflush(conn))
{
fprintf(stderr,
_("%s: could not send copy-end packet: %s"),
progname, PQerrorMessage(conn));
goto error;
}
res = PQgetResult(conn);
}
still_sending = false;
}
if (copybuf != NULL)
PQfreemem(copybuf);
*stoppos = blockpos;
return res;
}
if (r == -2)
{
fprintf(stderr, _("%s: could not read COPY data: %s"),
progname, PQerrorMessage(conn));
goto error;
}
/* Check the message type. */
if (copybuf[0] == 'k')
{
int pos;
bool replyRequested;
/*
* Parse the keepalive message, enclosed in the CopyData message.
* We just check if the server requested a reply, and ignore the
* rest.
*/
pos = 1; /* skip msgtype 'k' */
pos += 8; /* skip walEnd */
pos += 8; /* skip sendTime */
if (r < pos + 1)
{
fprintf(stderr, _("%s: streaming header too small: %d\n"),
progname, r);
goto error;
}
replyRequested = copybuf[pos];
/* If the server requested an immediate reply, send one. */
if (replyRequested && still_sending)
{
now = localGetCurrentTimestamp();
if (!sendFeedback(conn, blockpos, now, false))
goto error;
last_status = now;
}
}
else if (copybuf[0] == 'w')
{
/*
* Once we've decided we don't want to receive any more, just
* ignore any subsequent XLogData messages.
*/
if (!still_sending)
continue;
/*
* Read the header of the XLogData message, enclosed in the
* CopyData message. We only need the WAL location field
* (dataStart), the rest of the header is ignored.
*/
hdr_len = 1; /* msgtype 'w' */
hdr_len += 8; /* dataStart */
hdr_len += 8; /* walEnd */
hdr_len += 8; /* sendTime */
if (r < hdr_len + 1)
{
fprintf(stderr, _("%s: streaming header too small: %d\n"),
progname, r);
goto error;
}
blockpos = recvint64(©buf[1]);
/* Extract WAL location for this block */
xlogoff = blockpos % XLOG_SEG_SIZE;
/*
* Verify that the initial location in the stream matches where
* we think we are.
*/
if (walfile == -1)
{
/* No file open yet */
if (xlogoff != 0)
{
fprintf(stderr,
_("%s: received transaction log record for offset %u with no file open\n"),
progname, xlogoff);
goto error;
}
}
else
{
/* More data in existing segment */
/* XXX: store seek value don't reseek all the time */
if (lseek(walfile, 0, SEEK_CUR) != xlogoff)
{
fprintf(stderr,
_("%s: got WAL data offset %08x, expected %08x\n"),
progname, xlogoff, (int) lseek(walfile, 0, SEEK_CUR));
goto error;
}
}
bytes_left = r - hdr_len;
bytes_written = 0;
while (bytes_left)
{
int bytes_to_write;
/*
* If crossing a WAL boundary, only write up until we reach
* XLOG_SEG_SIZE.
*/
if (xlogoff + bytes_left > XLOG_SEG_SIZE)
bytes_to_write = XLOG_SEG_SIZE - xlogoff;
else
bytes_to_write = bytes_left;
if (walfile == -1)
{
if (!open_walfile(blockpos, timeline,
basedir, partial_suffix))
{
/* Error logged by open_walfile */
goto error;
}
}
if (write(walfile,
copybuf + hdr_len + bytes_written,
bytes_to_write) != bytes_to_write)
{
fprintf(stderr,
_("%s: could not write %u bytes to WAL file \"%s\": %s\n"),
progname, bytes_to_write, current_walfile_name,
strerror(errno));
goto error;
}
/* Write was successful, advance our position */
bytes_written += bytes_to_write;
bytes_left -= bytes_to_write;
blockpos += bytes_to_write;
xlogoff += bytes_to_write;
/* Did we reach the end of a WAL segment? */
if (blockpos % XLOG_SEG_SIZE == 0)
{
if (!close_walfile(basedir, partial_suffix))
/* Error message written in close_walfile() */
goto error;
xlogoff = 0;
if (still_sending && stream_stop(blockpos, timeline, false))
{
if (PQputCopyEnd(conn, NULL) <= 0 || PQflush(conn))
{
fprintf(stderr, _("%s: could not send copy-end packet: %s"),
progname, PQerrorMessage(conn));
goto error;
}
still_sending = false;
break; /* ignore the rest of this XLogData packet */
}
}
}
/* No more data left to write, receive next copy packet */
}
else
{
fprintf(stderr, _("%s: unrecognized streaming header: \"%c\"\n"),
progname, copybuf[0]);
goto error;
}
}
error:
if (copybuf != NULL)
PQfreemem(copybuf);
return NULL;
}
| static int64 localGetCurrentTimestamp | ( | void | ) | [static] |
Definition at line 198 of file receivelog.c.
References gettimeofday(), NULL, POSTGRES_EPOCH_JDATE, and UNIX_EPOCH_JDATE.
Referenced by HandleCopyStream().
{
int64 result;
struct timeval tp;
gettimeofday(&tp, NULL);
result = (int64) tp.tv_sec -
((POSTGRES_EPOCH_JDATE - UNIX_EPOCH_JDATE) * SECS_PER_DAY);
result = (result * USECS_PER_SEC) + tp.tv_usec;
return result;
}
| static void localTimestampDifference | ( | int64 | start_time, | |
| int64 | stop_time, | |||
| long * | secs, | |||
| int * | microsecs | |||
| ) | [static] |
Definition at line 218 of file receivelog.c.
References USECS_PER_SEC.
Referenced by HandleCopyStream().
{
int64 diff = stop_time - start_time;
if (diff <= 0)
{
*secs = 0;
*microsecs = 0;
}
else
{
*secs = (long) (diff / USECS_PER_SEC);
*microsecs = (int) (diff % USECS_PER_SEC);
}
}
| static bool localTimestampDifferenceExceeds | ( | int64 | start_time, | |
| int64 | stop_time, | |||
| int | msec | |||
| ) | [static] |
Definition at line 240 of file receivelog.c.
References INT64CONST.
Referenced by HandleCopyStream().
{
int64 diff = stop_time - start_time;
return (diff >= msec * INT64CONST(1000));
}
| static bool open_walfile | ( | XLogRecPtr | startpoint, | |
| uint32 | timeline, | |||
| char * | basedir, | |||
| char * | partial_suffix | |||
| ) | [static] |
Definition at line 47 of file receivelog.c.
References _, close, current_walfile_name, fn(), free, PG_BINARY, pg_malloc0(), progname, snprintf(), strerror(), unlink(), walfile, write, XLByteToSeg, XLogFileName, and XLogSegSize.
Referenced by HandleCopyStream().
{
int f;
char fn[MAXPGPATH];
struct stat statbuf;
char *zerobuf;
int bytes;
XLogSegNo segno;
XLByteToSeg(startpoint, segno);
XLogFileName(current_walfile_name, timeline, segno);
snprintf(fn, sizeof(fn), "%s/%s%s", basedir, current_walfile_name,
partial_suffix ? partial_suffix : "");
f = open(fn, O_WRONLY | O_CREAT | PG_BINARY, S_IRUSR | S_IWUSR);
if (f == -1)
{
fprintf(stderr,
_("%s: could not open transaction log file \"%s\": %s\n"),
progname, fn, strerror(errno));
return false;
}
/*
* Verify that the file is either empty (just created), or a complete
* XLogSegSize segment. Anything in between indicates a corrupt file.
*/
if (fstat(f, &statbuf) != 0)
{
fprintf(stderr,
_("%s: could not stat transaction log file \"%s\": %s\n"),
progname, fn, strerror(errno));
close(f);
return false;
}
if (statbuf.st_size == XLogSegSize)
{
/* File is open and ready to use */
walfile = f;
return true;
}
if (statbuf.st_size != 0)
{
fprintf(stderr,
_("%s: transaction log file \"%s\" has %d bytes, should be 0 or %d\n"),
progname, fn, (int) statbuf.st_size, XLogSegSize);
close(f);
return false;
}
/* New, empty, file. So pad it to 16Mb with zeroes */
zerobuf = pg_malloc0(XLOG_BLCKSZ);
for (bytes = 0; bytes < XLogSegSize; bytes += XLOG_BLCKSZ)
{
if (write(f, zerobuf, XLOG_BLCKSZ) != XLOG_BLCKSZ)
{
fprintf(stderr,
_("%s: could not pad transaction log file \"%s\": %s\n"),
progname, fn, strerror(errno));
free(zerobuf);
close(f);
unlink(fn);
return false;
}
}
free(zerobuf);
if (lseek(f, SEEK_SET, 0) != 0)
{
fprintf(stderr,
_("%s: could not seek to beginning of transaction log file \"%s\": %s\n"),
progname, fn, strerror(errno));
close(f);
return false;
}
walfile = f;
return true;
}
| bool ReceiveXlogStream | ( | PGconn * | conn, | |
| XLogRecPtr | startpos, | |||
| uint32 | timeline, | |||
| char * | sysidentifier, | |||
| char * | basedir, | |||
| stream_stop_callback | stream_stop, | |||
| int | standby_message_timeout, | |||
| char * | partial_suffix | |||
| ) |
Definition at line 503 of file receivelog.c.
References _, CheckServerVersionForStreaming(), close, current_walfile_name, error(), existsTimeLineHistoryFile(), HandleCopyStream(), NULL, PGRES_COMMAND_OK, PGRES_COPY_BOTH, PGRES_TUPLES_OK, PQclear(), PQerrorMessage(), PQexec(), PQgetResult(), PQgetvalue(), PQnfields(), PQntuples(), PQresultErrorMessage(), PQresultStatus(), progname, snprintf(), strerror(), walfile, and writeTimeLineHistoryFile().
Referenced by LogStreamerMain(), and StreamLog().
{
char query[128];
PGresult *res;
XLogRecPtr stoppos;
/*
* The caller should've checked the server version already, but doesn't do
* any harm to check it here too.
*/
if (!CheckServerVersionForStreaming(conn))
return false;
if (sysidentifier != NULL)
{
/* Validate system identifier hasn't changed */
res = PQexec(conn, "IDENTIFY_SYSTEM");
if (PQresultStatus(res) != PGRES_TUPLES_OK)
{
fprintf(stderr,
_("%s: could not send replication command \"%s\": %s"),
progname, "IDENTIFY_SYSTEM", PQerrorMessage(conn));
PQclear(res);
return false;
}
if (PQnfields(res) != 3 || PQntuples(res) != 1)
{
fprintf(stderr,
_("%s: could not identify system: got %d rows and %d fields, expected %d rows and %d fields\n"),
progname, PQntuples(res), PQnfields(res), 1, 3);
PQclear(res);
return false;
}
if (strcmp(sysidentifier, PQgetvalue(res, 0, 0)) != 0)
{
fprintf(stderr,
_("%s: system identifier does not match between base backup and streaming connection\n"),
progname);
PQclear(res);
return false;
}
if (timeline > atoi(PQgetvalue(res, 0, 1)))
{
fprintf(stderr,
_("%s: starting timeline %u is not present in the server\n"),
progname, timeline);
PQclear(res);
return false;
}
PQclear(res);
}
while (1)
{
/*
* Fetch the timeline history file for this timeline, if we don't
* have it already.
*/
if (!existsTimeLineHistoryFile(basedir, timeline))
{
snprintf(query, sizeof(query), "TIMELINE_HISTORY %u", timeline);
res = PQexec(conn, query);
if (PQresultStatus(res) != PGRES_TUPLES_OK)
{
/* FIXME: we might send it ok, but get an error */
fprintf(stderr, _("%s: could not send replication command \"%s\": %s"),
progname, "TIMELINE_HISTORY", PQresultErrorMessage(res));
PQclear(res);
return false;
}
/*
* The response to TIMELINE_HISTORY is a single row result set
* with two fields: filename and content
*/
if (PQnfields(res) != 2 || PQntuples(res) != 1)
{
fprintf(stderr,
_("%s: unexpected response to TIMELINE_HISTORY command: got %d rows and %d fields, expected %d rows and %d fields\n"),
progname, PQntuples(res), PQnfields(res), 1, 2);
}
/* Write the history file to disk */
writeTimeLineHistoryFile(basedir, timeline,
PQgetvalue(res, 0, 0),
PQgetvalue(res, 0, 1));
PQclear(res);
}
/*
* Before we start streaming from the requested location, check
* if the callback tells us to stop here.
*/
if (stream_stop(startpos, timeline, false))
return true;
/* Initiate the replication stream at specified location */
snprintf(query, sizeof(query), "START_REPLICATION %X/%X TIMELINE %u",
(uint32) (startpos >> 32), (uint32) startpos,
timeline);
res = PQexec(conn, query);
if (PQresultStatus(res) != PGRES_COPY_BOTH)
{
fprintf(stderr, _("%s: could not send replication command \"%s\": %s"),
progname, "START_REPLICATION", PQresultErrorMessage(res));
PQclear(res);
return false;
}
PQclear(res);
/* Stream the WAL */
res = HandleCopyStream(conn, startpos, timeline, basedir, stream_stop,
standby_message_timeout, partial_suffix,
&stoppos);
if (res == NULL)
goto error;
/*
* Streaming finished.
*
* There are two possible reasons for that: a controlled shutdown,
* or we reached the end of the current timeline. In case of
* end-of-timeline, the server sends a result set after Copy has
* finished, containing the next timeline's ID. Read that, and
* restart streaming from the next timeline.
*/
if (PQresultStatus(res) == PGRES_TUPLES_OK)
{
/*
* End-of-timeline. Read the next timeline's ID.
*/
uint32 newtimeline;
newtimeline = atoi(PQgetvalue(res, 0, 0));
PQclear(res);
if (newtimeline <= timeline)
{
/* shouldn't happen */
fprintf(stderr,
"server reported unexpected next timeline %u, following timeline %u\n",
newtimeline, timeline);
goto error;
}
/* Read the final result, which should be CommandComplete. */
res = PQgetResult(conn);
if (PQresultStatus(res) != PGRES_COMMAND_OK)
{
fprintf(stderr,
_("%s: unexpected termination of replication stream: %s"),
progname, PQresultErrorMessage(res));
goto error;
}
PQclear(res);
/*
* Loop back to start streaming from the new timeline.
* Always start streaming at the beginning of a segment.
*/
timeline = newtimeline;
startpos = stoppos - (stoppos % XLOG_SEG_SIZE);
continue;
}
else if (PQresultStatus(res) == PGRES_COMMAND_OK)
{
/*
* End of replication (ie. controlled shut down of the server).
*
* Check if the callback thinks it's OK to stop here. If not,
* complain.
*/
if (stream_stop(stoppos, timeline, false))
return true;
else
{
fprintf(stderr, _("%s: replication stream was terminated before stop point\n"),
progname);
goto error;
}
}
else
{
/* Server returned an error. */
fprintf(stderr,
_("%s: unexpected termination of replication stream: %s"),
progname, PQresultErrorMessage(res));
goto error;
}
}
error:
if (walfile != -1 && close(walfile) != 0)
fprintf(stderr, _("%s: could not close file \"%s\": %s\n"),
progname, current_walfile_name, strerror(errno));
walfile = -1;
return false;
}
| static int64 recvint64 | ( | char * | buf | ) | [static] |
Definition at line 389 of file receivelog.c.
Referenced by HandleCopyStream().
| static bool sendFeedback | ( | PGconn * | conn, | |
| XLogRecPtr | blockpos, | |||
| int64 | now, | |||
| bool | replyRequested | |||
| ) | [static] |
Definition at line 411 of file receivelog.c.
References _, InvalidXLogRecPtr, PQerrorMessage(), PQflush(), PQputCopyData(), progname, and sendint64().
Referenced by HandleCopyStream().
{
char replybuf[1 + 8 + 8 + 8 + 8 + 1];
int len = 0;
replybuf[len] = 'r';
len += 1;
sendint64(blockpos, &replybuf[len]); /* write */
len += 8;
sendint64(InvalidXLogRecPtr, &replybuf[len]); /* flush */
len += 8;
sendint64(InvalidXLogRecPtr, &replybuf[len]); /* apply */
len += 8;
sendint64(now, &replybuf[len]); /* sendTime */
len += 8;
replybuf[len] = replyRequested ? 1 : 0; /* replyRequested */
len += 1;
if (PQputCopyData(conn, replybuf, len) <= 0 || PQflush(conn))
{
fprintf(stderr, _("%s: could not send feedback packet: %s"),
progname, PQerrorMessage(conn));
return false;
}
return true;
}
| static void sendint64 | ( | int64 | i, | |
| char * | buf | |||
| ) | [static] |
Definition at line 370 of file receivelog.c.
Referenced by sendFeedback().
| static bool writeTimeLineHistoryFile | ( | char * | basedir, | |
| TimeLineID | tli, | |||
| char * | filename, | |||
| char * | content | |||
| ) | [static] |
Definition at line 286 of file receivelog.c.
References _, close, fsync, MAXPGPATH, PG_BINARY, progname, snprintf(), strerror(), TLHistoryFileName, unlink(), and write.
{
int size = strlen(content);
char path[MAXPGPATH];
char tmppath[MAXPGPATH];
char histfname[MAXFNAMELEN];
int fd;
/*
* Check that the server's idea of how timeline history files should be
* named matches ours.
*/
TLHistoryFileName(histfname, tli);
if (strcmp(histfname, filename) != 0)
{
fprintf(stderr, _("%s: server reported unexpected history file name for timeline %u: %s\n"),
progname, tli, filename);
return false;
}
/*
* Write into a temp file name.
*/
snprintf(tmppath, MAXPGPATH, "%s.tmp", path);
unlink(tmppath);
fd = open(tmppath, O_WRONLY | O_CREAT | PG_BINARY, S_IRUSR | S_IWUSR);
if (fd < 0)
{
fprintf(stderr, _("%s: could not create timeline history file \"%s\": %s\n"),
progname, tmppath, strerror(errno));
return false;
}
errno = 0;
if ((int) write(fd, content, size) != size)
{
int save_errno = errno;
/*
* If we fail to make the file, delete it to release disk space
*/
unlink(tmppath);
errno = save_errno;
fprintf(stderr, _("%s: could not write timeline history file \"%s\": %s\n"),
progname, tmppath, strerror(errno));
return false;
}
if (fsync(fd) != 0)
{
fprintf(stderr, _("%s: could not fsync file \"%s\": %s\n"),
progname, tmppath, strerror(errno));
return false;
}
if (close(fd) != 0)
{
fprintf(stderr, _("%s: could not close file \"%s\": %s\n"),
progname, tmppath, strerror(errno));
return false;
}
/*
* Now move the completed history file into place with its final name.
*/
snprintf(path, sizeof(path), "%s/%s", basedir, histfname);
if (rename(tmppath, path) < 0)
{
fprintf(stderr, _("%s: could not rename file \"%s\" to \"%s\": %s\n"),
progname, tmppath, path, strerror(errno));
return false;
}
return true;
}
char current_walfile_name[MAXPGPATH] = "" [static] |
Definition at line 33 of file receivelog.c.
Referenced by close_walfile(), HandleCopyStream(), open_walfile(), and ReceiveXlogStream().
int walfile = -1 [static] |
Definition at line 32 of file receivelog.c.
Referenced by CleanupPriorWALFiles(), close_walfile(), HandleCopyStream(), open_walfile(), and ReceiveXlogStream().
1.7.1