receivelog.c File Reference

#include "postgres_fe.h"
#include <sys/select.h>
#include <sys/stat.h>
#include <unistd.h>
#include "access/xlog_internal.h"
#include "common/logging.h"
#include "libpq-fe.h"
#include "libpq/protocol.h"
#include "receivelog.h"
#include "streamutil.h"

Include dependency graph for receivelog.c:

Go to the source code of this file.

Functions
static PGresult *	HandleCopyStream (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	ProcessWALDataMsg (PGconn *conn, StreamCtl *stream, char *copybuf, int len, XLogRecPtr *blockpos)
static PGresult *	HandleEndOfCopyStream (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 Walfile *	walfile = NULL
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 1249 of file receivelog.c.

{
    TimestampTz status_targettime = 0;
    long        sleeptime;
 
    if (standby_message_timeout && still_sending)
        status_targettime = last_status +
            (standby_message_timeout - 1) * ((int64) 1000);
 
    if (status_targettime > 0)
    {
        long        secs;
        int         usecs;
 
        feTimestampDifference(now,
                              status_targettime,
                              &secs,
                              &usecs);
        /* Always sleep at least 1 sec */
        if (secs <= 0)
        {
            secs = 1;
            usecs = 0;
        }
 
        sleeptime = secs * 1000 + usecs / 1000;
    }
    else
        sleeptime = -1;
 
    return sleeptime;
}

References fb(), 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 1224 of file receivelog.c.

{
    if (still_sending && stream->stream_stop(blockpos, stream->timeline, false))
    {
        if (!close_walfile(stream, blockpos))
        {
            /* Potential error message is written by close_walfile */
            return false;
        }
        if (PQputCopyEnd(conn, NULL) <= 0 || PQflush(conn))
        {
            pg_log_error("could not send copy-end packet: %s",
                         PQerrorMessage(conn));
            return false;
        }
        still_sending = false;
    }
 
    return true;
}

References close_walfile(), conn, fb(), 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 375 of file receivelog.c.

{
    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)
    {
        const char *serverver = PQparameterStatus(conn, "server_version");
 
        pg_log_error("incompatible server version %s; client does not support streaming from server versions older than %s",
                     serverver ? serverver : "'unknown'",
                     "9.3");
        return false;
    }
    else if (serverMajor > maxServerMajor)
    {
        const char *serverver = PQparameterStatus(conn, "server_version");
 
        pg_log_error("incompatible server version %s; client does not support streaming from server versions newer than %s",
                     serverver ? serverver : "'unknown'",
                     PG_VERSION);
        return false;
    }
    return true;
}

References conn, fb(), 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 192 of file receivelog.c.

{
    char       *fn;
    pgoff_t     currpos;
    int         r;
    char        walfile_name[MAXPGPATH];
 
    if (walfile == NULL)
        return true;
 
    strlcpy(walfile_name, walfile->pathname, MAXPGPATH);
    currpos = walfile->currpos;
 
    /* Note that this considers the compression used if necessary */
    fn = stream->walmethod->ops->get_file_name(stream->walmethod,
                                               walfile_name,
                                               stream->partial_suffix);
 
    if (stream->partial_suffix)
    {
        if (currpos == WalSegSz)
            r = stream->walmethod->ops->close(walfile, CLOSE_NORMAL);
        else
        {
            pg_log_info("not renaming \"%s\", segment is not complete", fn);
            r = stream->walmethod->ops->close(walfile, CLOSE_NO_RENAME);
        }
    }
    else
        r = stream->walmethod->ops->close(walfile, CLOSE_NORMAL);
 
    walfile = NULL;
 
    if (r != 0)
    {
        pg_log_error("could not close file \"%s\": %s",
                     fn, GetLastWalMethodError(stream->walmethod));
 
        pg_free(fn);
        return false;
    }
 
    pg_free(fn);
 
    /*
     * Mark file as archived if requested by the caller - pg_basebackup needs
     * to do so as files can otherwise get archived again after promotion of a
     * new node. This is in line with walreceiver.c always doing a
     * XLogArchiveForceDone() after a complete segment.
     */
    if (currpos == WalSegSz && stream->mark_done)
    {
        /* writes error message if failed */
        if (!mark_file_as_archived(stream, walfile_name))
            return false;
    }
 
    lastFlushPosition = pos;
    return true;
}

References WalWriteMethodOps::close, CLOSE_NO_RENAME, CLOSE_NORMAL, Walfile::currpos, fb(), fn(), WalWriteMethodOps::get_file_name, GetLastWalMethodError(), lastFlushPosition, StreamCtl::mark_done, mark_file_as_archived(), MAXPGPATH, WalWriteMethod::ops, StreamCtl::partial_suffix, Walfile::pathname, pg_free(), pg_log_error, pg_log_info, strlcpy(), walfile, StreamCtl::walmethod, and WalSegSz.

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

CopyStreamPoll()

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

static

Definition at line 884 of file receivelog.c.

{
    int         ret;
    fd_set      input_mask;
    int         connsocket;
    int         maxfd;
    struct timeval timeout;
    struct timeval *timeoutptr;
 
    connsocket = PQsocket(conn);
    if (connsocket < 0)
    {
        pg_log_error("invalid socket: %s", PQerrorMessage(conn));
        return -1;
    }
 
    FD_ZERO(&input_mask);
    FD_SET(connsocket, &input_mask);
    maxfd = connsocket;
    if (stop_socket != PGINVALID_SOCKET)
    {
        FD_SET(stop_socket, &input_mask);
        maxfd = Max(maxfd, stop_socket);
    }
 
    if (timeout_ms < 0)
        timeoutptr = NULL;
    else
    {
        timeout.tv_sec = timeout_ms / 1000L;
        timeout.tv_usec = (timeout_ms % 1000L) * 1000L;
        timeoutptr = &timeout;
    }
 
    ret = select(maxfd + 1, &input_mask, NULL, NULL, timeoutptr);
 
    if (ret < 0)
    {
        if (errno == EINTR)
            return 0;           /* Got a signal, so not an error */
        pg_log_error("%s() failed: %m", "select");
        return -1;
    }
    if (ret > 0 && FD_ISSET(connsocket, &input_mask))
        return 1;               /* Got input on connection socket */
 
    return 0;                   /* Got timeout or input on stop_socket */
}

References conn, EINTR, fb(), 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 946 of file receivelog.c.

{
    char       *copybuf = NULL;
    int         rawlen;
 
    /* Caller should have cleared any prior buffer */
    Assert(*buffer == NULL);
 
    /* Try to receive a CopyData message */
    rawlen = PQgetCopyData(conn, &copybuf, 1);
    if (rawlen == 0)
    {
        int         ret;
 
        /*
         * No data available.  Wait for some to appear, but not longer than
         * the specified timeout, so that we can ping the server.  Also stop
         * waiting if input appears on stop_socket.
         */
        ret = CopyStreamPoll(conn, timeout, stop_socket);
        if (ret <= 0)
            return ret;
 
        /* Now there is actually data on the socket */
        if (PQconsumeInput(conn) == 0)
        {
            pg_log_error("could not receive data from WAL stream: %s",
                         PQerrorMessage(conn));
            return -1;
        }
 
        /* Now that we've consumed some input, try again */
        rawlen = PQgetCopyData(conn, &copybuf, 1);
        if (rawlen == 0)
            return 0;
    }
    if (rawlen == -1)           /* end-of-streaming or error */
        return -2;
    if (rawlen == -2)
    {
        pg_log_error("could not read COPY data: %s", PQerrorMessage(conn));
        return -1;
    }
 
    /* Return received messages to caller */
    *buffer = copybuf;
    return rawlen;
}

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

Referenced by HandleCopyStream().

existsTimeLineHistoryFile()

static bool existsTimeLineHistoryFile ( StreamCtl *  stream )

static

Definition at line 258 of file receivelog.c.

{
    char        histfname[MAXFNAMELEN];
 
    /*
     * Timeline 1 never has a history file. We treat that as if it existed,
     * since we never need to stream it.
     */
    if (stream->timeline == 1)
        return true;
 
    TLHistoryFileName(histfname, stream->timeline);
 
    return stream->walmethod->ops->existsfile(stream->walmethod, histfname);
}

References WalWriteMethodOps::existsfile, fb(), MAXFNAMELEN, WalWriteMethod::ops, 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.

{
    char       *copybuf = NULL;
    TimestampTz last_status = -1;
    XLogRecPtr  blockpos = stream->startpos;
 
    still_sending = true;
 
    while (1)
    {
        int         r;
        TimestampTz now;
        long        sleeptime;
 
        /*
         * Check if we should continue streaming, or abort at this point.
         */
        if (!CheckCopyStreamStop(conn, stream, blockpos))
            goto error;
 
        now = feGetCurrentTimestamp();
 
        /*
         * If synchronous option is true, issue sync command as soon as there
         * are WAL data which has not been flushed yet.
         */
        if (stream->synchronous && lastFlushPosition < blockpos && walfile != NULL)
        {
            if (stream->walmethod->ops->sync(walfile) != 0)
                pg_fatal("could not fsync file \"%s\": %s",
                         walfile->pathname, GetLastWalMethodError(stream->walmethod));
            lastFlushPosition = blockpos;
 
            /*
             * Send feedback so that the server sees the latest WAL locations
             * immediately.
             */
            if (!sendFeedback(conn, blockpos, now, false))
                goto error;
            last_status = now;
        }
 
        /*
         * Potentially send a status message to the primary
         */
        if (still_sending && stream->standby_message_timeout > 0 &&
            feTimestampDifferenceExceeds(last_status, now,
                                         stream->standby_message_timeout))
        {
            /* Time to send feedback! */
            if (!sendFeedback(conn, blockpos, now, false))
                goto error;
            last_status = now;
        }
 
        /*
         * Calculate how long send/receive loops should sleep
         */
        sleeptime = CalculateCopyStreamSleeptime(now, stream->standby_message_timeout,
                                                 last_status);
 
        /* Done with any prior message */
        PQfreemem(copybuf);
        copybuf = NULL;
 
        r = CopyStreamReceive(conn, sleeptime, stream->stop_socket, &copybuf);
        while (r != 0)
        {
            if (r == -1)
                goto error;
            if (r == -2)
            {
                PGresult   *res = HandleEndOfCopyStream(conn, stream, copybuf, blockpos, stoppos);
 
                if (res == NULL)
                    goto error;
                PQfreemem(copybuf);
                return res;
            }
 
            /* Check the message type. */
            if (copybuf[0] == PqReplMsg_Keepalive)
            {
                if (!ProcessKeepaliveMsg(conn, stream, copybuf, r, blockpos,
                                         &last_status))
                    goto error;
            }
            else if (copybuf[0] == PqReplMsg_WALData)
            {
                if (!ProcessWALDataMsg(conn, stream, copybuf, r, &blockpos))
                    goto error;
 
                /*
                 * Check if we should continue streaming, or abort at this
                 * point.
                 */
                if (!CheckCopyStreamStop(conn, stream, blockpos))
                    goto error;
            }
            else
            {
                pg_log_error("unrecognized streaming header: \"%c\"",
                             copybuf[0]);
                goto error;
            }
 
            /* Done with that message */
            PQfreemem(copybuf);
            copybuf = NULL;
 
            /*
             * Process the received data, and any subsequent data we can read
             * without blocking.
             */
            r = CopyStreamReceive(conn, 0, stream->stop_socket, &copybuf);
        }
    }
 
error:
    PQfreemem(copybuf);
    return NULL;
}

References CalculateCopyStreamSleeptime(), CheckCopyStreamStop(), conn, copybuf, CopyStreamReceive(), error(), fb(), feGetCurrentTimestamp(), feTimestampDifferenceExceeds(), GetLastWalMethodError(), HandleEndOfCopyStream(), lastFlushPosition, now(), WalWriteMethod::ops, Walfile::pathname, pg_fatal, pg_log_error, PQfreemem(), PqReplMsg_Keepalive, PqReplMsg_WALData, ProcessKeepaliveMsg(), ProcessWALDataMsg(), sendFeedback(), StreamCtl::standby_message_timeout, StreamCtl::startpos, still_sending, StreamCtl::stop_socket, WalWriteMethodOps::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 1185 of file receivelog.c.

{
    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(stream, blockpos))
        {
            /* Error message written in close_walfile() */
            PQclear(res);
            return NULL;
        }
        if (PQresultStatus(res) == PGRES_COPY_IN)
        {
            if (PQputCopyEnd(conn, NULL) <= 0 || PQflush(conn))
            {
                pg_log_error("could not send copy-end packet: %s",
                             PQerrorMessage(conn));
                PQclear(res);
                return NULL;
            }
            res = PQgetResult(conn);
        }
        still_sending = false;
    }
    *stoppos = blockpos;
    return res;
}

