receivelog.h File Reference

#include "access/xlogdefs.h"
#include "libpq-fe.h"
#include "walmethods.h"

Include dependency graph for receivelog.h:

This graph shows which files directly or indirectly include this file:

Go to the source code of this file.

Data Structures
struct	StreamCtl

Typedefs
typedef bool(*	stream_stop_callback) (XLogRecPtr segendpos, uint32 timeline, bool segment_finished)
typedef struct StreamCtl	StreamCtl

Functions
bool	CheckServerVersionForStreaming (PGconn *conn)
bool	ReceiveXlogStream (PGconn *conn, StreamCtl *stream)

Typedef Documentation

stream_stop_callback

typedef bool(* stream_stop_callback) (XLogRecPtr segendpos, uint32 timeline, bool segment_finished)

Definition at line 23 of file receivelog.h.

StreamCtl

typedef struct StreamCtl StreamCtl

Function Documentation

CheckServerVersionForStreaming()

bool CheckServerVersionForStreaming ( PGconn *  conn )

extern

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

ReceiveXlogStream()

bool ReceiveXlogStream ( PGconn *  conn, StreamCtl *  stream  )

extern

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