local_source.c File Reference

#include "postgres_fe.h"
#include <fcntl.h>
#include <unistd.h>
#include "datapagemap.h"
#include "file_ops.h"
#include "filemap.h"
#include "pg_rewind.h"
#include "rewind_source.h"

Include dependency graph for local_source.c:

Go to the source code of this file.

Data Structures
struct	local_source

Functions
static void	local_traverse_files (rewind_source *source, process_file_callback_t callback)
static char *	local_fetch_file (rewind_source *source, const char *path, size_t *filesize)
static void	local_queue_fetch_file (rewind_source *source, const char *path, size_t len)
static void	local_queue_fetch_range (rewind_source *source, const char *path, off_t off, size_t len)
static void	local_finish_fetch (rewind_source *source)
static void	local_destroy (rewind_source *source)
rewind_source *	init_local_source (const char *datadir)

Function Documentation

init_local_source()

rewind_source* init_local_source

(

const char *

datadir

)

Definition at line 40 of file local_source.c.

{
    local_source *src;
 
    src = pg_malloc0(sizeof(local_source));
 
    src->common.traverse_files = local_traverse_files;
    src->common.fetch_file = local_fetch_file;
    src->common.queue_fetch_file = local_queue_fetch_file;
    src->common.queue_fetch_range = local_queue_fetch_range;
    src->common.finish_fetch = local_finish_fetch;
    src->common.get_current_wal_insert_lsn = NULL;
    src->common.destroy = local_destroy;
 
    src->datadir = datadir;
 
    return &src->common;
}

References local_source::common, local_source::datadir, datadir, rewind_source::destroy, rewind_source::fetch_file, rewind_source::finish_fetch, rewind_source::get_current_wal_insert_lsn, local_destroy(), local_fetch_file(), local_finish_fetch(), local_queue_fetch_file(), local_queue_fetch_range(), local_traverse_files(), pg_malloc0(), rewind_source::queue_fetch_file, rewind_source::queue_fetch_range, and rewind_source::traverse_files.

Referenced by main().

local_destroy()

static void local_destroy ( rewind_source *  source )

static

Definition at line 184 of file local_source.c.

{
    pfree(source);
}

References pfree(), and source.

Referenced by init_local_source().

local_fetch_file()

static char * local_fetch_file ( rewind_source *  source, const char *  path, size_t *  filesize  )

static

Definition at line 66 of file local_source.c.

{
    return slurpFile(((local_source *) source)->datadir, path, filesize);
}

References datadir, slurpFile(), and source.

Referenced by init_local_source().

local_finish_fetch()

static void local_finish_fetch ( rewind_source *  source )

static

Definition at line 176 of file local_source.c.

{
    /*
     * Nothing to do, local_queue_fetch_range() copies the ranges immediately.
     */
}

Referenced by init_local_source().

local_queue_fetch_file()

static void local_queue_fetch_file ( rewind_source *  source, const char *  path, size_t  len  )

static

Definition at line 77 of file local_source.c.

{
    const char *datadir = ((local_source *) source)->datadir;
    PGIOAlignedBlock buf;
    char        srcpath[MAXPGPATH];
    int         srcfd;
    size_t      written_len;
 
    snprintf(srcpath, sizeof(srcpath), "%s/%s", datadir, path);
 
    /* Open source file for reading */
    srcfd = open(srcpath, O_RDONLY | PG_BINARY, 0);
    if (srcfd < 0)
        pg_fatal("could not open source file \"%s\": %m",
                 srcpath);
 
    /* Truncate and open the target file for writing */
    open_target_file(path, true);
 
    written_len = 0;
    for (;;)
    {
        ssize_t     read_len;
 
        read_len = read(srcfd, buf.data, sizeof(buf));
 
        if (read_len < 0)
            pg_fatal("could not read file \"%s\": %m", srcpath);
        else if (read_len == 0)
            break;              /* EOF reached */
 
        write_target_range(buf.data, written_len, read_len);
        written_len += read_len;
    }
 
    /*
     * A local source is not expected to change while we're rewinding, so
     * check that the size of the file matches our earlier expectation.
     */
    if (written_len != len)
        pg_fatal("size of source file \"%s\" changed concurrently: %d bytes expected, %d copied",
                 srcpath, (int) len, (int) written_len);
 
    if (close(srcfd) != 0)
        pg_fatal("could not close file \"%s\": %m", srcpath);
}

References buf, close, datadir, len, MAXPGPATH, open_target_file(), PG_BINARY, pg_fatal, read, snprintf, source, and write_target_range().

Referenced by init_local_source().

local_queue_fetch_range()

static void local_queue_fetch_range ( rewind_source *  source, const char *  path, off_t  off, size_t  len  )

static

Definition at line 128 of file local_source.c.

{
    const char *datadir = ((local_source *) source)->datadir;
    PGIOAlignedBlock buf;
    char        srcpath[MAXPGPATH];
    int         srcfd;
    off_t       begin = off;
    off_t       end = off + len;
 
    snprintf(srcpath, sizeof(srcpath), "%s/%s", datadir, path);
 
    srcfd = open(srcpath, O_RDONLY | PG_BINARY, 0);
    if (srcfd < 0)
        pg_fatal("could not open source file \"%s\": %m",
                 srcpath);
 
    if (lseek(srcfd, begin, SEEK_SET) == -1)
        pg_fatal("could not seek in source file: %m");
 
    open_target_file(path, false);
 
    while (end - begin > 0)
    {
        ssize_t     readlen;
        size_t      thislen;
 
        if (end - begin > sizeof(buf))
            thislen = sizeof(buf);
        else
            thislen = end - begin;
 
        readlen = read(srcfd, buf.data, thislen);
 
        if (readlen < 0)
            pg_fatal("could not read file \"%s\": %m", srcpath);
        else if (readlen == 0)
            pg_fatal("unexpected EOF while reading file \"%s\"", srcpath);
 
        write_target_range(buf.data, begin, readlen);
        begin += readlen;
    }
 
    if (close(srcfd) != 0)
        pg_fatal("could not close file \"%s\": %m", srcpath);
}

References buf, close, datadir, len, MAXPGPATH, open_target_file(), PG_BINARY, pg_fatal, read, snprintf, source, and write_target_range().

Referenced by init_local_source().

local_traverse_files()

static void local_traverse_files ( rewind_source *  source, process_file_callback_t  callback  )

static

Definition at line 60 of file local_source.c.

{
    traverse_datadir(((local_source *) source)->datadir, callback);
}

References callback(), datadir, source, and traverse_datadir().

Referenced by init_local_source().