References close_walfile(), conn, fb(), pg_log_error, PGRES_COPY_IN, PQclear, PQerrorMessage(), PQflush(), PQgetResult, PQputCopyEnd(), PQresultStatus, 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 54 of file receivelog.c.

{
    Walfile    *f;
    static char tmppath[MAXPGPATH];
 
    snprintf(tmppath, sizeof(tmppath), "archive_status/%s.done",
             fname);
 
    f = stream->walmethod->ops->open_for_write(stream->walmethod, tmppath,
                                               NULL, 0);
    if (f == NULL)
    {
        pg_log_error("could not create archive status file \"%s\": %s",
                     tmppath, GetLastWalMethodError(stream->walmethod));
        return false;
    }
 
    if (stream->walmethod->ops->close(f, CLOSE_NORMAL) != 0)
    {
        pg_log_error("could not close archive status file \"%s\": %s",
                     tmppath, GetLastWalMethodError(stream->walmethod));
        return false;
    }
 
    return true;
}

References WalWriteMethodOps::close, CLOSE_NORMAL, fb(), GetLastWalMethodError(), MAXPGPATH, WalWriteMethodOps::open_for_write, WalWriteMethod::ops, 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 90 of file receivelog.c.

{
    Walfile    *f;
    char       *fn;
    ssize_t     size;
    XLogSegNo   segno;
    char        walfile_name[MAXPGPATH];
 
    XLByteToSeg(startpoint, segno, WalSegSz);
    XLogFileName(walfile_name, stream->timeline, segno, WalSegSz);
 
    /* Note that this considers the compression used if necessary */
    fn = stream->walmethod->ops->get_file_name(stream->walmethod,
                                               walfile_name,
                                               stream->partial_suffix);
 
    /*
     * When streaming to files, if an existing file exists we verify that it's
     * either empty (just created), or a complete WalSegSz segment (in which
     * case it has been created and padded). Anything else indicates a corrupt
     * file. Compressed files have no need for padding, so just ignore this
     * case.
     *
     * When streaming to tar, no file with this name will exist before, so we
     * never have to verify a size.
     */
    if (stream->walmethod->compression_algorithm == PG_COMPRESSION_NONE &&
        stream->walmethod->ops->existsfile(stream->walmethod, fn))
    {
        size = stream->walmethod->ops->get_file_size(stream->walmethod, fn);
        if (size < 0)
        {
            pg_log_error("could not get size of write-ahead log file \"%s\": %s",
                         fn, GetLastWalMethodError(stream->walmethod));
            pg_free(fn);
            return false;
        }
        if (size == WalSegSz)
        {
            /* Already padded file. Open it for use */
            f = stream->walmethod->ops->open_for_write(stream->walmethod, walfile_name, stream->partial_suffix, 0);
            if (f == NULL)
            {
                pg_log_error("could not open existing write-ahead log file \"%s\": %s",
                             fn, GetLastWalMethodError(stream->walmethod));
                pg_free(fn);
                return false;
            }
 
            /* fsync file in case of a previous crash */
            if (stream->walmethod->ops->sync(f) != 0)
            {
                pg_log_error("could not fsync existing write-ahead log file \"%s\": %s",
                             fn, GetLastWalMethodError(stream->walmethod));
                stream->walmethod->ops->close(f, CLOSE_UNLINK);
                exit(1);
            }
 
            walfile = f;
            pg_free(fn);
            return true;
        }
        if (size != 0)
        {
            /* if write didn't set errno, assume problem is no disk space */
            if (errno == 0)
                errno = ENOSPC;
            pg_log_error(ngettext("write-ahead log file \"%s\" has %zd byte, should be 0 or %d",
                                  "write-ahead log file \"%s\" has %zd bytes, should be 0 or %d",
                                  size),
                         fn, size, WalSegSz);
            pg_free(fn);
            return false;
        }
        /* File existed and was empty, so fall through and open */
    }
 
    /* No file existed, so create one */
 
    f = stream->walmethod->ops->open_for_write(stream->walmethod,
                                               walfile_name,
                                               stream->partial_suffix,
                                               WalSegSz);
    if (f == NULL)
    {
        pg_log_error("could not open write-ahead log file \"%s\": %s",
                     fn, GetLastWalMethodError(stream->walmethod));
        pg_free(fn);
        return false;
    }
 
    pg_free(fn);
    walfile = f;
    return true;
}

