astreamer_tar.c File Reference

#include "postgres_fe.h"
#include <time.h>
#include "common/logging.h"
#include "fe_utils/astreamer.h"
#include "pgtar.h"

Include dependency graph for astreamer_tar.c:

Go to the source code of this file.

Data Structures
struct	astreamer_tar_parser
struct	astreamer_tar_archiver

Typedefs
typedef struct astreamer_tar_parser	astreamer_tar_parser
typedef struct astreamer_tar_archiver	astreamer_tar_archiver

Functions
static void	astreamer_tar_parser_content (astreamer *streamer, astreamer_member *member, const char *data, int len, astreamer_archive_context context)
static void	astreamer_tar_parser_finalize (astreamer *streamer)
static void	astreamer_tar_parser_free (astreamer *streamer)
static bool	astreamer_tar_header (astreamer_tar_parser *mystreamer)
static void	astreamer_tar_archiver_content (astreamer *streamer, astreamer_member *member, const char *data, int len, astreamer_archive_context context)
static void	astreamer_tar_archiver_finalize (astreamer *streamer)
static void	astreamer_tar_archiver_free (astreamer *streamer)
static void	astreamer_tar_terminator_content (astreamer *streamer, astreamer_member *member, const char *data, int len, astreamer_archive_context context)
static void	astreamer_tar_terminator_finalize (astreamer *streamer)
static void	astreamer_tar_terminator_free (astreamer *streamer)
astreamer *	astreamer_tar_parser_new (astreamer *next)
astreamer *	astreamer_tar_archiver_new (astreamer *next)
astreamer *	astreamer_tar_terminator_new (astreamer *next)

Variables
static const astreamer_ops	astreamer_tar_parser_ops
static const astreamer_ops	astreamer_tar_archiver_ops
static const astreamer_ops	astreamer_tar_terminator_ops

Typedef Documentation

astreamer_tar_archiver

typedef struct astreamer_tar_archiver astreamer_tar_archiver

astreamer_tar_parser

typedef struct astreamer_tar_parser astreamer_tar_parser

Function Documentation

astreamer_tar_archiver_content()

static void astreamer_tar_archiver_content ( astreamer *  streamer, astreamer_member *  member, const char *  data, int  len, astreamer_archive_context  context  )

static

Definition at line 390 of file astreamer_tar.c.

{
    astreamer_tar_archiver *mystreamer = (astreamer_tar_archiver *) streamer;
    char        buffer[2 * TAR_BLOCK_SIZE];
 
    Assert(context != ASTREAMER_UNKNOWN);
 
    if (context == ASTREAMER_MEMBER_HEADER && len != TAR_BLOCK_SIZE)
    {
        Assert(len == 0);
 
        /* Replace zero-length tar header with a newly constructed one. */
        tarCreateHeader(buffer, member->pathname, NULL,
                        member->size, member->mode, member->uid, member->gid,
                        time(NULL));
        data = buffer;
        len = TAR_BLOCK_SIZE;
 
        /* Also make a note to replace padding, in case size changed. */
        mystreamer->rearchive_member = true;
    }
    else if (context == ASTREAMER_MEMBER_TRAILER &&
             mystreamer->rearchive_member)
    {
        int         pad_bytes = tarPaddingBytesRequired(member->size);
 
        /* Also replace padding, if we regenerated the header. */
        memset(buffer, 0, pad_bytes);
        data = buffer;
        len = pad_bytes;
 
        /* Don't do this again unless we replace another header. */
        mystreamer->rearchive_member = false;
    }
    else if (context == ASTREAMER_ARCHIVE_TRAILER)
    {
        /* Trailer should always be two blocks of zero bytes. */
        memset(buffer, 0, 2 * TAR_BLOCK_SIZE);
        data = buffer;
        len = 2 * TAR_BLOCK_SIZE;
    }
 
    astreamer_content(streamer->bbs_next, member, data, len, context);
}

