basebackup_copy.c File Reference

#include "postgres.h"
#include "catalog/pg_type_d.h"
#include "libpq/libpq.h"
#include "libpq/pqformat.h"
#include "replication/basebackup.h"
#include "replication/basebackup_sink.h"
#include "utils/timestamp.h"

Include dependency graph for basebackup_copy.c:

Go to the source code of this file.

Data Structures
struct	bbsink_copystream

Macros
#define	PROGRESS_REPORT_BYTE_INTERVAL   65536
#define	PROGRESS_REPORT_MILLISECOND_THRESHOLD   1000

Typedefs
typedef struct bbsink_copystream	bbsink_copystream

Functions
static void	bbsink_copystream_begin_backup (bbsink *sink)
static void	bbsink_copystream_begin_archive (bbsink *sink, const char *archive_name)
static void	bbsink_copystream_archive_contents (bbsink *sink, size_t len)
static void	bbsink_copystream_end_archive (bbsink *sink)
static void	bbsink_copystream_begin_manifest (bbsink *sink)
static void	bbsink_copystream_manifest_contents (bbsink *sink, size_t len)
static void	bbsink_copystream_end_manifest (bbsink *sink)
static void	bbsink_copystream_end_backup (bbsink *sink, XLogRecPtr endptr, TimeLineID endtli)
static void	bbsink_copystream_cleanup (bbsink *sink)
static void	SendCopyOutResponse (void)
static void	SendCopyDone (void)
static void	SendXlogRecPtrResult (XLogRecPtr ptr, TimeLineID tli)
static void	SendTablespaceList (List *tablespaces)
static void	send_int8_string (StringInfoData *buf, int64 intval)
bbsink *	bbsink_copystream_new (bool send_to_client)

Variables
static const bbsink_ops	bbsink_copystream_ops

Macro Definition Documentation

PROGRESS_REPORT_BYTE_INTERVAL

#define PROGRESS_REPORT_BYTE_INTERVAL   65536

Definition at line 68 of file basebackup_copy.c.

PROGRESS_REPORT_MILLISECOND_THRESHOLD

#define PROGRESS_REPORT_MILLISECOND_THRESHOLD   1000

Definition at line 69 of file basebackup_copy.c.

Typedef Documentation

bbsink_copystream

typedef struct bbsink_copystream bbsink_copystream

Function Documentation

bbsink_copystream_archive_contents()

static void bbsink_copystream_archive_contents ( bbsink *  sink, size_t  len  )

static

Definition at line 180 of file basebackup_copy.c.

{
    bbsink_copystream *mysink = (bbsink_copystream *) sink;
    bbsink_state *state = mysink->base.bbs_state;
    StringInfoData buf;
    uint64      targetbytes;
 
    /* Send the archive content to the client, if appropriate. */
    if (mysink->send_to_client)
    {
        /* Add one because we're also sending a leading type byte. */
        pq_putmessage('d', mysink->msgbuffer, len + 1);
    }
 
    /* Consider whether to send a progress report to the client. */
    targetbytes = mysink->bytes_done_at_last_time_check
        + PROGRESS_REPORT_BYTE_INTERVAL;
    if (targetbytes <= state->bytes_done)
    {
        TimestampTz now = GetCurrentTimestamp();
        long        ms;
 
        /*
         * OK, we've sent a decent number of bytes, so check the system time
         * to see whether we're due to send a progress report.
         */
        mysink->bytes_done_at_last_time_check = state->bytes_done;
        ms = TimestampDifferenceMilliseconds(mysink->last_progress_report_time,
                                             now);
 
        /*
         * Send a progress report if enough time has passed. Also send one if
         * the system clock was set backward, so that such occurrences don't
         * have the effect of suppressing further progress messages.
         */
        if (ms < 0 || ms >= PROGRESS_REPORT_MILLISECOND_THRESHOLD)
        {
            mysink->last_progress_report_time = now;
 
            pq_beginmessage(&buf, 'd'); /* CopyData */
            pq_sendbyte(&buf, 'p'); /* Progress report */
            pq_sendint64(&buf, state->bytes_done);
            pq_endmessage(&buf);
            pq_flush_if_writable();
        }
    }
}

References bbsink_copystream::base, bbsink::bbs_state, buf, bbsink_copystream::bytes_done_at_last_time_check, GetCurrentTimestamp(), bbsink_copystream::last_progress_report_time, len, bbsink_copystream::msgbuffer, now(), pq_beginmessage(), pq_endmessage(), pq_flush_if_writable, pq_putmessage, pq_sendbyte(), pq_sendint64(), PROGRESS_REPORT_BYTE_INTERVAL, PROGRESS_REPORT_MILLISECOND_THRESHOLD, bbsink_copystream::send_to_client, and TimestampDifferenceMilliseconds().