References WalWriteMethodOps::close, CLOSE_UNLINK, WalWriteMethod::compression_algorithm, WalWriteMethodOps::existsfile, fb(), fn(), WalWriteMethodOps::get_file_name, WalWriteMethodOps::get_file_size, GetLastWalMethodError(), MAXPGPATH, ngettext, WalWriteMethodOps::open_for_write, WalWriteMethod::ops, StreamCtl::partial_suffix, PG_COMPRESSION_NONE, pg_free(), pg_log_error, WalWriteMethodOps::sync, StreamCtl::timeline, walfile, StreamCtl::walmethod, WalSegSz, XLByteToSeg, and XLogFileName().

Referenced by ProcessWALDataMsg().

ProcessKeepaliveMsg()

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

static

Definition at line 1000 of file receivelog.c.

{
    int         pos;
    bool        replyRequested;
    TimestampTz now;
 
    /*
     * 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 PqReplMsg_Keepalive */
    pos += 8;                   /* skip walEnd */
    pos += 8;                   /* skip sendTime */
 
    if (len < pos + 1)
    {
        pg_log_error("streaming header too small: %d", len);
        return false;
    }
    replyRequested = copybuf[pos];
 
    /* If the server requested an immediate reply, send one. */
    if (replyRequested && still_sending)
    {
        if (reportFlushPosition && lastFlushPosition < blockpos &&
            walfile != NULL)
        {
            /*
             * If a valid flush location needs to be reported, flush the
             * current WAL file so that the latest flush location is sent back
             * to the server. This is necessary to see whether the last WAL
             * data has been successfully replicated or not, at the normal
             * shutdown of the server.
             */
            if (stream->walmethod->ops->sync(walfile) != 0)
                pg_fatal("could not fsync file \"%s\": %s",
                         walfile->pathname, GetLastWalMethodError(stream->walmethod));
            lastFlushPosition = blockpos;
        }
 
        now = feGetCurrentTimestamp();
        if (!sendFeedback(conn, blockpos, now, false))
            return false;
        *last_status = now;
    }
 
    return true;
}