References Assert, ASTREAMER_ARCHIVE_TRAILER, astreamer_content(), ASTREAMER_MEMBER_HEADER, ASTREAMER_MEMBER_TRAILER, ASTREAMER_UNKNOWN, astreamer::bbs_next, data, astreamer_member::gid, len, astreamer_member::mode, astreamer_member::pathname, astreamer_tar_archiver::rearchive_member, astreamer_member::size, TAR_BLOCK_SIZE, tarCreateHeader(), tarPaddingBytesRequired(), and astreamer_member::uid.

astreamer_tar_archiver_finalize()

static void astreamer_tar_archiver_finalize ( astreamer *  streamer )

static

Definition at line 442 of file astreamer_tar.c.

{
    astreamer_finalize(streamer->bbs_next);
}

References astreamer_finalize(), and astreamer::bbs_next.

astreamer_tar_archiver_free()

static void astreamer_tar_archiver_free ( astreamer *  streamer )

static

Definition at line 451 of file astreamer_tar.c.

{
    astreamer_free(streamer->bbs_next);
    pfree(streamer);
}

References astreamer_free(), astreamer::bbs_next, and pfree().

astreamer_tar_archiver_new()

astreamer * astreamer_tar_archiver_new

(

astreamer *

next

)

Definition at line 356 of file astreamer_tar.c.

{
    astreamer_tar_archiver *streamer;
 
    streamer = palloc0(sizeof(astreamer_tar_archiver));
    *((const astreamer_ops **) &streamer->base.bbs_ops) =
        &astreamer_tar_archiver_ops;
    streamer->base.bbs_next = next;
 
    return &streamer->base;
}

References astreamer_tar_archiver_ops, astreamer_tar_archiver::base, astreamer::bbs_next, astreamer::bbs_ops, next, and palloc0().

Referenced by CreateBackupStreamer().

astreamer_tar_header()

static bool astreamer_tar_header ( astreamer_tar_parser *  mystreamer )

static

Definition at line 261 of file astreamer_tar.c.

{
    bool        has_nonzero_byte = false;
    int         i;
    astreamer_member *member = &mystreamer->member;
    char       *buffer = mystreamer->base.bbs_buffer.data;
 
    Assert(mystreamer->base.bbs_buffer.len == TAR_BLOCK_SIZE);
 
    /* Check whether we've got a block of all zero bytes. */
    for (i = 0; i < TAR_BLOCK_SIZE; ++i)
    {
        if (buffer[i] != '\0')
        {
            has_nonzero_byte = true;
            break;
        }
    }
 
    /*
     * If the entire block was zeros, this is the end of the archive, not the
     * start of the next file.
     */
    if (!has_nonzero_byte)
        return false;
 
    /*
     * Parse key fields out of the header.
     */
    strlcpy(member->pathname, &buffer[TAR_OFFSET_NAME], MAXPGPATH);
    if (member->pathname[0] == '\0')
        pg_fatal("tar member has empty name");
    member->size = read_tar_number(&buffer[TAR_OFFSET_SIZE], 12);
    member->mode = read_tar_number(&buffer[TAR_OFFSET_MODE], 8);
    member->uid = read_tar_number(&buffer[TAR_OFFSET_UID], 8);
    member->gid = read_tar_number(&buffer[TAR_OFFSET_GID], 8);
    member->is_directory =
        (buffer[TAR_OFFSET_TYPEFLAG] == TAR_FILETYPE_DIRECTORY);
    member->is_link =
        (buffer[TAR_OFFSET_TYPEFLAG] == TAR_FILETYPE_SYMLINK);
    if (member->is_link)
        strlcpy(member->linktarget, &buffer[TAR_OFFSET_LINKNAME], 100);
 
    /* Compute number of padding bytes. */
    mystreamer->pad_bytes_expected = tarPaddingBytesRequired(member->size);
 
    /* Forward the entire header to the next astreamer. */
    astreamer_content(mystreamer->base.bbs_next, member,
                      buffer, TAR_BLOCK_SIZE,
                      ASTREAMER_MEMBER_HEADER);
 
    return true;
}

