Header And Logo

PostgreSQL
| The world's most advanced open source database.

Functions | Variables

receivelog.c File Reference

#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"
Include dependency graph for receivelog.c:

Go to the source code of this file.

Functions

static PGresultHandleCopyStream (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] = ""

Function Documentation

bool CheckServerVersionForStreaming ( PGconn conn  ) 

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, &copybuf, 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(&copybuf[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().

{
    int64       result;
    uint32      h32;
    uint32      l32;

    memcpy(&h32, buf, 4);
    memcpy(&l32, buf + 4, 4);
    h32 = ntohl(h32);
    l32 = ntohl(l32);

    result = h32;
    result <<= 32;
    result |= l32;

    return result;
}

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

{
    uint32      n32;

    /* High order half first, since we're doing MSB-first */
    n32 = (uint32) (i >> 32);
    n32 = htonl(n32);
    memcpy(&buf[0], &n32, 4);

    /* Now the low order half */
    n32 = (uint32) i;
    n32 = htonl(n32);
    memcpy(&buf[4], &n32, 4);
}

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;
}


Variable Documentation

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]