References conn, copybuf, fb(), feGetCurrentTimestamp(), GetLastWalMethodError(), lastFlushPosition, len, now(), WalWriteMethod::ops, Walfile::pathname, pg_fatal, pg_log_error, reportFlushPosition, sendFeedback(), still_sending, WalWriteMethodOps::sync, walfile, and StreamCtl::walmethod.

Referenced by HandleCopyStream().

ProcessWALDataMsg()

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

static

Definition at line 1054 of file receivelog.c.

{
    int         xlogoff;
    int         bytes_left;
    int         bytes_written;
    int         hdr_len;
 
    /*
     * Once we've decided we don't want to receive any more, just ignore any
     * subsequent WALData messages.
     */
    if (!(still_sending))
        return true;
 
    /*
     * Read the header of the WALData 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 PqReplMsg_WALData */
    hdr_len += 8;               /* dataStart */
    hdr_len += 8;               /* walEnd */
    hdr_len += 8;               /* sendTime */
    if (len < hdr_len)
    {
        pg_log_error("streaming header too small: %d", len);
        return false;
    }
    *blockpos = fe_recvint64(&copybuf[1]);
 
    /* Extract WAL location for this block */
    xlogoff = XLogSegmentOffset(*blockpos, WalSegSz);
 
    /*
     * Verify that the initial location in the stream matches where we think
     * we are.
     */
    if (walfile == NULL)
    {
        /* No file open yet */
        if (xlogoff != 0)
        {
            pg_log_error("received write-ahead log record for offset %u with no file open",
                         xlogoff);
            return false;
        }
    }
    else
    {
        /* More data in existing segment */
        if (walfile->currpos != xlogoff)
        {
            pg_log_error("got WAL data offset %08x, expected %08x",
                         xlogoff, (int) walfile->currpos);
            return false;
        }
    }
 
    bytes_left = len - hdr_len;
    bytes_written = 0;
 
    while (bytes_left)
    {
        int         bytes_to_write;
 
        /*
         * If crossing a WAL boundary, only write up until we reach wal
         * segment size.
         */
        if (xlogoff + bytes_left > WalSegSz)
            bytes_to_write = WalSegSz - xlogoff;
        else
            bytes_to_write = bytes_left;
 
        if (walfile == NULL)
        {
            if (!open_walfile(stream, *blockpos))
            {
                /* Error logged by open_walfile */
                return false;
            }
        }
 
        if (stream->walmethod->ops->write(walfile,
                                          copybuf + hdr_len + bytes_written,
                                          bytes_to_write) != bytes_to_write)
        {
            pg_log_error("could not write %d bytes to WAL file \"%s\": %s",
                         bytes_to_write, walfile->pathname,
                         GetLastWalMethodError(stream->walmethod));
            return false;
        }
 
        /* 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 (XLogSegmentOffset(*blockpos, WalSegSz) == 0)
        {
            if (!close_walfile(stream, *blockpos))
                /* Error message written in close_walfile() */
                return false;
 
            xlogoff = 0;
 
            if (still_sending && stream->stream_stop(*blockpos, stream->timeline, true))
            {
                if (PQputCopyEnd(conn, NULL) <= 0 || PQflush(conn))
                {
                    pg_log_error("could not send copy-end packet: %s",
                                 PQerrorMessage(conn));
                    return false;
                }
                still_sending = false;
                return true;    /* ignore the rest of this WALData packet */
            }
        }
    }
    /* No more data left to write, receive next copy packet */
 
    return true;
}