bbsink_copystream_begin_archive()

static void bbsink_copystream_begin_archive ( bbsink *  sink, const char *  archive_name  )

static

Definition at line 162 of file basebackup_copy.c.

{
    bbsink_state *state = sink->bbs_state;
    tablespaceinfo *ti;
    StringInfoData buf;
 
    ti = list_nth(state->tablespaces, state->tablespace_num);
    pq_beginmessage(&buf, 'd'); /* CopyData */
    pq_sendbyte(&buf, 'n');     /* New archive */
    pq_sendstring(&buf, archive_name);
    pq_sendstring(&buf, ti->path == NULL ? "" : ti->path);
    pq_endmessage(&buf);
}

References bbsink::bbs_state, buf, list_nth(), tablespaceinfo::path, pq_beginmessage(), pq_endmessage(), pq_sendbyte(), and pq_sendstring().

bbsink_copystream_begin_backup()

static void bbsink_copystream_begin_backup ( bbsink *  sink )

static

Definition at line 123 of file basebackup_copy.c.

{
    bbsink_copystream *mysink = (bbsink_copystream *) sink;
    bbsink_state *state = sink->bbs_state;
    char       *buf;
 
    /*
     * Initialize buffer. We ultimately want to send the archive and manifest
     * data by means of CopyData messages where the payload portion of each
     * message begins with a type byte. However, basebackup.c expects the
     * buffer to be aligned, so we can't just allocate one extra byte for the
     * type byte. Instead, allocate enough extra bytes that the portion of the
     * buffer we reveal to our callers can be aligned, while leaving room to
     * slip the type byte in just beforehand.  That will allow us to ship the
     * data with a single call to pq_putmessage and without needing any extra
     * copying.
     */
    buf = palloc(mysink->base.bbs_buffer_length + MAXIMUM_ALIGNOF);
    mysink->msgbuffer = buf + (MAXIMUM_ALIGNOF - 1);
    mysink->base.bbs_buffer = buf + MAXIMUM_ALIGNOF;
    mysink->msgbuffer[0] = 'd'; /* archive or manifest data */
 
    /* Tell client the backup start location. */
    SendXlogRecPtrResult(state->startptr, state->starttli);
 
    /* Send client a list of tablespaces. */
    SendTablespaceList(state->tablespaces);
 
    /* Send a CommandComplete message */
    pq_puttextmessage('C', "SELECT");
 
    /* Begin COPY stream. This will be used for all archives + manifest. */
    SendCopyOutResponse();
}

References bbsink_copystream::base, bbsink::bbs_buffer, bbsink::bbs_buffer_length, bbsink::bbs_state, buf, bbsink_copystream::msgbuffer, palloc(), pq_puttextmessage(), SendCopyOutResponse(), SendTablespaceList(), and SendXlogRecPtrResult().

bbsink_copystream_begin_manifest()

static void bbsink_copystream_begin_manifest ( bbsink *  sink )

static

Definition at line 256 of file basebackup_copy.c.

{
    StringInfoData buf;
 
    pq_beginmessage(&buf, 'd'); /* CopyData */
    pq_sendbyte(&buf, 'm');     /* Manifest */
    pq_endmessage(&buf);
}

References buf, pq_beginmessage(), pq_endmessage(), and pq_sendbyte().

bbsink_copystream_cleanup()

static void bbsink_copystream_cleanup ( bbsink *  sink )

static

Definition at line 304 of file basebackup_copy.c.

{
    /* Nothing to do. */
}

bbsink_copystream_end_archive()

static void bbsink_copystream_end_archive ( bbsink *  sink )

static

Definition at line 237 of file basebackup_copy.c.

{
    bbsink_copystream *mysink = (bbsink_copystream *) sink;
    bbsink_state *state = mysink->base.bbs_state;
    StringInfoData buf;
 
    mysink->bytes_done_at_last_time_check = state->bytes_done;
    mysink->last_progress_report_time = GetCurrentTimestamp();
    pq_beginmessage(&buf, 'd'); /* CopyData */
    pq_sendbyte(&buf, 'p');     /* Progress report */
    pq_sendint64(&buf, state->bytes_done);
    pq_endmessage(&buf);
    pq_flush_if_writable();
}

References bbsink_copystream::base, bbsink::bbs_state, buf, bbsink_copystream::bytes_done_at_last_time_check, GetCurrentTimestamp(), bbsink_copystream::last_progress_report_time, pq_beginmessage(), pq_endmessage(), pq_flush_if_writable, pq_sendbyte(), and pq_sendint64().

