mirror of https://github.com/postgres/postgres
This makes the abstraction of a "source" server more clear, by introducing a common abstract class, borrowing the object-oriented programming term, that represents all the operations that can be done on the source server. There are two implementations of it, one for fetching via libpq, and another to fetch from a local directory. This adds some code, but makes it easier to understand what's going on. The copy_executeFileMap() and libpq_executeFileMap() functions contained basically the same logic, just calling different functions to fetch the source files. Refactor so that the common logic is in one place, in a new function called perform_rewind(). Reviewed-by: Kyotaro Horiguchi, Soumyadeep Chakraborty Discussion: https://www.postgresql.org/message-id/0c5b3783-af52-3ee5-f8fa-6e794061f70d%40iki.fipull/57/head
Heikki Linnakangas
5 years ago
parent
f81e97d047
commit
37d2ff3803
@ -1,266 +0,0 @@ |
||||
/*-------------------------------------------------------------------------
|
||||
* |
||||
* copy_fetch.c |
||||
* Functions for using a data directory as the source. |
||||
* |
||||
* Portions Copyright (c) 2013-2020, PostgreSQL Global Development Group |
||||
* |
||||
*------------------------------------------------------------------------- |
||||
*/ |
||||
#include "postgres_fe.h" |
||||
|
||||
#include <sys/stat.h> |
||||
#include <dirent.h> |
||||
#include <fcntl.h> |
||||
#include <unistd.h> |
||||
|
||||
#include "datapagemap.h" |
||||
#include "fetch.h" |
||||
#include "file_ops.h" |
||||
#include "filemap.h" |
||||
#include "pg_rewind.h" |
||||
|
||||
static void recurse_dir(const char *datadir, const char *path, |
||||
process_file_callback_t callback); |
||||
|
||||
static void execute_pagemap(datapagemap_t *pagemap, const char *path); |
||||
|
||||
/*
|
||||
* Traverse through all files in a data directory, calling 'callback' |
||||
* for each file. |
||||
*/ |
||||
void |
||||
traverse_datadir(const char *datadir, process_file_callback_t callback) |
||||
{ |
||||
recurse_dir(datadir, NULL, callback); |
||||
} |
||||
|
||||
/*
|
||||
* recursive part of traverse_datadir |
||||
* |
||||
* parentpath is the current subdirectory's path relative to datadir, |
||||
* or NULL at the top level. |
||||
*/ |
||||
static void |
||||
recurse_dir(const char *datadir, const char *parentpath, |
||||
process_file_callback_t callback) |
||||
{ |
||||
DIR *xldir; |
||||
struct dirent *xlde; |
||||
char fullparentpath[MAXPGPATH]; |
||||
|
||||
if (parentpath) |
||||
snprintf(fullparentpath, MAXPGPATH, "%s/%s", datadir, parentpath); |
||||
else |
||||
snprintf(fullparentpath, MAXPGPATH, "%s", datadir); |
||||
|
||||
xldir = opendir(fullparentpath); |
||||
if (xldir == NULL) |
||||
pg_fatal("could not open directory \"%s\": %m", |
||||
fullparentpath); |
||||
|
||||
while (errno = 0, (xlde = readdir(xldir)) != NULL) |
||||
{ |
||||
struct stat fst; |
||||
char fullpath[MAXPGPATH * 2]; |
||||
char path[MAXPGPATH * 2]; |
||||
|
||||
if (strcmp(xlde->d_name, ".") == 0 || |
||||
strcmp(xlde->d_name, "..") == 0) |
||||
continue; |
||||
|
||||
snprintf(fullpath, sizeof(fullpath), "%s/%s", fullparentpath, xlde->d_name); |
||||
|
||||
if (lstat(fullpath, &fst) < 0) |
||||
{ |
||||
if (errno == ENOENT) |
||||
{ |
||||
/*
|
||||
* File doesn't exist anymore. This is ok, if the new primary |
||||
* is running and the file was just removed. If it was a data |
||||
* file, there should be a WAL record of the removal. If it |
||||
* was something else, it couldn't have been anyway. |
||||
* |
||||
* TODO: But complain if we're processing the target dir! |
||||
*/ |
||||
} |
||||
else |
||||
pg_fatal("could not stat file \"%s\": %m", |
||||
fullpath); |
||||
} |
||||
|
||||
if (parentpath) |
||||
snprintf(path, sizeof(path), "%s/%s", parentpath, xlde->d_name); |
||||
else |
||||
snprintf(path, sizeof(path), "%s", xlde->d_name); |
||||
|
||||
if (S_ISREG(fst.st_mode)) |
||||
callback(path, FILE_TYPE_REGULAR, fst.st_size, NULL); |
||||
else if (S_ISDIR(fst.st_mode)) |
||||
{ |
||||
callback(path, FILE_TYPE_DIRECTORY, 0, NULL); |
||||
/* recurse to handle subdirectories */ |
||||
recurse_dir(datadir, path, callback); |
||||
} |
||||
#ifndef WIN32 |
||||
else if (S_ISLNK(fst.st_mode)) |
||||
#else |
||||
else if (pgwin32_is_junction(fullpath)) |
||||
#endif |
||||
{ |
||||
#if defined(HAVE_READLINK) || defined(WIN32) |
||||
char link_target[MAXPGPATH]; |
||||
int len; |
||||
|
||||
len = readlink(fullpath, link_target, sizeof(link_target)); |
||||
if (len < 0) |
||||
pg_fatal("could not read symbolic link \"%s\": %m", |
||||
fullpath); |
||||
if (len >= sizeof(link_target)) |
||||
pg_fatal("symbolic link \"%s\" target is too long", |
||||
fullpath); |
||||
link_target[len] = '\0'; |
||||
|
||||
callback(path, FILE_TYPE_SYMLINK, 0, link_target); |
||||
|
||||
/*
|
||||
* If it's a symlink within pg_tblspc, we need to recurse into it, |
||||
* to process all the tablespaces. We also follow a symlink if |
||||
* it's for pg_wal. Symlinks elsewhere are ignored. |
||||
*/ |
||||
if ((parentpath && strcmp(parentpath, "pg_tblspc") == 0) || |
||||
strcmp(path, "pg_wal") == 0) |
||||
recurse_dir(datadir, path, callback); |
||||
#else |
||||
pg_fatal("\"%s\" is a symbolic link, but symbolic links are not supported on this platform", |
||||
fullpath); |
||||
#endif /* HAVE_READLINK */ |
||||
} |
||||
} |
||||
|
||||
if (errno) |
||||
pg_fatal("could not read directory \"%s\": %m", |
||||
fullparentpath); |
||||
|
||||
if (closedir(xldir)) |
||||
pg_fatal("could not close directory \"%s\": %m", |
||||
fullparentpath); |
||||
} |
||||
|
||||
/*
|
||||
* Copy a file from source to target, between 'begin' and 'end' offsets. |
||||
* |
||||
* If 'trunc' is true, any existing file with the same name is truncated. |
||||
*/ |
||||
static void |
||||
rewind_copy_file_range(const char *path, off_t begin, off_t end, bool trunc) |
||||
{ |
||||
PGAlignedBlock buf; |
||||
char srcpath[MAXPGPATH]; |
||||
int srcfd; |
||||
|
||||
snprintf(srcpath, sizeof(srcpath), "%s/%s", datadir_source, 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, trunc); |
||||
|
||||
while (end - begin > 0) |
||||
{ |
||||
int readlen; |
||||
int len; |
||||
|
||||
if (end - begin > sizeof(buf)) |
||||
len = sizeof(buf); |
||||
else |
||||
len = end - begin; |
||||
|
||||
readlen = read(srcfd, buf.data, len); |
||||
|
||||
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); |
||||
} |
||||
|
||||
/*
|
||||
* Copy all relation data files from datadir_source to datadir_target, which |
||||
* are marked in the given data page map. |
||||
*/ |
||||
void |
||||
copy_executeFileMap(filemap_t *map) |
||||
{ |
||||
file_entry_t *entry; |
||||
int i; |
||||
|
||||
for (i = 0; i < map->nentries; i++) |
||||
{ |
||||
entry = map->entries[i]; |
||||
execute_pagemap(&entry->target_pages_to_overwrite, entry->path); |
||||
|
||||
switch (entry->action) |
||||
{ |
||||
case FILE_ACTION_NONE: |
||||
/* ok, do nothing.. */ |
||||
break; |
||||
|
||||
case FILE_ACTION_COPY: |
||||
rewind_copy_file_range(entry->path, 0, entry->source_size, true); |
||||
break; |
||||
|
||||
case FILE_ACTION_TRUNCATE: |
||||
truncate_target_file(entry->path, entry->source_size); |
||||
break; |
||||
|
||||
case FILE_ACTION_COPY_TAIL: |
||||
rewind_copy_file_range(entry->path, entry->target_size, |
||||
entry->source_size, false); |
||||
break; |
||||
|
||||
case FILE_ACTION_CREATE: |
||||
create_target(entry); |
||||
break; |
||||
|
||||
case FILE_ACTION_REMOVE: |
||||
remove_target(entry); |
||||
break; |
||||
|
||||
case FILE_ACTION_UNDECIDED: |
||||
pg_fatal("no action decided for \"%s\"", entry->path); |
||||
break; |
||||
} |
||||
} |
||||
|
||||
close_target_file(); |
||||
} |
||||
|
||||
static void |
||||
execute_pagemap(datapagemap_t *pagemap, const char *path) |
||||
{ |
||||
datapagemap_iterator_t *iter; |
||||
BlockNumber blkno; |
||||
off_t offset; |
||||
|
||||
iter = datapagemap_iterate(pagemap); |
||||
while (datapagemap_next(iter, &blkno)) |
||||
{ |
||||
offset = blkno * BLCKSZ; |
||||
rewind_copy_file_range(path, offset, offset + BLCKSZ, false); |
||||
/* Ok, this block has now been copied from new data dir to old */ |
||||
} |
||||
pg_free(iter); |
||||
} |
||||
@ -1,60 +0,0 @@ |
||||
/*-------------------------------------------------------------------------
|
||||
* |
||||
* fetch.c |
||||
* Functions for fetching files from a local or remote data dir |
||||
* |
||||
* This file forms an abstraction of getting files from the "source". |
||||
* There are two implementations of this interface: one for copying files |
||||
* from a data directory via normal filesystem operations (copy_fetch.c), |
||||
* and another for fetching files from a remote server via a libpq |
||||
* connection (libpq_fetch.c) |
||||
* |
||||
* |
||||
* Portions Copyright (c) 1996-2020, PostgreSQL Global Development Group |
||||
* |
||||
*------------------------------------------------------------------------- |
||||
*/ |
||||
#include "postgres_fe.h" |
||||
|
||||
#include <sys/stat.h> |
||||
#include <unistd.h> |
||||
|
||||
#include "fetch.h" |
||||
#include "file_ops.h" |
||||
#include "filemap.h" |
||||
#include "pg_rewind.h" |
||||
|
||||
void |
||||
fetchSourceFileList(void) |
||||
{ |
||||
if (datadir_source) |
||||
traverse_datadir(datadir_source, &process_source_file); |
||||
else |
||||
libpqProcessFileList(); |
||||
} |
||||
|
||||
/*
|
||||
* Fetch all relation data files that are marked in the given data page map. |
||||
*/ |
||||
void |
||||
execute_file_actions(filemap_t *filemap) |
||||
{ |
||||
if (datadir_source) |
||||
copy_executeFileMap(filemap); |
||||
else |
||||
libpq_executeFileMap(filemap); |
||||
} |
||||
|
||||
/*
|
||||
* Fetch a single file into a malloc'd buffer. The file size is returned |
||||
* in *filesize. The returned buffer is always zero-terminated, which is |
||||
* handy for text files. |
||||
*/ |
||||
char * |
||||
fetchFile(const char *filename, size_t *filesize) |
||||
{ |
||||
if (datadir_source) |
||||
return slurpFile(datadir_source, filename, filesize); |
||||
else |
||||
return libpqGetFile(filename, filesize); |
||||
} |
||||
@ -1,44 +0,0 @@ |
||||
/*-------------------------------------------------------------------------
|
||||
* |
||||
* fetch.h |
||||
* Fetching data from a local or remote data directory. |
||||
* |
||||
* This file includes the prototypes for functions used to copy files from |
||||
* one data directory to another. The source to copy from can be a local |
||||
* directory (copy method), or a remote PostgreSQL server (libpq fetch |
||||
* method). |
||||
* |
||||
* Copyright (c) 2013-2020, PostgreSQL Global Development Group |
||||
* |
||||
*------------------------------------------------------------------------- |
||||
*/ |
||||
#ifndef FETCH_H |
||||
#define FETCH_H |
||||
|
||||
#include "access/xlogdefs.h" |
||||
|
||||
#include "filemap.h" |
||||
|
||||
/*
|
||||
* Common interface. Calls the copy or libpq method depending on global |
||||
* config options. |
||||
*/ |
||||
extern void fetchSourceFileList(void); |
||||
extern char *fetchFile(const char *filename, size_t *filesize); |
||||
extern void execute_file_actions(filemap_t *filemap); |
||||
|
||||
/* in libpq_fetch.c */ |
||||
extern void libpqProcessFileList(void); |
||||
extern char *libpqGetFile(const char *filename, size_t *filesize); |
||||
extern void libpq_executeFileMap(filemap_t *map); |
||||
|
||||
extern void libpqConnect(const char *connstr); |
||||
extern XLogRecPtr libpqGetCurrentXlogInsertLocation(void); |
||||
|
||||
/* in copy_fetch.c */ |
||||
extern void copy_executeFileMap(filemap_t *map); |
||||
|
||||
typedef void (*process_file_callback_t) (const char *path, file_type_t type, size_t size, const char *link_target); |
||||
extern void traverse_datadir(const char *datadir, process_file_callback_t callback); |
||||
|
||||
#endif /* FETCH_H */ |
||||
@ -0,0 +1,131 @@ |
||||
/*-------------------------------------------------------------------------
|
||||
* |
||||
* local_source.c |
||||
* Functions for using a local data directory as the source. |
||||
* |
||||
* Portions Copyright (c) 2013-2020, PostgreSQL Global Development Group |
||||
* |
||||
*------------------------------------------------------------------------- |
||||
*/ |
||||
#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" |
||||
|
||||
typedef struct |
||||
{ |
||||
rewind_source common; /* common interface functions */ |
||||
|
||||
const char *datadir; /* path to the source data directory */ |
||||
} local_source; |
||||
|
||||
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_fetch_file_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) |
||||
{ |
||||
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_range = local_fetch_file_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; |
||||
} |
||||
|
||||
static void |
||||
local_traverse_files(rewind_source *source, process_file_callback_t callback) |
||||
{ |
||||
traverse_datadir(((local_source *) source)->datadir, &process_source_file); |
||||
} |
||||
|
||||
static char * |
||||
local_fetch_file(rewind_source *source, const char *path, size_t *filesize) |
||||
{ |
||||
return slurpFile(((local_source *) source)->datadir, path, filesize); |
||||
} |
||||
|
||||
/*
|
||||
* Copy a file from source to target, starting at 'off', for 'len' bytes. |
||||
*/ |
||||
static void |
||||
local_fetch_file_range(rewind_source *source, const char *path, off_t off, |
||||
size_t len) |
||||
{ |
||||
const char *datadir = ((local_source *) source)->datadir; |
||||
PGAlignedBlock 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 len; |
||||
|
||||
if (end - begin > sizeof(buf)) |
||||
len = sizeof(buf); |
||||
else |
||||
len = end - begin; |
||||
|
||||
readlen = read(srcfd, buf.data, len); |
||||
|
||||
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); |
||||
} |
||||
|
||||
static void |
||||
local_finish_fetch(rewind_source *source) |
||||
{ |
||||
/*
|
||||
* Nothing to do, local_fetch_file_range() copies the ranges immediately. |
||||
*/ |
||||
} |
||||
|
||||
static void |
||||
local_destroy(rewind_source *source) |
||||
{ |
||||
pfree(source); |
||||
} |
||||
@ -0,0 +1,73 @@ |
||||
/*-------------------------------------------------------------------------
|
||||
* |
||||
* rewind_source.h |
||||
* Abstraction for fetching from source server. |
||||
* |
||||
* The source server can be either a libpq connection to a live system, |
||||
* or a local data directory. The 'rewind_source' struct abstracts the |
||||
* operations to fetch data from the source system, so that the rest of |
||||
* the code doesn't need to care what kind of a source its dealing with. |
||||
* |
||||
* Copyright (c) 2013-2020, PostgreSQL Global Development Group |
||||
* |
||||
*------------------------------------------------------------------------- |
||||
*/ |
||||
#ifndef REWIND_SOURCE_H |
||||
#define REWIND_SOURCE_H |
||||
|
||||
#include "access/xlogdefs.h" |
||||
#include "file_ops.h" |
||||
#include "filemap.h" |
||||
#include "libpq-fe.h" |
||||
|
||||
typedef struct rewind_source |
||||
{ |
||||
/*
|
||||
* Traverse all files in the source data directory, and call 'callback' on |
||||
* each file. |
||||
*/ |
||||
void (*traverse_files) (struct rewind_source *, |
||||
process_file_callback_t callback); |
||||
|
||||
/*
|
||||
* Fetch a single file into a malloc'd buffer. The file size is returned |
||||
* in *filesize. The returned buffer is always zero-terminated, which is |
||||
* handy for text files. |
||||
*/ |
||||
char *(*fetch_file) (struct rewind_source *, const char *path, |
||||
size_t *filesize); |
||||
|
||||
/*
|
||||
* Request to fetch (part of) a file in the source system, specified by an |
||||
* offset and length, and write it to the same offset in the corresponding |
||||
* target file. The source implementation may queue up the request and |
||||
* execute it later when convenient. Call finish_fetch() to flush the |
||||
* queue and execute all requests. |
||||
*/ |
||||
void (*queue_fetch_range) (struct rewind_source *, const char *path, |
||||
off_t offset, size_t len); |
||||
|
||||
/*
|
||||
* Execute all requests queued up with queue_fetch_range(). |
||||
*/ |
||||
void (*finish_fetch) (struct rewind_source *); |
||||
|
||||
/*
|
||||
* Get the current WAL insert position in the source system. |
||||
*/ |
||||
XLogRecPtr (*get_current_wal_insert_lsn) (struct rewind_source *); |
||||
|
||||
/*
|
||||
* Free this rewind_source object. |
||||
*/ |
||||
void (*destroy) (struct rewind_source *); |
||||
|
||||
} rewind_source; |
||||
|
||||
/* in libpq_source.c */ |
||||
extern rewind_source *init_libpq_source(PGconn *conn); |
||||
|
||||
/* in local_source.c */ |
||||
extern rewind_source *init_local_source(const char *datadir); |
||||
|
||||
#endif /* FETCH_H */ |
||||
Loading…
Reference in new issue