References Assert, astreamer_content(), ASTREAMER_MEMBER_HEADER, astreamer_tar_parser::base, astreamer::bbs_buffer, astreamer::bbs_next, StringInfoData::data, astreamer_member::gid, i, astreamer_member::is_directory, astreamer_member::is_link, StringInfoData::len, astreamer_member::linktarget, MAXPGPATH, astreamer_tar_parser::member, astreamer_member::mode, astreamer_tar_parser::pad_bytes_expected, astreamer_member::pathname, pg_fatal, read_tar_number(), astreamer_member::size, strlcpy(), TAR_BLOCK_SIZE, TAR_FILETYPE_DIRECTORY, TAR_FILETYPE_SYMLINK, TAR_OFFSET_GID, TAR_OFFSET_LINKNAME, TAR_OFFSET_MODE, TAR_OFFSET_NAME, TAR_OFFSET_SIZE, TAR_OFFSET_TYPEFLAG, TAR_OFFSET_UID, tarPaddingBytesRequired(), and astreamer_member::uid.

Referenced by astreamer_tar_parser_content().

astreamer_tar_parser_content()

static void astreamer_tar_parser_content ( astreamer *  streamer, astreamer_member *  member, const char *  data, int  len, astreamer_archive_context  context  )

static

Definition at line 111 of file astreamer_tar.c.

{
    astreamer_tar_parser *mystreamer = (astreamer_tar_parser *) streamer;
    size_t      nbytes;
 
    /* Expect unparsed input. */
    Assert(member == NULL);
    Assert(context == ASTREAMER_UNKNOWN);
 
    while (len > 0)
    {
        switch (mystreamer->next_context)
        {
            case ASTREAMER_MEMBER_HEADER:
 
                /*
                 * If we're expecting an archive member header, accumulate a
                 * full block of data before doing anything further.
                 */
                if (!astreamer_buffer_until(streamer, &data, &len,
                                            TAR_BLOCK_SIZE))
                    return;
 
                /*
                 * Now we can process the header and get ready to process the
                 * file contents; however, we might find out that what we
                 * thought was the next file header is actually the start of
                 * the archive trailer. Switch modes accordingly.
                 */
                if (astreamer_tar_header(mystreamer))
                {
                    if (mystreamer->member.size == 0)
                    {
                        /* No content; trailer is zero-length. */
                        astreamer_content(mystreamer->base.bbs_next,
                                          &mystreamer->member,
                                          NULL, 0,
                                          ASTREAMER_MEMBER_TRAILER);
 
                        /* Expect next header. */
                        mystreamer->next_context = ASTREAMER_MEMBER_HEADER;
                    }
                    else
                    {
                        /* Expect contents. */
                        mystreamer->next_context = ASTREAMER_MEMBER_CONTENTS;
                    }
                    mystreamer->base.bbs_buffer.len = 0;
                    mystreamer->file_bytes_sent = 0;
                }
                else
                    mystreamer->next_context = ASTREAMER_ARCHIVE_TRAILER;
                break;
 
            case ASTREAMER_MEMBER_CONTENTS:
 
                /*
                 * Send as much content as we have, but not more than the
                 * remaining file length.
                 */
                Assert(mystreamer->file_bytes_sent < mystreamer->member.size);
                nbytes = mystreamer->member.size - mystreamer->file_bytes_sent;
                nbytes = Min(nbytes, len);
                Assert(nbytes > 0);
                astreamer_content(mystreamer->base.bbs_next,
                                  &mystreamer->member,
                                  data, nbytes,
                                  ASTREAMER_MEMBER_CONTENTS);
                mystreamer->file_bytes_sent += nbytes;
                data += nbytes;
                len -= nbytes;
 
                /*
                 * If we've not yet sent the whole file, then there's more
                 * content to come; otherwise, it's time to expect the file
                 * trailer.
                 */
                Assert(mystreamer->file_bytes_sent <= mystreamer->member.size);
                if (mystreamer->file_bytes_sent == mystreamer->member.size)
                {
                    if (mystreamer->pad_bytes_expected == 0)
                    {
                        /* Trailer is zero-length. */
                        astreamer_content(mystreamer->base.bbs_next,
                                          &mystreamer->member,
                                          NULL, 0,
                                          ASTREAMER_MEMBER_TRAILER);
 
                        /* Expect next header. */
                        mystreamer->next_context = ASTREAMER_MEMBER_HEADER;
                    }
                    else
                    {
                        /* Trailer is not zero-length. */
                        mystreamer->next_context = ASTREAMER_MEMBER_TRAILER;
                    }
                    mystreamer->base.bbs_buffer.len = 0;
                }
                break;
 
            case ASTREAMER_MEMBER_TRAILER:
 
                /*
                 * If we're expecting an archive member trailer, accumulate
                 * the expected number of padding bytes before sending
                 * anything onward.
                 */
                if (!astreamer_buffer_until(streamer, &data, &len,
                                            mystreamer->pad_bytes_expected))
                    return;
 
                /* OK, now we can send it. */
                astreamer_content(mystreamer->base.bbs_next,
                                  &mystreamer->member,
                                  data, mystreamer->pad_bytes_expected,
                                  ASTREAMER_MEMBER_TRAILER);
 
                /* Expect next file header. */
                mystreamer->next_context = ASTREAMER_MEMBER_HEADER;
                mystreamer->base.bbs_buffer.len = 0;
                break;
 
            case ASTREAMER_ARCHIVE_TRAILER:
 
                /*
                 * We've seen an end-of-archive indicator, so anything more is
                 * buffered and sent as part of the archive trailer. But we
                 * don't expect more than 2 blocks.
                 */
                astreamer_buffer_bytes(streamer, &data, &len, len);
                if (len > 2 * TAR_BLOCK_SIZE)
                    pg_fatal("tar file trailer exceeds 2 blocks");
                return;
 
            default:
                /* Shouldn't happen. */
                pg_fatal("unexpected state while parsing tar archive");
        }
    }
}