bbsink_copystream_end_backup()

static void bbsink_copystream_end_backup ( bbsink *  sink, XLogRecPtr  endptr, TimeLineID  endtli  )

static

Definition at line 293 of file basebackup_copy.c.

{
    SendCopyDone();
    SendXlogRecPtrResult(endptr, endtli);
}

References SendCopyDone(), and SendXlogRecPtrResult().

bbsink_copystream_end_manifest()

static void bbsink_copystream_end_manifest ( bbsink *  sink )

static

Definition at line 284 of file basebackup_copy.c.

{
    /* Do nothing. */
}

bbsink_copystream_manifest_contents()

static void bbsink_copystream_manifest_contents ( bbsink *  sink, size_t  len  )

static

Definition at line 269 of file basebackup_copy.c.

{
    bbsink_copystream *mysink = (bbsink_copystream *) sink;
 
    if (mysink->send_to_client)
    {
        /* Add one because we're also sending a leading type byte. */
        pq_putmessage('d', mysink->msgbuffer, len + 1);
    }
}

References len, bbsink_copystream::msgbuffer, pq_putmessage, and bbsink_copystream::send_to_client.

bbsink_copystream_new()

bbsink* bbsink_copystream_new

(

bool

send_to_client

)

Definition at line 105 of file basebackup_copy.c.

{
    bbsink_copystream *sink = palloc0(sizeof(bbsink_copystream));
 
    *((const bbsink_ops **) &sink->base.bbs_ops) = &bbsink_copystream_ops;
    sink->send_to_client = send_to_client;
 
    /* Set up for periodic progress reporting. */
    sink->last_progress_report_time = GetCurrentTimestamp();
    sink->bytes_done_at_last_time_check = UINT64CONST(0);
 
    return &sink->base;
}

References bbsink_copystream::base, bbsink::bbs_ops, bbsink_copystream_ops, bbsink_copystream::bytes_done_at_last_time_check, GetCurrentTimestamp(), bbsink_copystream::last_progress_report_time, palloc0(), and bbsink_copystream::send_to_client.

Referenced by SendBaseBackup().

send_int8_string()

static void send_int8_string ( StringInfoData *  buf, int64  intval  )

static

Definition at line 466 of file basebackup_copy.c.

{
    char        is[32];
 
    sprintf(is, INT64_FORMAT, intval);
    pq_sendint32(buf, strlen(is));
    pq_sendbytes(buf, is, strlen(is));
}

References buf, INT64_FORMAT, pq_sendbytes(), pq_sendint32(), and sprintf.

Referenced by SendTablespaceList().

SendCopyDone()

static void SendCopyDone ( void  )

static

Definition at line 327 of file basebackup_copy.c.

{
    pq_putemptymessage('c');
}

References pq_putemptymessage().

Referenced by bbsink_copystream_end_backup().

SendCopyOutResponse()

static void SendCopyOutResponse ( void  )

static

Definition at line 313 of file basebackup_copy.c.

{
    StringInfoData buf;
 
    pq_beginmessage(&buf, 'H');
    pq_sendbyte(&buf, 0);       /* overall format */
    pq_sendint16(&buf, 0);      /* natts */
    pq_endmessage(&buf);
}

References buf, pq_beginmessage(), pq_endmessage(), pq_sendbyte(), and pq_sendint16().

Referenced by bbsink_copystream_begin_backup().

SendTablespaceList()

static void SendTablespaceList ( List *  tablespaces )

static

Definition at line 392 of file basebackup_copy.c.

{
    StringInfoData buf;
    ListCell   *lc;
 
    /* Construct and send the directory information */
    pq_beginmessage(&buf, 'T'); /* RowDescription */
    pq_sendint16(&buf, 3);      /* 3 fields */
 
    /* First field - spcoid */
    pq_sendstring(&buf, "spcoid");
    pq_sendint32(&buf, 0);      /* table oid */
    pq_sendint16(&buf, 0);      /* attnum */
    pq_sendint32(&buf, OIDOID); /* type oid */
    pq_sendint16(&buf, 4);      /* typlen */
    pq_sendint32(&buf, 0);      /* typmod */
    pq_sendint16(&buf, 0);      /* format code */
 
    /* Second field - spclocation */
    pq_sendstring(&buf, "spclocation");
    pq_sendint32(&buf, 0);
    pq_sendint16(&buf, 0);
    pq_sendint32(&buf, TEXTOID);
    pq_sendint16(&buf, -1);
    pq_sendint32(&buf, 0);
    pq_sendint16(&buf, 0);
 
    /* Third field - size */
    pq_sendstring(&buf, "size");
    pq_sendint32(&buf, 0);
    pq_sendint16(&buf, 0);
    pq_sendint32(&buf, INT8OID);
    pq_sendint16(&buf, 8);
    pq_sendint32(&buf, 0);
    pq_sendint16(&buf, 0);
    pq_endmessage(&buf);
 
    foreach(lc, tablespaces)
    {
        tablespaceinfo *ti = lfirst(lc);
 
        /* Send one datarow message */
        pq_beginmessage(&buf, 'D');
        pq_sendint16(&buf, 3);  /* number of columns */
        if (ti->path == NULL)
        {
            pq_sendint32(&buf, -1); /* Length = -1 ==> NULL */
            pq_sendint32(&buf, -1);
        }
        else
        {
            Size        len;
 
            len = strlen(ti->oid);
            pq_sendint32(&buf, len);
            pq_sendbytes(&buf, ti->oid, len);
 
            len = strlen(ti->path);
            pq_sendint32(&buf, len);
            pq_sendbytes(&buf, ti->path, len);
        }
        if (ti->size >= 0)
            send_int8_string(&buf, ti->size / 1024);
        else
            pq_sendint32(&buf, -1); /* NULL */
 
        pq_endmessage(&buf);
    }
}

