From b127345cd5e4d1cf438831ffed22b869940536aa Mon Sep 17 00:00:00 2001 From: stalkerg Date: Fri, 25 Nov 2016 14:26:58 +0300 Subject: [PATCH] Add muti-thread validation. --- pg_probackup.c | 3 ++ restore.c | 3 -- validate.c | 81 +++++++++++++++++++++++++++++++++++++++----------- 3 files changed, 66 insertions(+), 21 deletions(-) diff --git a/pg_probackup.c b/pg_probackup.c index 694b1ed0..5d375907 100644 --- a/pg_probackup.c +++ b/pg_probackup.c @@ -178,6 +178,9 @@ main(int argc, char *argv[]) if (target_time != NULL && target_xid != NULL) elog(ERROR, "You can't specify recovery-target-time and recovery-target-xid at the same time"); + if (num_threads < 1) + num_threads = 1; + /* do actual operation */ if (pg_strcasecmp(cmd, "init") == 0) return do_init(); diff --git a/restore.c b/restore.c index 33192f76..47c3bb95 100644 --- a/restore.c +++ b/restore.c @@ -335,9 +335,6 @@ restore_database(pgBackup *backup) pgFileFree(parray_remove(files, i)); } - if (num_threads < 1) - num_threads = 1; - for (i = 0; i < parray_num(files); i++) { pgFile *file = (pgFile *) parray_get(files, i); diff --git a/validate.c b/validate.c index 50e4d004..df4a46a6 100644 --- a/validate.c +++ b/validate.c @@ -10,8 +10,17 @@ #include "pg_probackup.h" #include +#include -static bool pgBackupValidateFiles(parray *files, const char *root, bool size_only); +static void pgBackupValidateFiles(void *arg); + +typedef struct +{ + parray *files; + const char *root; + bool size_only; + bool corrupted; +} validate_files_args; /* * Validate files in the backup and update its status to OK. @@ -85,6 +94,8 @@ pgBackupValidate(pgBackup *backup, char path[MAXPGPATH]; parray *files; bool corrupted = false; + pthread_t validate_threads[num_threads]; + validate_files_args *validate_threads_args[num_threads]; time2iso(timestamp, lengthof(timestamp), backup->recovery_time); if (!for_get_timeline) @@ -102,13 +113,41 @@ pgBackupValidate(pgBackup *backup, backup->backup_mode == BACKUP_MODE_DIFF_PAGE || backup->backup_mode == BACKUP_MODE_DIFF_PTRACK) { + int i; elog(LOG, "database files..."); pgBackupGetPath(backup, base_path, lengthof(base_path), DATABASE_DIR); pgBackupGetPath(backup, path, lengthof(path), DATABASE_FILE_LIST); files = dir_read_file_list(base_path, path); - if (!pgBackupValidateFiles(files, base_path, size_only)) - corrupted = true; + + /* setup threads */ + for (i = 0; i < parray_num(files); i++) + { + pgFile *file = (pgFile *) parray_get(files, i); + __sync_lock_release(&file->lock); + } + + /* restore files into $PGDATA */ + for (i = 0; i < num_threads; i++) + { + validate_files_args *arg = pg_malloc(sizeof(validate_files_args)); + arg->files = files; + arg->root = base_path; + arg->size_only = size_only; + arg->corrupted = false; + + validate_threads_args[i] = arg; + pthread_create(&validate_threads[i], NULL, (void *(*)(void *)) pgBackupValidateFiles, arg); + } + + /* Wait theads */ + for (i = 0; i < num_threads; i++) + { + pthread_join(validate_threads[i], NULL); + if (validate_threads_args[i]->corrupted) + corrupted = true; + pg_free(validate_threads_args[i]); + } parray_walk(files, pgFileFree); parray_free(files); } @@ -140,16 +179,21 @@ get_relative_path(const char *path, const char *root) /* * Validate files in the backup with size or CRC. */ -static bool -pgBackupValidateFiles(parray *files, const char *root, bool size_only) +static void +pgBackupValidateFiles(void *arg) { int i; - for (i = 0; i < parray_num(files); i++) + validate_files_args *arguments = (validate_files_args *)arg; + + + for (i = 0; i < parray_num(arguments->files); i++) { struct stat st; - pgFile *file = (pgFile *) parray_get(files, i); + pgFile *file = (pgFile *) parray_get(arguments->files, i); + if (__sync_lock_test_and_set(&file->lock, 1) != 0) + continue; if (interrupted) elog(ERROR, "interrupted during validate"); @@ -159,8 +203,8 @@ pgBackupValidateFiles(parray *files, const char *root, bool size_only) continue; /* print progress */ - elog(LOG, "(%d/%lu) %s", i + 1, (unsigned long) parray_num(files), - get_relative_path(file->path, root)); + elog(LOG, "(%d/%lu) %s", i + 1, (unsigned long) parray_num(arguments->files), + get_relative_path(file->path, arguments->root)); /* always validate file size */ if (stat(file->path, &st) == -1) @@ -169,20 +213,22 @@ pgBackupValidateFiles(parray *files, const char *root, bool size_only) elog(WARNING, "backup file \"%s\" vanished", file->path); else elog(ERROR, "cannot stat backup file \"%s\": %s", - get_relative_path(file->path, root), strerror(errno)); - return false; + get_relative_path(file->path, arguments->root), strerror(errno)); + arguments->corrupted = true; + return; } if (file->write_size != st.st_size) { elog(WARNING, "size of backup file \"%s\" must be %lu but %lu", - get_relative_path(file->path, root), + get_relative_path(file->path, arguments->root), (unsigned long) file->write_size, (unsigned long) st.st_size); - return false; + arguments->corrupted = true; + return; } /* validate CRC too */ - if (!size_only) + if (!arguments->size_only) { pg_crc32 crc; @@ -190,11 +236,10 @@ pgBackupValidateFiles(parray *files, const char *root, bool size_only) if (crc != file->crc) { elog(WARNING, "CRC of backup file \"%s\" must be %X but %X", - get_relative_path(file->path, root), file->crc, crc); - return false; + get_relative_path(file->path, arguments->root), file->crc, crc); + arguments->corrupted = true; + return; } } } - - return true; }