References Assert, ASTREAMER_ARCHIVE_TRAILER, astreamer_buffer_bytes(), astreamer_buffer_until(), astreamer_content(), ASTREAMER_MEMBER_CONTENTS, ASTREAMER_MEMBER_HEADER, ASTREAMER_MEMBER_TRAILER, astreamer_tar_header(), ASTREAMER_UNKNOWN, astreamer_tar_parser::base, astreamer::bbs_buffer, astreamer::bbs_next, data, astreamer_tar_parser::file_bytes_sent, StringInfoData::len, len, astreamer_tar_parser::member, Min, astreamer_tar_parser::next_context, astreamer_tar_parser::pad_bytes_expected, pg_fatal, astreamer_member::size, and TAR_BLOCK_SIZE.

astreamer_tar_parser_finalize()

static void astreamer_tar_parser_finalize ( astreamer *  streamer )

static

Definition at line 319 of file astreamer_tar.c.

{
    astreamer_tar_parser *mystreamer = (astreamer_tar_parser *) streamer;
 
    if (mystreamer->next_context != ASTREAMER_ARCHIVE_TRAILER &&
        (mystreamer->next_context != ASTREAMER_MEMBER_HEADER ||
         mystreamer->base.bbs_buffer.len > 0))
        pg_fatal("COPY stream ended before last file was finished");
 
    /* Send the archive trailer, even if empty. */
    astreamer_content(streamer->bbs_next, NULL,
                      streamer->bbs_buffer.data, streamer->bbs_buffer.len,
                      ASTREAMER_ARCHIVE_TRAILER);
 
    /* Now finalize successor. */
    astreamer_finalize(streamer->bbs_next);
}

References ASTREAMER_ARCHIVE_TRAILER, astreamer_content(), astreamer_finalize(), ASTREAMER_MEMBER_HEADER, astreamer_tar_parser::base, astreamer::bbs_buffer, astreamer::bbs_next, StringInfoData::data, StringInfoData::len, astreamer_tar_parser::next_context, and pg_fatal.

astreamer_tar_parser_free()

static void astreamer_tar_parser_free ( astreamer *  streamer )

static

Definition at line 341 of file astreamer_tar.c.

{
    pfree(streamer->bbs_buffer.data);
    astreamer_free(streamer->bbs_next);
}

References astreamer_free(), astreamer::bbs_buffer, astreamer::bbs_next, StringInfoData::data, and pfree().

astreamer_tar_parser_new()

astreamer * astreamer_tar_parser_new

(

astreamer *

next

)

Definition at line 93 of file astreamer_tar.c.

