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

Merge branch 'simplify_backup_data_file' into ptrack

Browse Source
This commit is contained in:
Artur Zakirov
2017-02-15 12:18:42 +03:00
parent 4cb028cd97 547470abb8
commit 0bdffbcfe5
5 changed files with 649 additions and 300 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backup.c
+173 -13
View File
@@ -1105,6 +1105,75 @@ backup_cleanup(bool fatal, void *userdata)
}
}
/* Count bytes in file */
static long
file_size(const char *file)
{
long r;
FILE *f = fopen(file, "r");
if (!f)
{
elog(ERROR, "pg_probackup: could not open file \"%s\" for reading: %s\n",
file, strerror(errno));
return -1;
}
fseek(f, 0, SEEK_END);
r = ftell(f);
fclose(f);
return r;
}
/*
* Find corresponding file in previous backup.
* Compare generations and return true if we don't need full copy
* of the file, but just part of it.
*
* skip_size - size of the file in previous backup. We can skip it
* and copy just remaining part of the file.
*/
bool
backup_compressed_file_partially(pgFile *file, void *arg, size_t *skip_size)
{
bool result = false;
pgFile *prev_file = NULL;
size_t current_file_size;
backup_files_args *arguments = (backup_files_args *) arg;
if (arguments->prev_files)
{
pgFile **p = (pgFile **) parray_bsearch(arguments->prev_files,
file, pgFileComparePath);
if (p)
prev_file = *p;
/* If file's gc generation has changed since last backup, just copy it*/
if (prev_file && prev_file->generation == file->generation)
{
current_file_size = file_size(file->path);
if (prev_file->write_size == BYTES_INVALID)
return false;
*skip_size = prev_file->write_size;
if (current_file_size >= prev_file->write_size)
{
elog(LOG, "Backup file %s partially: prev_size %lu, current_size %lu",
file->path, prev_file->write_size, current_file_size);
result = true;
}
else
elog(ERROR, "Something is wrong with %s. current_file_size %lu, prev %lu",
file->path, current_file_size, prev_file->write_size);
}
else
elog(LOG, "Copy full %s.", file->path);
}
return result;
}
/*
* Take differential backup at page level.
*/
@@ -1203,9 +1272,47 @@ backup_files(void *arg)
}
/* copy the file into backup */
if (!(file->is_datafile
? backup_data_file(arguments->from_root, arguments->to_root, file, arguments->lsn)
: copy_file(arguments->from_root, arguments->to_root, file)))
if (file->is_datafile)
{
if (!backup_data_file(arguments->from_root,
arguments->to_root, file,
arguments->lsn))
{
/* record as skipped file in file_xxx.txt */
file->write_size = BYTES_INVALID;
elog(LOG, "skip");
continue;
}
}
else if (is_compressed_data_file(file))
{
size_t skip_size = 0;
if (backup_compressed_file_partially(file, arguments, &skip_size))
{
/* backup cfs segment partly */
if (!copy_file_partly(arguments->from_root,
arguments->to_root,
file, skip_size))
{
/* record as skipped file in file_xxx.txt */
file->write_size = BYTES_INVALID;
elog(LOG, "skip");
continue;
}
}
else if (!copy_file(arguments->from_root,
arguments->to_root,
file))
{
/* record as skipped file in file_xxx.txt */
file->write_size = BYTES_INVALID;
elog(LOG, "skip");
continue;
}
}
else if (!copy_file(arguments->from_root,
arguments->to_root,
file))
{
/* record as skipped file in file_xxx.txt */
file->write_size = BYTES_INVALID;
@@ -1254,14 +1361,14 @@ add_files(parray *files, const char *root, bool add_root, bool is_pgdata)
relative = file->path + strlen(root) + 1;
if (is_pgdata &&
!path_is_prefix_of_path("base", relative) &&
/*!path_is_prefix_of_path("global", relative) &&*/
/*!path_is_prefix_of_path("global", relative) &&*/ //TODO What's wrong with this line?
!path_is_prefix_of_path("pg_tblspc", relative))
continue;
/* Get file name from path */
fname = last_dir_separator(relative);
/* Remove temp tables */
/* Remove temp tables from the list */
if (fname[0] == 't' && isdigit(fname[1]))
{
pgFileFree(file);
@@ -1271,7 +1378,7 @@ add_files(parray *files, const char *root, bool add_root, bool is_pgdata)
}
path_len = strlen(file->path);
/* Get link ptrack file to realations files */
/* Get link ptrack file to relations files */
if (path_len > 6 && strncmp(file->path+(path_len-6), "ptrack", 6) == 0)
{
pgFile *search_file;
@@ -1280,12 +1387,15 @@ add_files(parray *files, const char *root, bool add_root, bool is_pgdata)
while(true) {
pgFile tmp_file;
tmp_file.path = pg_strdup(file->path);
/* I hope segno not more than 999999 */
/* Segno fits into 6 digits since it is not more than 4000 */
if (segno > 0)
sprintf(tmp_file.path+path_len-7, ".%d", segno);
else
tmp_file.path[path_len-7] = '\0';
pre_search_file = (pgFile **) parray_bsearch(list_file, &tmp_file, pgFileComparePath);
if (pre_search_file != NULL)
{
search_file = *pre_search_file;
@@ -1299,6 +1409,7 @@ add_files(parray *files, const char *root, bool add_root, bool is_pgdata)
segno++;
}
/* Remove ptrack file itself from backup list */
pgFileFree(file);
parray_remove(list_file, i);
i--;
@@ -1307,12 +1418,7 @@ add_files(parray *files, const char *root, bool add_root, bool is_pgdata)
/* compress map file it is not data file */
if (path_len > 4 && strncmp(file->path+(path_len-4), ".cfm", 4) == 0)
{
if (current.backup_mode == BACKUP_MODE_DIFF_PTRACK ||
current.backup_mode == BACKUP_MODE_DIFF_PAGE)
elog(ERROR, "You can't use incremental backup with compress tablespace");
continue;
}
/* name of data file start with digit */
if (fname == NULL)
@@ -1358,11 +1464,34 @@ add_files(parray *files, const char *root, bool add_root, bool is_pgdata)
pgFile tmp_file;
tmp_file.path = pg_strdup(file->path);
tmp_file.path[path_len-4] = '\0';
pre_search_file = (pgFile **) parray_bsearch(list_file, &tmp_file, pgFileComparePath);
pre_search_file = (pgFile **) parray_bsearch(list_file,
&tmp_file, pgFileComparePath);
if (pre_search_file != NULL)
{
FileMap* map;
int md = open(file->path, O_RDWR|PG_BINARY, 0);
if (md < 0)
elog(ERROR, "add_files(). cannot open cfm file '%s'", file->path);
map = cfs_mmap(md);
if (map == MAP_FAILED)
{
elog(LOG, "add_files(). cfs_compression_ration failed to map file %s: %m", file->path);
close(md);
break;
}
(*pre_search_file)->generation = map->generation;
(*pre_search_file)->is_datafile = false;
if (cfs_munmap(map) < 0)
elog(LOG, "add_files(). CFS failed to unmap file %s: %m", file->path);
if (close(md) < 0)
elog(LOG, "add_files(). CFS failed to close file %s: %m", file->path);
}
else
elog(ERROR, "corresponding segment '%s' is not found", tmp_file.path);
pg_free(tmp_file.path);
}
}
@@ -1636,3 +1765,34 @@ StreamLog(void *arg)
PQfinish(conn);
conn = NULL;
}
FileMap* cfs_mmap(int md)
{
FileMap* map;
#ifdef WIN32
HANDLE mh = CreateFileMapping(_get_osfhandle(md), NULL, PAGE_READWRITE,
0, (DWORD)sizeof(FileMap), NULL);
if (mh == NULL)
return (FileMap*)MAP_FAILED;
map = (FileMap*)MapViewOfFile(mh, FILE_MAP_ALL_ACCESS, 0, 0, 0);
CloseHandle(mh);
if (map == NULL)
return (FileMap*)MAP_FAILED;
#else
map = (FileMap*)mmap(NULL, sizeof(FileMap),
PROT_WRITE | PROT_READ, MAP_SHARED, md, 0);
#endif
return map;
}
int cfs_munmap(FileMap* map)
{
#ifdef WIN32
return UnmapViewOfFile(map) ? 0 : -1;
#else
return munmap(map, sizeof(FileMap));
#endif
}
data.c
+412 -278
View File
@@ -22,13 +22,10 @@
typedef struct BackupPageHeader
{
BlockNumber block; /* block number */
uint16 hole_offset; /* number of bytes before "hole" */
uint16 hole_length; /* number of bytes in "hole" */
} BackupPageHeader;
static bool
parse_page(const DataPage *page,
XLogRecPtr *lsn, uint16 *offset, uint16 *length)
parse_page(const DataPage *page, XLogRecPtr *lsn)
{
const PageHeaderData *page_data = &page->page_data;
@@ -43,14 +40,124 @@ parse_page(const DataPage *page,
page_data->pd_upper <= page_data->pd_special &&
page_data->pd_special <= BLCKSZ &&
page_data->pd_special == MAXALIGN(page_data->pd_special))
{
*offset = page_data->pd_lower;
*length = page_data->pd_upper - page_data->pd_lower;
return true;
return false;
}
static void
backup_data_page(pgFile *file, const XLogRecPtr *lsn,
BlockNumber blknum, BlockNumber nblocks,
FILE *in, FILE *out,
pg_crc32 *crc)
{
BackupPageHeader header;
off_t offset;
DataPage page; /* used as read buffer */
size_t write_buffer_size = sizeof(header) + BLCKSZ;
char write_buffer[write_buffer_size];
size_t read_len = 0;
XLogRecPtr page_lsn;
int ret;
int try_checksum = 100;
struct stat st;
header.block = blknum;
offset = blknum * BLCKSZ;
while(try_checksum--)
{
ret = fseek(in, offset, SEEK_SET);
if (ret != 0)
elog(ERROR, "Can't seek in file offset: %llu ret:%i\n",
(long long unsigned int) offset, ret);
read_len = fread(&page, 1, sizeof(page), in);
if (read_len != sizeof(page))
{
stat(file->path, &st);
if (st.st_size/BLCKSZ <= blknum)
{
elog(WARNING, "File: %s, file was truncated after backup start."
"Expected nblocks %u. Real nblocks %ld. Cannot read block %u ",
file->path, nblocks, st.st_size/BLCKSZ, blknum);
return;
}
else
elog(ERROR, "File: %s, block size of block %u of nblocks %u is incorrect %lu",
file->path, blknum, nblocks, read_len);
}
/*
* If an invalid data page was found, fallback to simple copy to ensure
* all pages in the file don't have BackupPageHeader.
*/
if (!parse_page(&page, &page_lsn))
{
int i;
/* Check if the page is zeroed. */
for(i = 0; i < BLCKSZ && page.data[i] == 0; i++);
if (i == BLCKSZ)
{
// FIXME Fix this hell.
elog(ERROR, "File: %s blknum %u, empty page", file->path, blknum);
}
/* Try to read and verify this page again several times. */
if (try_checksum)
{
elog(WARNING, "File: %s blknum %u have wrong page header, try again",
file->path, blknum);
usleep(100);
continue;
}
else
elog(ERROR, "File: %s blknum %u have wrong page header.", file->path, blknum);
}
/*
* Verify checksum.
* If it's wrong, sleep a bit and then try again
* several times. If it didn't help, throw error.
*/
if(current.checksum_version &&
pg_checksum_page(page.data, file->segno * RELSEG_SIZE + blknum) != ((PageHeader) page.data)->pd_checksum)
{
if (try_checksum)
{
elog(WARNING, "File: %s blknum %u have wrong checksum, try again",
file->path, blknum);
usleep(100);
}
else
elog(ERROR, "File: %s blknum %u have wrong checksum.",
file->path, blknum);
}
else
try_checksum = 0;
}
*offset = *length = 0;
return false;
file->read_size += read_len;
memcpy(write_buffer, &header, sizeof(header));
memcpy(write_buffer + sizeof(header), page.data, BLCKSZ);
/* write data page excluding hole */
if(fwrite(write_buffer, 1, write_buffer_size, out) != write_buffer_size)
{
int errno_tmp = errno;
fclose(in);
fclose(out);
elog(ERROR, "File: %s, cannot write backup at block %u : %s",
file->path, blknum, strerror(errno_tmp));
}
/* update CRC */
COMP_CRC32C(*crc, &header, sizeof(header));
COMP_CRC32C(*crc, page.data, BLCKSZ);
file->write_size += write_buffer_size;
}
/*
@@ -66,14 +173,10 @@ backup_data_file(const char *from_root, const char *to_root,
char to_path[MAXPGPATH];
FILE *in;
FILE *out;
BackupPageHeader header;
DataPage page; /* used as read buffer */
BlockNumber blknum = 0;
size_t read_len = 0;
BlockNumber nblocks = 0;
pg_crc32 crc;
off_t offset;
char write_buffer[sizeof(header)+BLCKSZ];
size_t write_buffer_real_size;
struct stat st;
INIT_CRC32C(crc);
@@ -95,6 +198,17 @@ backup_data_file(const char *from_root, const char *to_root,
elog(ERROR, "cannot open backup mode file \"%s\": %s",
file->path, strerror(errno));
}
stat(file->path, &st);
if (st.st_size < file->size)
elog(WARNING, "File: %s, file was truncated after backup start. Expected size %lu",
file->path, file->size);
if (file->size % BLCKSZ != 0)
elog(ERROR, "File: %s, file size %lu is incorrect",
file->path, file->size);
nblocks = file->size/BLCKSZ;
/* open backup file for write */
if (check)
@@ -113,7 +227,6 @@ backup_data_file(const char *from_root, const char *to_root,
/* confirm server version */
check_server_version();
/*
* Read each page and write the page excluding hole. If it has been
* determined that the page can be copied safely, but no page map
@@ -124,122 +237,8 @@ backup_data_file(const char *from_root, const char *to_root,
*/
if (file->pagemap.bitmapsize == 0)
{
for (blknum = 0;
(read_len = fread(&page, 1, sizeof(page), in)) == sizeof(page);
++blknum)
{
XLogRecPtr page_lsn;
int upper_offset;
int upper_length;
int try_checksum = 100;
bool stop_backup = false;
header.block = blknum;
while(try_checksum)
{
try_checksum--;
/*
* If an invalid data page was found, fallback to simple copy to ensure
* all pages in the file don't have BackupPageHeader.
*/
if (!parse_page(&page, &page_lsn,
&header.hole_offset, &header.hole_length))
{
struct stat st;
int i;
for(i=0; i<BLCKSZ && page.data[i] == 0; i++);
if (i == BLCKSZ)
{
elog(LOG, "File: %s blknum %u, empty page", file->path, blknum);
goto end_checks;
}
stat(file->path, &st);
elog(WARNING, "SIZE: %lu %lu pages:%lu pages:%lu i:%i", file->size, st.st_size, file->size/BLCKSZ, st.st_size/BLCKSZ, i);
if (st.st_size != file->size && blknum >= file->size/BLCKSZ-1)
{
stop_backup = true;
elog(WARNING, "File: %s blknum %u, file size has changed before backup start", file->path, blknum);
break;
}
if (blknum >= file->size/BLCKSZ-1)
{
stop_backup = true;
elog(WARNING, "File: %s blknum %u, the last page is empty, skip", file->path, blknum);
break;
}
if (st.st_size != file->size && blknum < file->size/BLCKSZ-1)
{
elog(WARNING, "File: %s blknum %u, file size has changed before backup start, it seems bad", file->path, blknum);
if (!try_checksum)
break;
}
if (try_checksum)
{
elog(WARNING, "File: %s blknum %u have wrong page header, try again", file->path, blknum);
fseek(in, -sizeof(page), SEEK_CUR);
fread(&page, 1, sizeof(page), in);
continue;
}
else
elog(ERROR, "File: %s blknum %u have wrong page header.", file->path, blknum);
}
if(current.checksum_version &&
pg_checksum_page(page.data, file->segno * RELSEG_SIZE + blknum) != ((PageHeader) page.data)->pd_checksum)
{
if (try_checksum)
{
elog(WARNING, "File: %s blknum %u have wrong checksum, try again", file->path, blknum);
usleep(100);
fseek(in, -sizeof(page), SEEK_CUR);
fread(&page, 1, sizeof(page), in);
}
else
elog(ERROR, "File: %s blknum %u have wrong checksum.", file->path, blknum);
} else {
try_checksum = 0;
}
}
end_checks:
file->read_size += read_len;
if(stop_backup)
break;
upper_offset = header.hole_offset + header.hole_length;
upper_length = BLCKSZ - upper_offset;
write_buffer_real_size = sizeof(header)+header.hole_offset+upper_length;
memcpy(write_buffer, &header, sizeof(header));
if (header.hole_offset)
memcpy(write_buffer+sizeof(header), page.data, header.hole_offset);
if (upper_length)
memcpy(write_buffer+sizeof(header)+header.hole_offset, page.data + upper_offset, upper_length);
/* write data page excluding hole */
if(fwrite(write_buffer, 1, write_buffer_real_size, out) != write_buffer_real_size)
{
int errno_tmp = errno;
/* oops */
fclose(in);
fclose(out);
elog(ERROR, "cannot write at block %u of \"%s\": %s",
blknum, to_path, strerror(errno_tmp));
}
/* update CRC */
COMP_CRC32C(crc, &header, sizeof(header));
COMP_CRC32C(crc, page.data, header.hole_offset);
COMP_CRC32C(crc, page.data + upper_offset, upper_length);
file->write_size += sizeof(header) + read_len - header.hole_length;
}
for (blknum = 0; blknum < nblocks; blknum++)
backup_data_page(file, lsn, blknum, nblocks, in, out, &crc);
}
else
{
@@ -247,122 +246,10 @@ backup_data_file(const char *from_root, const char *to_root,
iter = datapagemap_iterate(&file->pagemap);
while (datapagemap_next(iter, &blknum))
{
XLogRecPtr page_lsn;
int upper_offset;
int upper_length;
int ret;
int try_checksum = 100;
bool stop_backup = false;
offset = blknum * BLCKSZ;
while(try_checksum)
{
if (offset > 0)
{
ret = fseek(in, offset, SEEK_SET);
if (ret != 0)
elog(ERROR,
"Can't seek in file offset: %llu ret:%i\n",
(long long unsigned int) offset, ret);
}
read_len = fread(&page, 1, sizeof(page), in);
header.block = blknum;
try_checksum--;
/*
* If an invalid data page was found, fallback to simple copy to ensure
* all pages in the file don't have BackupPageHeader.
*/
if (!parse_page(&page, &page_lsn,
&header.hole_offset, &header.hole_length))
{
struct stat st;
int i;
for(i=0; i<BLCKSZ && page.data[i] == 0; i++);
if (i == BLCKSZ)
{
elog(LOG, "File: %s blknum %u, empty page", file->path, blknum);
goto end_checks2;
}
stat(file->path, &st);
elog(WARNING, "PTRACK SIZE: %lu %lu pages:%lu pages:%lu i:%i", file->size, st.st_size, file->size/BLCKSZ, st.st_size/BLCKSZ, i);
if (st.st_size != file->size && blknum >= file->size/BLCKSZ-1)
{
stop_backup = true;
elog(WARNING, "File: %s blknum %u, file size has changed before backup start", file->path, blknum);
break;
}
if (st.st_size != file->size && blknum < file->size/BLCKSZ-1)
{
elog(WARNING, "File: %s blknum %u, file size has changed before backup start, it seems bad", file->path, blknum);
if (!try_checksum)
break;
}
if (try_checksum)
{
elog(WARNING, "File: %s blknum %u have wrong page header, try again", file->path, blknum);
usleep(100);
fseek(in, -sizeof(page), SEEK_CUR);
fread(&page, 1, sizeof(page), in);
continue;
}
else
elog(ERROR, "File: %s blknum %u have wrong page header.", file->path, blknum);
}
if(current.checksum_version &&
pg_checksum_page(page.data, file->segno * RELSEG_SIZE + blknum) != ((PageHeader) page.data)->pd_checksum)
{
if (try_checksum)
elog(LOG, "File: %s blknum %u have wrong checksum, try again", file->path, blknum);
else
elog(ERROR, "File: %s blknum %u have wrong checksum.", file->path, blknum);
}
else
{
try_checksum = 0;
}
}
file->read_size += read_len;
if(stop_backup)
break;
end_checks2:
upper_offset = header.hole_offset + header.hole_length;
upper_length = BLCKSZ - upper_offset;
write_buffer_real_size = sizeof(header)+header.hole_offset+upper_length;
memcpy(write_buffer, &header, sizeof(header));
if (header.hole_offset)
memcpy(write_buffer+sizeof(header), page.data, header.hole_offset);
if (upper_length)
memcpy(write_buffer+sizeof(header)+header.hole_offset, page.data + upper_offset, upper_length);
/* write data page excluding hole */
if(fwrite(write_buffer, 1, write_buffer_real_size, out) != write_buffer_real_size)
{
int errno_tmp = errno;
/* oops */
fclose(in);
fclose(out);
elog(ERROR, "cannot write at block %u of \"%s\": %s",
blknum, to_path, strerror(errno_tmp));
}
/* update CRC */
COMP_CRC32C(crc, &header, sizeof(header));
COMP_CRC32C(crc, page.data, header.hole_offset);
COMP_CRC32C(crc, page.data + upper_offset, upper_length);
file->write_size += sizeof(header) + read_len - header.hole_length;
elog(WARNING, "Iter bitmap. blknum %u, nblocks %u", blknum, nblocks);
backup_data_page(file, lsn, blknum, nblocks, in, out, &crc);
}
pg_free(iter);
/*
* If we have pagemap then file can't be a zero size.
@@ -373,10 +260,7 @@ backup_data_file(const char *from_root, const char *to_root,
file->read_size++;
}
/*
* update file permission
* FIXME: Should set permission on open?
*/
/* update file permission */
if (!check && chmod(to_path, FILE_PERMISSION) == -1)
{
int errno_tmp = errno;
@@ -413,6 +297,120 @@ backup_data_file(const char *from_root, const char *to_root,
return true;
}
/*
* Restore compressed file that was backed up partly.
*
*/
static void
restore_file_partly(const char *from_root,const char *to_root, pgFile *file)
{
FILE *in;
FILE *out;
size_t read_len = 0;
int errno_tmp;
struct stat st;
char to_path[MAXPGPATH];
char buf[8192];
size_t write_size = 0;
join_path_components(to_path, to_root, file->path + strlen(from_root) + 1);
/* open backup mode file for read */
in = fopen(file->path, "r");
if (in == NULL)
{
elog(ERROR, "cannot open backup file \"%s\": %s", file->path,
strerror(errno));
}
out = fopen(to_path, "r+");
/* stat source file to change mode of destination file */
if (fstat(fileno(in), &st) == -1)
{
fclose(in);
fclose(out);
elog(ERROR, "cannot stat \"%s\": %s", file->path,
strerror(errno));
}
if (fseek(out, 0, SEEK_END) < 0)
elog(ERROR, "cannot seek END of \"%s\": %s",
to_path, strerror(errno));
/* copy everything from backup to the end of the file */
for (;;)
{
if ((read_len = fread(buf, 1, sizeof(buf), in)) != sizeof(buf))
break;
if (fwrite(buf, 1, read_len, out) != read_len)
{
errno_tmp = errno;
/* oops */
fclose(in);
fclose(out);
elog(ERROR, "cannot write to \"%s\": %s", to_path,
strerror(errno_tmp));
}
write_size += read_len;
}
errno_tmp = errno;
if (!feof(in))
{
fclose(in);
fclose(out);
elog(ERROR, "cannot read backup mode file \"%s\": %s",
file->path, strerror(errno_tmp));
}
/* copy odd part. */
if (read_len > 0)
{
if (fwrite(buf, 1, read_len, out) != read_len)
{
errno_tmp = errno;
/* oops */
fclose(in);
fclose(out);
elog(ERROR, "cannot write to \"%s\": %s", to_path,
strerror(errno_tmp));
}
write_size += read_len;
}
// elog(LOG, "restore_file_partly(). %s write_size %lu, file->write_size %lu",
// file->path, write_size, file->write_size);
/* update file permission */
if (chmod(to_path, file->mode) == -1)
{
int errno_tmp = errno;
fclose(in);
fclose(out);
elog(ERROR, "cannot change mode of \"%s\": %s", to_path,
strerror(errno_tmp));
}
fclose(in);
fclose(out);
}
static void
restore_compressed_file(const char *from_root,
const char *to_root,
pgFile *file)
{
if (file->is_partial_copy == 0)
copy_file(from_root, to_root, file);
else if (file->is_partial_copy == 1)
restore_file_partly(from_root, to_root, file);
else
elog(ERROR, "restore_compressed_file(). Unknown is_partial_copy value %d",
file->is_partial_copy);
}
/*
* Restore files in the from_root directory to the to_root directory with
* same relative path.
@@ -429,10 +427,17 @@ restore_data_file(const char *from_root,
BackupPageHeader header;
BlockNumber blknum;
/* If the file is not a datafile, just copy it. */
if (!file->is_datafile)
{
copy_file(from_root, to_root, file);
/*
* If the file is not a datafile and not compressed file,
* just copy it.
*/
if (file->generation == -1)
copy_file(from_root, to_root, file);
else
restore_compressed_file(from_root, to_root, file);
return;
}
@@ -465,8 +470,6 @@ restore_data_file(const char *from_root,
{
size_t read_len;
DataPage page; /* used as read buffer */
int upper_offset;
int upper_length;
/* read BackupPageHeader */
read_len = fread(&header, 1, sizeof(header), in);
@@ -476,53 +479,40 @@ restore_data_file(const char *from_root,
if (read_len == 0 && feof(in))
break; /* EOF found */
else if (read_len != 0 && feof(in))
{
elog(ERROR,
"odd size page found at block %u of \"%s\"",
blknum, file->path);
}
else
{
elog(ERROR, "cannot read block %u of \"%s\": %s",
blknum, file->path, strerror(errno_tmp));
}
}
if (header.block < blknum || header.hole_offset > BLCKSZ ||
(int) header.hole_offset + (int) header.hole_length > BLCKSZ)
{
if (header.block < blknum)
elog(ERROR, "backup is broken at block %u",
blknum);
}
upper_offset = header.hole_offset + header.hole_length;
upper_length = BLCKSZ - upper_offset;
/* read lower/upper into page.data and restore hole */
memset(page.data + header.hole_offset, 0, header.hole_length);
if (fread(page.data, 1, header.hole_offset, in) != header.hole_offset ||
fread(page.data + upper_offset, 1, upper_length, in) != upper_length)
{
if (fread(page.data, 1, BLCKSZ, in) != BLCKSZ)
elog(ERROR, "cannot read block %u of \"%s\": %s",
blknum, file->path, strerror(errno));
}
/* update checksum because we are not save whole */
if(backup->checksum_version)
{
/* skip calc checksum if zero page */
bool is_zero_page = false;
if(page.page_data.pd_upper == 0)
{
int i;
for(i=0; i<BLCKSZ && page.data[i] == 0; i++);
for(i = 0; i < BLCKSZ && page.data[i] == 0; i++);
if (i == BLCKSZ)
goto skip_checksum;
is_zero_page = true;
}
((PageHeader) page.data)->pd_checksum = pg_checksum_page(page.data, file->segno * RELSEG_SIZE + header.block);
}
skip_checksum:
/* skip calc checksum if zero page */
if (!is_zero_page)
((PageHeader) page.data)->pd_checksum = pg_checksum_page(page.data, file->segno * RELSEG_SIZE + header.block);
}
/*
* Seek and write the restored page. Backup might have holes in
@@ -552,6 +542,16 @@ restore_data_file(const char *from_root,
fclose(out);
}
/* If someone's want to use this function before correct
* generation values is set, he can look up for corresponding
* .cfm file in the file_list
*/
bool
is_compressed_data_file(pgFile *file)
{
return (file->generation != -1);
}
bool
copy_file(const char *from_root, const char *to_root, pgFile *file)
{
@@ -681,6 +681,140 @@ copy_file(const char *from_root, const char *to_root, pgFile *file)
return true;
}
/*
* Save part of the file into backup.
* skip_size - size of the file in previous backup. We can skip it
* and copy just remaining part of the file
*/
bool
copy_file_partly(const char *from_root, const char *to_root,
pgFile *file, size_t skip_size)
{
char to_path[MAXPGPATH];
FILE *in;
FILE *out;
size_t read_len = 0;
int errno_tmp;
struct stat st;
char buf[8192];
/* reset size summary */
file->read_size = 0;
file->write_size = 0;
/* open backup mode file for read */
in = fopen(file->path, "r");
if (in == NULL)
{
/* maybe deleted, it's not error */
if (errno == ENOENT)
return false;
elog(ERROR, "cannot open source file \"%s\": %s", file->path,
strerror(errno));
}
/* open backup file for write */
if (check)
snprintf(to_path, lengthof(to_path), "%s/tmp", backup_path);
else
join_path_components(to_path, to_root, file->path + strlen(from_root) + 1);
out = fopen(to_path, "w");
if (out == NULL)
{
int errno_tmp = errno;
fclose(in);
elog(ERROR, "cannot open destination file \"%s\": %s",
to_path, strerror(errno_tmp));
}
/* stat source file to change mode of destination file */
if (fstat(fileno(in), &st) == -1)
{
fclose(in);
fclose(out);
elog(ERROR, "cannot stat \"%s\": %s", file->path,
strerror(errno));
}
if (fseek(in, skip_size, SEEK_SET) < 0)
elog(ERROR, "cannot seek %lu of \"%s\": %s",
skip_size, file->path, strerror(errno));
/*
* copy content
* NOTE: Now CRC is not computed for compressed files now.
*/
for (;;)
{
if ((read_len = fread(buf, 1, sizeof(buf), in)) != sizeof(buf))
break;
if (fwrite(buf, 1, read_len, out) != read_len)
{
errno_tmp = errno;
/* oops */
fclose(in);
fclose(out);
elog(ERROR, "cannot write to \"%s\": %s", to_path,
strerror(errno_tmp));
}
file->write_size += sizeof(buf);
file->read_size += sizeof(buf);
}
errno_tmp = errno;
if (!feof(in))
{
fclose(in);
fclose(out);
elog(ERROR, "cannot read backup mode file \"%s\": %s",
file->path, strerror(errno_tmp));
}
/* copy odd part. */
if (read_len > 0)
{
if (fwrite(buf, 1, read_len, out) != read_len)
{
errno_tmp = errno;
/* oops */
fclose(in);
fclose(out);
elog(ERROR, "cannot write to \"%s\": %s", to_path,
strerror(errno_tmp));
}
file->write_size += read_len;
file->read_size += read_len;
}
/* update file permission */
if (chmod(to_path, st.st_mode) == -1)
{
errno_tmp = errno;
fclose(in);
fclose(out);
elog(ERROR, "cannot change mode of \"%s\": %s", to_path,
strerror(errno_tmp));
}
/* add meta information needed for recovery */
file->is_partial_copy = 1;
// elog(LOG, "copy_file_partly(). %s file->write_size %lu", to_path, file->write_size);
fclose(in);
fclose(out);
if (check)
remove(to_path);
return true;
}
bool
calc_file(pgFile *file)
{
dir.c
+25 -8
View File
@@ -98,6 +98,8 @@ pgFileNew(const char *path, bool omit_symlink)
file->ptrack_path = NULL;
file->segno = 0;
file->path = pgut_malloc(strlen(path) + 1);
file->generation = -1;
file->is_partial_copy = 0;
strcpy(file->path, path); /* enough buffer size guaranteed */
return file;
@@ -547,17 +549,19 @@ dir_print_file_list(FILE *out, const parray *files, const char *root, const char
type = '?';
fprintf(out, "%s %c %lu %u 0%o", path, type,
(unsigned long) file->write_size,
file->crc, file->mode & (S_IRWXU | S_IRWXG | S_IRWXO));
(unsigned long) file->write_size,
file->crc, file->mode & (S_IRWXU | S_IRWXG | S_IRWXO));
if (S_ISLNK(file->mode))
fprintf(out, " %s\n", file->linked);
fprintf(out, " %s", file->linked);
else
{
char timestamp[20];
time2iso(timestamp, 20, file->mtime);
fprintf(out, " %s\n", timestamp);
fprintf(out, " %s", timestamp);
}
fprintf(out, " %d %d\n", file->generation, file->is_partial_copy);
}
}
@@ -583,6 +587,8 @@ dir_read_file_list(const char *root, const char *file_txt)
{
char path[MAXPGPATH];
char type;
int generation = -1;
int is_partial_copy = 0;
unsigned long write_size;
pg_crc32 crc;
unsigned int mode; /* bit length of mode_t depends on platforms */
@@ -590,14 +596,23 @@ dir_read_file_list(const char *root, const char *file_txt)
pgFile *file;
memset(&tm, 0, sizeof(tm));
if (sscanf(buf, "%s %c %lu %u %o %d-%d-%d %d:%d:%d",
if (sscanf(buf, "%s %c %lu %u %o %d-%d-%d %d:%d:%d %d %d",
path, &type, &write_size, &crc, &mode,
&tm.tm_year, &tm.tm_mon, &tm.tm_mday,
&tm.tm_hour, &tm.tm_min, &tm.tm_sec) != 11)
&tm.tm_hour, &tm.tm_min, &tm.tm_sec,
&generation, &is_partial_copy) != 13)
{
elog(ERROR, "invalid format found in \"%s\"",
file_txt);
/* Maybe the file_list we're trying to recovery is in old format */
if (sscanf(buf, "%s %c %lu %u %o %d-%d-%d %d:%d:%d",
path, &type, &write_size, &crc, &mode,
&tm.tm_year, &tm.tm_mon, &tm.tm_mday,
&tm.tm_hour, &tm.tm_min, &tm.tm_sec) != 11)
{
elog(ERROR, "invalid format found in \"%s\"",
file_txt);
}
}
if (type != 'f' && type != 'F' && type != 'd' && type != 'l')
{
elog(ERROR, "invalid type '%c' found in \"%s\"",
@@ -618,6 +633,8 @@ dir_read_file_list(const char *root, const char *file_txt)
file->mode = mode |
((type == 'f' || type == 'F') ? S_IFREG :
type == 'd' ? S_IFDIR : type == 'l' ? S_IFLNK : 0);
file->generation = generation;
file->is_partial_copy = is_partial_copy;
file->size = 0;
file->read_size = 0;
file->write_size = write_size;
pg_probackup.h
+34
View File
@@ -25,6 +25,10 @@
#include "storage/block.h"
#include "storage/checksum.h"
#ifndef WIN32
#include <sys/mman.h>
#endif
/* Query to fetch current transaction ID */
#define TXID_CURRENT_SQL "SELECT txid_current();"
#define TXID_CURRENT_IF_SQL "SELECT txid_snapshot_xmax(txid_current_snapshot());"
@@ -66,6 +70,10 @@ typedef struct pgFile
char *path; /* path of the file */
char *ptrack_path;
int segno; /* Segment number for ptrack */
int generation; /* Generation of compressed file.
* -1 for non-compressed files */
int is_partial_copy; /* for compressed files.
* 1 if backed up via copy_file_partly() */
volatile uint32 lock;
datapagemap_t pagemap;
} pgFile;
@@ -166,6 +174,26 @@ typedef union DataPage
char data[BLCKSZ];
} DataPage;
/*
* This struct definition mirrors one from cfs.h,
* but doesn't use atomic variables, since they are not allowed in
* frontend code.
*/
typedef struct
{
uint32 physSize;
uint32 virtSize;
uint32 usedSize;
uint32 lock;
pid_t postmasterPid;
uint64 generation;
uint64 inodes[RELSEG_SIZE];
} FileMap;
extern FileMap* cfs_mmap(int md);
extern int cfs_munmap(FileMap* map);
/*
* return pointer that exceeds the length of prefix from character string.
* ex. str="/xxx/yyy/zzz", prefix="/xxx/yyy", return="zzz".
@@ -310,8 +338,14 @@ extern bool backup_data_file(const char *from_root, const char *to_root,
pgFile *file, const XLogRecPtr *lsn);
extern void restore_data_file(const char *from_root, const char *to_root,
pgFile *file, pgBackup *backup);
extern bool is_compressed_data_file(pgFile *file);
extern bool backup_compressed_file_partially(pgFile *file,
void *arg,
size_t *skip_size);
extern bool copy_file(const char *from_root, const char *to_root,
pgFile *file);
extern bool copy_file_partly(const char *from_root, const char *to_root,
pgFile *file, size_t skip_size);
extern bool calc_file(pgFile *file);
validate.c
+5 -1
View File
@@ -325,7 +325,11 @@ pgBackupValidateFiles(void *arg)
elog(ERROR, "interrupted during validate");
/* skipped backup while differential backup */
if (file->write_size == BYTES_INVALID || !S_ISREG(file->mode))
/* NOTE We don't compute checksums for compressed data,
* so skip it too */
if (file->write_size == BYTES_INVALID
|| !S_ISREG(file->mode)
|| file->generation != -1)
continue;
/* print progress */