References buf, len, lfirst, tablespaceinfo::oid, tablespaceinfo::path, pq_beginmessage(), pq_endmessage(), pq_sendbytes(), pq_sendint16(), pq_sendint32(), pq_sendstring(), send_int8_string(), and tablespaceinfo::size.

Referenced by bbsink_copystream_begin_backup().

SendXlogRecPtrResult()

static void SendXlogRecPtrResult ( XLogRecPtr  ptr, TimeLineID  tli  )

static

Definition at line 337 of file basebackup_copy.c.

{
    StringInfoData buf;
    char        str[MAXFNAMELEN];
    Size        len;
 
    pq_beginmessage(&buf, 'T'); /* RowDescription */
    pq_sendint16(&buf, 2);      /* 2 fields */
 
    /* Field headers */
    pq_sendstring(&buf, "recptr");
    pq_sendint32(&buf, 0);      /* table oid */
    pq_sendint16(&buf, 0);      /* attnum */
    pq_sendint32(&buf, TEXTOID);    /* type oid */
    pq_sendint16(&buf, -1);
    pq_sendint32(&buf, 0);
    pq_sendint16(&buf, 0);
 
    pq_sendstring(&buf, "tli");
    pq_sendint32(&buf, 0);      /* table oid */
    pq_sendint16(&buf, 0);      /* attnum */
 
    /*
     * int8 may seem like a surprising data type for this, but in theory int4
     * would not be wide enough for this, as TimeLineID is unsigned.
     */
    pq_sendint32(&buf, INT8OID);    /* type oid */
    pq_sendint16(&buf, -1);
    pq_sendint32(&buf, 0);
    pq_sendint16(&buf, 0);
    pq_endmessage(&buf);
 
    /* Data row */
    pq_beginmessage(&buf, 'D');
    pq_sendint16(&buf, 2);      /* number of columns */
 
    len = snprintf(str, sizeof(str),
                   "%X/%X", LSN_FORMAT_ARGS(ptr));
    pq_sendint32(&buf, len);
    pq_sendbytes(&buf, str, len);
 
    len = snprintf(str, sizeof(str), "%u", tli);
    pq_sendint32(&buf, len);
    pq_sendbytes(&buf, str, len);
 
    pq_endmessage(&buf);
 
    /* Send a CommandComplete message */
    pq_puttextmessage('C', "SELECT");
}

References buf, len, LSN_FORMAT_ARGS, MAXFNAMELEN, pq_beginmessage(), pq_endmessage(), pq_puttextmessage(), pq_sendbytes(), pq_sendint16(), pq_sendint32(), pq_sendstring(), snprintf, and generate_unaccent_rules::str.

Referenced by bbsink_copystream_begin_backup(), and bbsink_copystream_end_backup().

Variable Documentation

bbsink_copystream_ops

const bbsink_ops bbsink_copystream_ops

static

Initial value:

= {
    .begin_backup = bbsink_copystream_begin_backup,
    .begin_archive = bbsink_copystream_begin_archive,
    .archive_contents = bbsink_copystream_archive_contents,
    .end_archive = bbsink_copystream_end_archive,
    .begin_manifest = bbsink_copystream_begin_manifest,
    .manifest_contents = bbsink_copystream_manifest_contents,
    .end_manifest = bbsink_copystream_end_manifest,
    .end_backup = bbsink_copystream_end_backup,
    .cleanup = bbsink_copystream_cleanup
}

Definition at line 89 of file basebackup_copy.c.

Referenced by bbsink_copystream_new().