References close_walfile(), conn, copybuf, Walfile::currpos, fb(), fe_recvint64(), GetLastWalMethodError(), len, open_walfile(), WalWriteMethod::ops, Walfile::pathname, pg_log_error, PQerrorMessage(), PQflush(), PQputCopyEnd(), still_sending, StreamCtl::stream_stop, StreamCtl::timeline, walfile, StreamCtl::walmethod, WalSegSz, WalWriteMethodOps::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.

{
    uint32      startpos_xlogid,
                startpos_xrecoff;
 
    /*----------
     * The result set consists of one row and two columns, e.g:
     *
     *  next_tli | next_tli_startpos
     * ----------+-------------------
     *         4 | 0/9949AE0
     *
     * next_tli is the timeline ID of the next timeline after the one that
     * just finished streaming. next_tli_startpos is the WAL location where
     * the server switched to it.
     *----------
     */
    if (PQnfields(res) < 2 || PQntuples(res) != 1)
    {
        pg_log_error("unexpected result set after end-of-timeline: got %d rows and %d fields, expected %d rows and %d fields",
                     PQntuples(res), PQnfields(res), 1, 2);
        return false;
    }
 
    *timeline = atoi(PQgetvalue(res, 0, 0));
    if (sscanf(PQgetvalue(res, 0, 1), "%X/%08X", &startpos_xlogid,
               &startpos_xrecoff) != 2)
    {
        pg_log_error("could not parse next timeline's starting point \"%s\"",
                     PQgetvalue(res, 0, 1));
        return false;
    }
    *startpos = ((uint64) startpos_xlogid << 32) | startpos_xrecoff;
 
    return true;
}

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

