db/pg_probackup
1
0
Fork 0
mirror of https://github.com/postgrespro/pg_probackup.git synced 2026-06-21 01:34:15 +02:00
Code Issues Releases Activity

[PBCKP-318] rewrite backup_non_data_file to pio.

Browse Source
This commit is contained in:
Daniil Shelepanov
2022-12-20 00:16:01 +03:00
committed by Yuriy Sokolov
parent e43dfb47a2
commit 2b11c59e48
3 changed files with 170 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/data.c
+69 -46
View File
@@ -71,6 +71,8 @@ static size_t restore_data_file_internal(pioReader_i in, pioDBWriter_i out, pgFi
datapagemap_t *map, PageState *checksum_map, int checksum_version,
datapagemap_t *lsn_map, BackupPageHeader2 *headers);
static err_i send_file(const char *to_fullpath, const char *from_path, bool cut_zero_tail, pgFile *file);
#ifdef HAVE_LIBZ
/* Implementation of zlib compression method */
static int32
@@ -1153,11 +1155,10 @@ backup_non_data_file_internal(const char *from_fullpath,
const char *to_fullpath, pgFile *file,
bool missing_ok)
{
FILE *out = NULL;
char *errmsg = NULL;
int rc;
bool cut_zero_tail;
err_i err;
FOBJ_FUNC_ARP();
cut_zero_tail = file->forkName == cfm;
INIT_CRC32C(file->crc);
@@ -1167,57 +1168,79 @@ backup_non_data_file_internal(const char *from_fullpath,
file->write_size = 0;
file->uncompressed_size = 0;
/* open backup file for write */
out = fopen(to_fullpath, PG_BINARY_W);
if (out == NULL)
elog(ERROR, "Cannot open destination file \"%s\": %s",
to_fullpath, strerror(errno));
/* update file permission */
if (chmod(to_fullpath, file->mode) == -1)
elog(ERROR, "Cannot change mode of \"%s\": %s", to_fullpath,
strerror(errno));
/* backup remote file */
if (fio_is_remote(FIO_DB_HOST))
rc = fio_send_file(from_fullpath, out, cut_zero_tail, file, &errmsg);
else
rc = fio_send_file_local(from_fullpath, out, cut_zero_tail, file, &errmsg);
/* backup non-data file */
err = send_file(to_fullpath, from_fullpath, cut_zero_tail, file);
/* handle errors */
if (rc == FILE_MISSING)
{
/* maybe deleted, it's not error in case of backup */
if (missing_ok)
{
elog(LOG, "File \"%s\" is not found", from_fullpath);
file->write_size = FILE_NOT_FOUND;
goto cleanup;
}
else
elog(ERROR, "File \"%s\" is not found", from_fullpath);
}
else if (rc == WRITE_FAILED)
elog(ERROR, "Cannot write to \"%s\": %s", to_fullpath, strerror(errno));
else if (rc != SEND_OK)
{
if (errmsg)
elog(ERROR, "%s", errmsg);
else
elog(ERROR, "Cannot access remote file \"%s\"", from_fullpath);
if($haserr(err)) {
if(getErrno(err) == ENOENT) {
if(missing_ok) {
elog(LOG, "File \"%s\" is not found", from_fullpath);
file->write_size = FILE_NOT_FOUND;
return;
} else
elog(ERROR, "File \"%s\" is not found", from_fullpath);
} else
elog(ERROR, "An error occured while copying %s: %s",
from_fullpath, $errmsg(err));
}
file->uncompressed_size = file->read_size;
}
static err_i
send_file(const char *to_fullpath, const char *from_fullpath, bool cut_zero_tail, pgFile *file) {
FOBJ_FUNC_ARP();
err_i err = $noerr();
pioReadStream_i in;
pioWriteCloser_i out;
pioDrive_i backup_drive = pioDriveForLocation(FIO_BACKUP_HOST);
pioDrive_i db_drive = pioDriveForLocation(FIO_DB_HOST);
/* open to_fullpath */
out = $i(pioOpenRewrite, backup_drive, .path = to_fullpath,
.permissions = file->mode, .err = &err);
if($haserr(err))
elog(ERROR, "Cannot open destination file \"%s\": %s",
to_fullpath, $errmsg(err));
/* open from_fullpath */
in = $i(pioOpenReadStream, db_drive, .path = from_fullpath, .err = &err);
if($haserr(err))
goto cleanup;
/*
* Copy content and calc CRC as it gets copied. Optionally pioZeroTail
* will be used.
*/
pioCRC32Counter *c = pioCRC32Counter_alloc();
pioZeroTail *zt = pioZeroTail_alloc();
pioFilter_i ztFlt = bind_pioFilter(zt);
pioFilter_i crcFlt = bind_pioFilter(c);
pioFilter_i fltrs[] = { ztFlt, crcFlt };
err = pioCopyWithFilters($reduce(pioWriteFlush, out), $reduce(pioRead, in),
cut_zero_tail ? fltrs : &fltrs[1],
cut_zero_tail ? 2 : 1,
NULL);
if($haserr(err))
goto cleanup;
if (file) {
file->crc = pioCRC32Counter_getCRC32(c);
file->read_size = pioCRC32Counter_getSize(c);
file->write_size = pioCRC32Counter_getSize(c);
}
cleanup:
if (errmsg != NULL)
pg_free(errmsg);
$i(pioClose, in);
$i(pioClose, out);
/* finish CRC calculation and store into pgFile */
FIN_CRC32C(file->crc);
if (out && fclose(out))
elog(ERROR, "Cannot close the file \"%s\": %s", to_fullpath, strerror(errno));
// has $noerr() by default
return $iresult(err);
}
/*
src/utils/file.c
+93
View File
@@ -2711,6 +2711,15 @@ typedef struct pioReSeekableReader {
#define kls__pioReSeekableReader iface__pioReader, mth(fobjDispose)
fobj_klass(pioReSeekableReader);
/* zero tail detector */
typedef struct pioZeroTail {
int64_t read_size;
int64_t write_size;
} pioZeroTail;
#define kls__pioZeroTail iface__pioFilter, mth(fobjDispose), iface(pioFilter)
fobj_klass(pioZeroTail);
/* CRC32 counter */
typedef struct pioDevNull
{
@@ -4918,6 +4927,89 @@ pioWriteFilter_fobjRepr(VSelf)
(filter, self->filter.self));
}
pioZeroTail*
pioZeroTail_alloc(void) {
return $alloc(pioZeroTail, .read_size = 0, .write_size = 0);
}
static pioFltTransformResult
pioZeroTail_pioFltTransform(VSelf, ft_bytes_t rbuf, ft_bytes_t wbuf, err_i *err) {
Self(pioZeroTail);
pioFltTransformResult tr = {0, 0};
fobj_reset_err(err);
size_t non_zero_len = find_zero_tail(rbuf.ptr, rbuf.len);
/*
* It is dirty trick to silence warnings in CFS GC process:
* backup at least cfs header size bytes.
*/
if (self->read_size + non_zero_len < PAGE_ZEROSEARCH_FINE_GRANULARITY &&
self->read_size + rbuf.len > 0)
{
non_zero_len = Min(PAGE_ZEROSEARCH_FINE_GRANULARITY,
self->read_size + rbuf.len);
non_zero_len -= self->read_size;
}
if(non_zero_len > 0) {
/*
* Calculating how many zeroes we can actually copy.
* self->read_size - self->write_size always equals the number of zero bytes.
*/
ssize_t zeroes_to_fill = ft_min(self->read_size - self->write_size, wbuf.len);
/* Restoring the zeroes gap */
memset(wbuf.ptr, 0, zeroes_to_fill);
ft_bytes_consume(&wbuf, zeroes_to_fill);
self->write_size += zeroes_to_fill;
tr.produced += zeroes_to_fill;
/*
* At this moment, wbuf.len will be deacreased by zeroes_to_fill, so it
* represents the room left in wbuf.
*/
if(self->read_size == self->write_size && wbuf.len > 0) {
/*
* All zeroes are in buffer at this point so self->read_size == self->write_size
* and all we need to copy is non_zero_len bytes. This can be done in multiple
* calls and the data will not be lost since tr.consumed will be increased only
* by to_copy bytes.
*/
ft_bytes_t copybuf = ft_bytes_split(&rbuf, ft_min(wbuf.len, non_zero_len));
size_t to_move = ft_bytes_move(&wbuf, &copybuf);
self->write_size += to_move;
tr.consumed = to_move + (to_move < non_zero_len ? 0 : rbuf.len);
self->read_size += tr.consumed;
tr.produced += to_move;
}
/*
* In case there are some unwritten zeroes left tr.consumed == 0 here and the next
* time this filter is called we'll read the same data chunk again.
*/
} else {
// There's a zero tail, skip it for now
self->read_size += rbuf.len;
tr.consumed += rbuf.len;
}
return tr;
}
static size_t
pioZeroTail_pioFltFinish(VSelf, ft_bytes_t wbuf, err_i *err) {
Self(pioZeroTail);
fobj_reset_err(err);
return 0;
}
static void
pioZeroTail_fobjDispose(VSelf) {
}
#ifdef HAVE_LIBZ
#define MAX_WBITS 15 /* 32K LZ77 window */
#define DEF_MEM_LEVEL 8
@@ -6002,6 +6094,7 @@ fobj_klass_handle(pioReadFilter, mth(fobjDispose, fobjRepr));
fobj_klass_handle(pioDevNull);
fobj_klass_handle(pioCRC32Counter);
fobj_klass_handle(pioReSeekableReader);
fobj_klass_handle(pioZeroTail);
#ifdef HAVE_LIBZ
fobj_klass_handle(pioGZCompress, mth(fobjRepr));
src/utils/file.h
+8
View File
@@ -20,6 +20,10 @@
#define DIR_PERMISSION (0700)
#define FILE_PERMISSION (0600)
/* define it now to avoid ugly mess of includes */
struct pgFile;
typedef struct pgFile pgFile;
typedef enum
{
/* message for compatibility check */
@@ -402,6 +406,7 @@ extern pioFilter_i pioGZCompressFilter(int level);
extern pioFilter_i pioGZDecompressFilter(bool ignoreTruncate);
extern pioWrapRead_i pioGZDecompressWrapper(bool ignoreTruncate);
#endif
extern pioFilter_i pioCRC32Filter(void);
typedef struct pioCRC32Counter pioCRC32Counter;
#define kls__pioCRC32Counter iface__pioFilter, mth(pioFltInPlace), iface(pioFilter)
@@ -410,6 +415,9 @@ extern pioCRC32Counter* pioCRC32Counter_alloc(void);
extern pg_crc32 pioCRC32Counter_getCRC32(pioCRC32Counter* flt);
extern int64_t pioCRC32Counter_getSize(pioCRC32Counter* flt);
typedef struct pioZeroTail pioZeroTail;
extern pioZeroTail* pioZeroTail_alloc(void);
extern pioWriteFlush_i pioDevNull_alloc(void);
extern err_i pioCopyWithFilters(pioWriteFlush_i dest, pioRead_i src,