From 2b11c59e48f9c90c6b374fec77555314e24d4103 Mon Sep 17 00:00:00 2001 From: Daniil Shelepanov Date: Tue, 20 Dec 2022 00:16:01 +0300 Subject: [PATCH] [PBCKP-318] rewrite backup_non_data_file to pio. --- src/data.c | 115 ++++++++++++++++++++++++++++------------------- src/utils/file.c | 93 ++++++++++++++++++++++++++++++++++++++ src/utils/file.h | 8 ++++ 3 files changed, 170 insertions(+), 46 deletions(-) diff --git a/src/data.c b/src/data.c index 35eece9f..356b2b0d 100644 --- a/src/data.c +++ b/src/data.c @@ -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); } /* diff --git a/src/utils/file.c b/src/utils/file.c index f8aa4d84..192662c1 100644 --- a/src/utils/file.c +++ b/src/utils/file.c @@ -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, ©buf); + + 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)); diff --git a/src/utils/file.h b/src/utils/file.h index 8a2fe1c7..a7e1069d 100644 --- a/src/utils/file.h +++ b/src/utils/file.h @@ -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,