Referenced by ReceiveXlogStream().

ReceiveXlogStream()

bool ReceiveXlogStream	(	PGconn *	conn,
		StreamCtl *	stream
	)

Definition at line 453 of file receivelog.c.

{
    PQExpBuffer query;
    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;
 
    /*
     * Decide whether we want to report the flush position. If we report the
     * flush position, the primary will know what WAL we'll possibly
     * re-request, and it can then remove older WAL safely. We must always do
     * that when we are using slots.
     *
     * Reporting the flush position makes one eligible as a synchronous
     * replica. People shouldn't include generic names in
     * synchronous_standby_names, but we've protected them against it so far,
     * so let's continue to do so unless specifically requested.
     */
    if (stream->replication_slot != NULL)
    {
        reportFlushPosition = true;
    }
    else
    {
        if (stream->synchronous)
            reportFlushPosition = true;
        else
            reportFlushPosition = false;
    }
 
    if (stream->sysidentifier != NULL)
    {
        char       *sysidentifier = NULL;
        TimeLineID  servertli;
 
        /*
         * Get the server system identifier and timeline, and validate them.
         */
        if (!RunIdentifySystem(conn, &sysidentifier, &servertli, NULL, NULL))
        {
            pg_free(sysidentifier);
            return false;
        }
 
        if (strcmp(stream->sysidentifier, sysidentifier) != 0)
        {
            pg_log_error("system identifier does not match between base backup and streaming connection");
            pg_free(sysidentifier);
            return false;
        }
        pg_free(sysidentifier);
 
        if (stream->timeline > servertli)
        {
            pg_log_error("starting timeline %u is not present in the server",
                         stream->timeline);
            return false;
        }
    }
 
    /*
     * initialize flush position to starting point, it's the caller's
     * responsibility that that's sane.
     */
    lastFlushPosition = stream->startpos;
 
    while (1)
    {
        /*
         * Fetch the timeline history file for this timeline, if we don't have
         * it already. When streaming log to tar, this will always return
         * false, as we are never streaming into an existing file and
         * therefore there can be no pre-existing timeline history file.
         */
        if (!existsTimeLineHistoryFile(stream))
        {
            query = createPQExpBuffer();
            appendPQExpBuffer(query, "TIMELINE_HISTORY %u", stream->timeline);
            res = PQexec(conn, query->data);
            destroyPQExpBuffer(query);
            if (PQresultStatus(res) != PGRES_TUPLES_OK)
            {
                /* FIXME: we might send it ok, but get an error */
                pg_log_error("could not send replication command \"%s\": %s",
                             "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)
            {
                pg_log_warning("unexpected response to TIMELINE_HISTORY command: got %d rows and %d fields, expected %d rows and %d fields",
                               PQntuples(res), PQnfields(res), 1, 2);
            }
 
            /* Write the history file to disk */
            writeTimeLineHistoryFile(stream,
                                     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->stream_stop(stream->startpos, stream->timeline, false))
            return true;
 
        /* Initiate the replication stream at specified location */
        query = createPQExpBuffer();
        appendPQExpBufferStr(query, "START_REPLICATION");
        if (stream->replication_slot != NULL)
        {
            appendPQExpBufferStr(query, " SLOT ");
            AppendQuotedIdentifier(query, stream->replication_slot);
        }
        appendPQExpBuffer(query, " %X/%08X TIMELINE %u",
                          LSN_FORMAT_ARGS(stream->startpos),
                          stream->timeline);
        res = PQexec(conn, query->data);
        destroyPQExpBuffer(query);
        if (PQresultStatus(res) != PGRES_COPY_BOTH)
        {
            pg_log_error("could not send replication command \"%s\": %s",
                         "START_REPLICATION", PQresultErrorMessage(res));
            PQclear(res);
            return false;
        }
        PQclear(res);
 
        /* Stream the WAL */
        res = HandleCopyStream(conn, stream, &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 information about the next timeline. Read
         * that, and restart streaming from the next timeline. In case of
         * controlled shutdown, stop here.
         */
        if (PQresultStatus(res) == PGRES_TUPLES_OK)
        {
            /*
             * End-of-timeline. Read the next timeline's ID and starting
             * position. Usually, the starting position will match the end of
             * the previous timeline, but there are corner cases like if the
             * server had sent us half of a WAL record, when it was promoted.
             * The new timeline will begin at the end of the last complete
             * record in that case, overlapping the partial WAL record on the
             * old timeline.
             */
            uint32      newtimeline;
            bool        parsed;
 
            parsed = ReadEndOfStreamingResult(res, &stream->startpos, &newtimeline);
            PQclear(res);
            if (!parsed)
                goto error;
 
            /* Sanity check the values the server gave us */
            if (newtimeline <= stream->timeline)
            {
                pg_log_error("server reported unexpected next timeline %u, following timeline %u",
                             newtimeline, stream->timeline);
                goto error;
            }
            if (stream->startpos > stoppos)
            {
                pg_log_error("server stopped streaming timeline %u at %X/%08X, but reported next timeline %u to begin at %X/%08X",
                             stream->timeline, LSN_FORMAT_ARGS(stoppos),
                             newtimeline, LSN_FORMAT_ARGS(stream->startpos));
                goto error;
            }
 
            /* Read the final result, which should be CommandComplete. */
            res = PQgetResult(conn);
            if (PQresultStatus(res) != PGRES_COMMAND_OK)
            {
                pg_log_error("unexpected termination of replication stream: %s",
                             PQresultErrorMessage(res));
                PQclear(res);
                goto error;
            }
            PQclear(res);
 
            /*
             * Loop back to start streaming from the new timeline. Always
             * start streaming at the beginning of a segment.
             */
            stream->timeline = newtimeline;
            stream->startpos = stream->startpos -
                XLogSegmentOffset(stream->startpos, WalSegSz);
            continue;
        }
        else if (PQresultStatus(res) == PGRES_COMMAND_OK)
        {
            PQclear(res);
 
            /*
             * 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->stream_stop(stoppos, stream->timeline, false))
                return true;
            else
            {
                pg_log_error("replication stream was terminated before stop point");
                goto error;
            }
        }
        else
        {
            /* Server returned an error. */
            pg_log_error("unexpected termination of replication stream: %s",
                         PQresultErrorMessage(res));
            PQclear(res);
            goto error;
        }
    }
 
error:
    if (walfile != NULL && stream->walmethod->ops->close(walfile, CLOSE_NO_RENAME) != 0)
        pg_log_error("could not close file \"%s\": %s",
                     walfile->pathname, GetLastWalMethodError(stream->walmethod));
    walfile = NULL;
    return false;
}

References appendPQExpBuffer(), appendPQExpBufferStr(), AppendQuotedIdentifier, CheckServerVersionForStreaming(), WalWriteMethodOps::close, CLOSE_NO_RENAME, conn, createPQExpBuffer(), PQExpBufferData::data, destroyPQExpBuffer(), error(), existsTimeLineHistoryFile(), fb(), GetLastWalMethodError(), HandleCopyStream(), lastFlushPosition, LSN_FORMAT_ARGS, WalWriteMethod::ops, Walfile::pathname, 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, RunIdentifySystem(), 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 337 of file receivelog.c.

{
    char        replybuf[1 + 8 + 8 + 8 + 8 + 1];
    int         len = 0;
 
    replybuf[len] = PqReplMsg_StandbyStatusUpdate;
    len += 1;
    fe_sendint64(blockpos, &replybuf[len]); /* write */
    len += 8;
    if (reportFlushPosition)
        fe_sendint64(lastFlushPosition, &replybuf[len]);    /* flush */
    else
        fe_sendint64(InvalidXLogRecPtr, &replybuf[len]);    /* flush */
    len += 8;
    fe_sendint64(InvalidXLogRecPtr, &replybuf[len]);    /* apply */
    len += 8;
    fe_sendint64(now, &replybuf[len]);  /* sendTime */
    len += 8;
    replybuf[len] = replyRequested ? 1 : 0; /* replyRequested */
    len += 1;
 
    if (PQputCopyData(conn, replybuf, len) <= 0 || PQflush(conn))
    {
        pg_log_error("could not send feedback packet: %s",
                     PQerrorMessage(conn));
        return false;
    }
 
    return true;
}

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

Referenced by HandleCopyStream(), and ProcessKeepaliveMsg().

writeTimeLineHistoryFile()

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

static

Definition at line 275 of file receivelog.c.

{
    int         size = strlen(content);
    char        histfname[MAXFNAMELEN];
    Walfile    *f;
 
    /*
     * Check that the server's idea of how timeline history files should be
     * named matches ours.
     */
    TLHistoryFileName(histfname, stream->timeline);
    if (strcmp(histfname, filename) != 0)
    {
        pg_log_error("server reported unexpected history file name for timeline %u: %s",
                     stream->timeline, filename);
        return false;
    }
 
    f = stream->walmethod->ops->open_for_write(stream->walmethod,
                                               histfname, ".tmp", 0);
    if (f == NULL)
    {
        pg_log_error("could not create timeline history file \"%s\": %s",
                     histfname, GetLastWalMethodError(stream->walmethod));
        return false;
    }
 
    if ((int) stream->walmethod->ops->write(f, content, size) != size)
    {
        pg_log_error("could not write timeline history file \"%s\": %s",
                     histfname, GetLastWalMethodError(stream->walmethod));
 
        /*
         * If we fail to make the file, delete it to release disk space
         */
        stream->walmethod->ops->close(f, CLOSE_UNLINK);
 
        return false;
    }
 
    if (stream->walmethod->ops->close(f, CLOSE_NORMAL) != 0)
    {
        pg_log_error("could not close file \"%s\": %s",
                     histfname, GetLastWalMethodError(stream->walmethod));
        return false;
    }
 
    /* Maintain archive_status, check close_walfile() for details. */
    if (stream->mark_done)
    {
        /* writes error message if failed */
        if (!mark_file_as_archived(stream, histfname))
            return false;
    }
 
    return true;
}

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

Referenced by ReceiveXlogStream().

Variable Documentation

lastFlushPosition

XLogRecPtr lastFlushPosition = InvalidXLogRecPtr

static

Definition at line 31 of file receivelog.c.

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

reportFlushPosition

bool reportFlushPosition = false

static

Definition at line 30 of file receivelog.c.

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

still_sending

bool still_sending = true

static

Definition at line 33 of file receivelog.c.

Referenced by CalculateCopyStreamSleeptime(), CheckCopyStreamStop(), HandleCopyStream(), HandleEndOfCopyStream(), ProcessKeepaliveMsg(), and ProcessWALDataMsg().

walfile

Walfile* walfile = NULL

static

Definition at line 29 of file receivelog.c.

Referenced by CleanupPriorWALFiles(), close_walfile(), HandleCopyStream(), open_walfile(), ProcessKeepaliveMsg(), ProcessWALDataMsg(), and ReceiveXlogStream().