{
    astreamer_tar_parser *streamer;
 
    streamer = palloc0(sizeof(astreamer_tar_parser));
    *((const astreamer_ops **) &streamer->base.bbs_ops) =
        &astreamer_tar_parser_ops;
    streamer->base.bbs_next = next;
    initStringInfo(&streamer->base.bbs_buffer);
    streamer->next_context = ASTREAMER_MEMBER_HEADER;
 
    return &streamer->base;
}

References ASTREAMER_MEMBER_HEADER, astreamer_tar_parser_ops, astreamer_tar_parser::base, astreamer::bbs_buffer, astreamer::bbs_next, astreamer::bbs_ops, initStringInfo(), next, astreamer_tar_parser::next_context, and palloc0().

Referenced by create_archive_verifier(), and CreateBackupStreamer().

astreamer_tar_terminator_content()

static void astreamer_tar_terminator_content ( astreamer *  streamer, astreamer_member *  member, const char *  data, int  len, astreamer_archive_context  context  )

static

Definition at line 478 of file astreamer_tar.c.

{
    /* Expect unparsed input. */
    Assert(member == NULL);
    Assert(context == ASTREAMER_UNKNOWN);
 
    /* Just forward it. */
    astreamer_content(streamer->bbs_next, member, data, len, context);
}

References Assert, astreamer_content(), ASTREAMER_UNKNOWN, astreamer::bbs_next, data, and len.

astreamer_tar_terminator_finalize()

static void astreamer_tar_terminator_finalize ( astreamer *  streamer )

static

Definition at line 496 of file astreamer_tar.c.

{
    char        buffer[2 * TAR_BLOCK_SIZE];
 
    memset(buffer, 0, 2 * TAR_BLOCK_SIZE);
    astreamer_content(streamer->bbs_next, NULL, buffer,
                      2 * TAR_BLOCK_SIZE, ASTREAMER_UNKNOWN);
    astreamer_finalize(streamer->bbs_next);
}

References astreamer_content(), astreamer_finalize(), ASTREAMER_UNKNOWN, astreamer::bbs_next, and TAR_BLOCK_SIZE.

astreamer_tar_terminator_free()

static void astreamer_tar_terminator_free ( astreamer *  streamer )

static

Definition at line 510 of file astreamer_tar.c.

{
    astreamer_free(streamer->bbs_next);
    pfree(streamer);
}

References astreamer_free(), astreamer::bbs_next, and pfree().

astreamer_tar_terminator_new()

astreamer * astreamer_tar_terminator_new

(

astreamer *

next

)

Definition at line 462 of file astreamer_tar.c.

{
    astreamer  *streamer;
 
    streamer = palloc0(sizeof(astreamer));
    *((const astreamer_ops **) &streamer->bbs_ops) =
        &astreamer_tar_terminator_ops;
    streamer->bbs_next = next;
 
    return streamer;
}

References astreamer_tar_terminator_ops, astreamer::bbs_next, astreamer::bbs_ops, next, and palloc0().

Referenced by CreateBackupStreamer().

Variable Documentation

astreamer_tar_archiver_ops

const astreamer_ops astreamer_tar_archiver_ops

static

Initial value:

= {
    .content = astreamer_tar_archiver_content,
    .finalize = astreamer_tar_archiver_finalize,
    .free = astreamer_tar_archiver_free
}

Definition at line 66 of file astreamer_tar.c.

Referenced by astreamer_tar_archiver_new().

astreamer_tar_parser_ops

const astreamer_ops astreamer_tar_parser_ops

static

Initial value:

= {
    .content = astreamer_tar_parser_content,
    .finalize = astreamer_tar_parser_finalize,
    .free = astreamer_tar_parser_free
}

Definition at line 53 of file astreamer_tar.c.

Referenced by astreamer_tar_parser_new().

astreamer_tar_terminator_ops

const astreamer_ops astreamer_tar_terminator_ops

static

Initial value:

= {
    .content = astreamer_tar_terminator_content,
    .finalize = astreamer_tar_terminator_finalize,
    .free = astreamer_tar_terminator_free
}

Definition at line 79 of file astreamer_tar.c.

Referenced by astreamer_tar_terminator_new().