From 51af8931b715f61b1cbbb570fd3a8d49fd1c0138 Mon Sep 17 00:00:00 2001 From: Andrea Mazzoleni Date: Wed, 30 Dec 2015 11:08:20 +0100 Subject: [PATCH] Add support for multithread data and parity reading in scrub --- Makefile.am | 2 + cmdline/io.c | 396 +++++++++++++++++++++++++++++++++++++++++ cmdline/io.h | 281 +++++++++++++++++++++++++++++ cmdline/scrub.c | 460 ++++++++++++++++++++++++++++++------------------ 4 files changed, 967 insertions(+), 172 deletions(-) create mode 100644 cmdline/io.c create mode 100644 cmdline/io.h diff --git a/Makefile.am b/Makefile.am index 389b54cc..d5f253b5 100644 --- a/Makefile.am +++ b/Makefile.am @@ -16,6 +16,7 @@ snapraid_SOURCES = \ raid/tag.c \ tommyds/tommy.c \ cmdline/snapraid.c \ + cmdline/io.c \ cmdline/util.c \ cmdline/stream.c \ cmdline/support.c \ @@ -70,6 +71,7 @@ noinst_HEADERS = \ tommyds/tommytree.c \ cmdline/portable.h \ cmdline/snapraid.h \ + cmdline/io.h \ cmdline/util.h \ cmdline/stream.h \ cmdline/support.h \ diff --git a/cmdline/io.c b/cmdline/io.c new file mode 100644 index 00000000..c29b58c7 --- /dev/null +++ b/cmdline/io.c @@ -0,0 +1,396 @@ +/* + * Copyright (C) 2016 Andrea Mazzoleni + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see . + */ + +#include "portable.h" + +#include "io.h" + +void io_init(struct snapraid_io* io, struct snapraid_state* state, unsigned buffer_max, + void (*data_reader)(struct snapraid_worker*, struct snapraid_task*), + struct snapraid_handle* handle_map, unsigned handle_max, + void (*parity_reader)(struct snapraid_worker*, struct snapraid_task*), + struct snapraid_parity_handle* parity_handle_map, unsigned parity_handle_max) +{ + unsigned i; + + pthread_mutex_init(&io->mutex, 0); + pthread_cond_init(&io->not_empty, 0); + pthread_cond_init(&io->not_full, 0); + + io->state = state; + + io->buffer_max = buffer_max; + for (i = 0; i < IO_MAX; ++i) { + io->buffer_map[i] = malloc_nofail_vector_align(handle_max, buffer_max, state->block_size, &io->buffer_alloc_map[i]); + if (!state->opt.skip_self) + mtest_vector(io->buffer_max, state->block_size, io->buffer_map[i]); + } + + io->worker_max = handle_max + parity_handle_max; + io->worker_map = malloc_nofail(sizeof(struct snapraid_worker) * io->worker_max); + io->worker_list = malloc_nofail(io->worker_max + 1); + + io->data_base = 0; + io->data_count = handle_max; + io->parity_base = handle_max; + io->parity_count = parity_handle_max; + + for (i = 0; i < io->worker_max; ++i) { + struct snapraid_worker* worker = &io->worker_map[i]; + + worker->io = io; + worker->fs_last = 0; + + if (i < handle_max) { + worker->handle = &handle_map[i]; + worker->parity_handle = 0; + worker->func = data_reader; + + /* data read is put in lower buffer index */ + worker->buffer_skew = 0; + } else { + worker->handle = 0; + worker->parity_handle = &parity_handle_map[i - handle_max]; + worker->func = parity_reader; + + /* parity read is put after data and computed parity */ + worker->buffer_skew = parity_handle_max; + } + } +} + +void io_done(struct snapraid_io* io) +{ + unsigned i; + + for (i = 0; i < IO_MAX; ++i) { + free(io->buffer_map[i]); + free(io->buffer_alloc_map[i]); + } + + free(io->worker_map); + free(io->worker_list); + + pthread_mutex_destroy(&io->mutex); + pthread_cond_destroy(&io->not_empty); + pthread_cond_destroy(&io->not_full); +} + +/** + * Get the next block position to process. + */ +static block_off_t io_position_next(struct snapraid_io* io) +{ + block_off_t blockcur; + + /* get the next position */ + while (io->block_next < io->block_max && !io->block_is_enabled(io->block_arg, io->block_next)) + ++io->block_next; + + blockcur = io->block_next; + + /* next block for the next call */ + ++io->block_next; + + return blockcur; +} + +/** + * Setup the next pending task for all workers. + */ +static void io_pending_next(struct snapraid_io* io, int index, block_off_t blockcur) +{ + unsigned i; + + for (i = 0; i < io->worker_max; ++i) { + struct snapraid_worker* worker = &io->worker_map[i]; + struct snapraid_task* task = &worker->task_map[index]; + + /* setup the new pending task */ + if (blockcur < io->block_max) + task->state = TASK_STATE_READY; + else + task->state = TASK_STATE_EMPTY; + + task->path[0] = 0; + if (worker->handle) + task->disk = worker->handle->disk; + else + task->disk = 0; + task->buffer = io->buffer_map[index][worker->buffer_skew + i]; + task->position = blockcur; + task->block = 0; + task->file = 0; + task->file_pos = 0; + task->read_size = 0; + task->is_timestamp_different = 0; + } +} + +/** + * Get the next task to work on for a worker. + * + * This is the synchronization point for workers with the io. + */ +static struct snapraid_task* io_worker_next(struct snapraid_worker* worker) +{ + struct snapraid_io* io = worker->io; + struct snapraid_task* task = 0; + + /* the synchronization is protected by the io mutex */ + pthread_mutex_lock(&io->mutex); + + while (1) { + unsigned index; + + /* check if the worker has to exit */ + if (io->done) + break; + + /* get the next pending task */ + index = (worker->index + 1) % IO_MAX; + + /* if the queue of pending tasks is not empty */ + if (index != io->index) { + /* get the new working task */ + worker->index = index; + task = &worker->task_map[worker->index]; + + /* notify the io that we started working */ + pthread_cond_signal(&io->not_empty); + + /* return the new task */ + break; + } + + /* otherwise wait for a not_full event */ + pthread_cond_wait(&io->not_full, &io->mutex); + } + + pthread_mutex_unlock(&io->mutex); + + return task; +} + +/** + * Get the next block position to operate on. + * + * This is the synchronization point for workers with the io. + */ +block_off_t io_next(struct snapraid_io* io, void*** buffer) +{ + block_off_t blockcur_schedule; + block_off_t blockcur_caller; + unsigned i; + + blockcur_schedule = io_position_next(io); + + /* setup the list of workers to process */ + for (i = 0; i <= io->worker_max; ++i) + io->worker_list[i] = i; + + /* the synchronization is protected by the io mutex */ + pthread_mutex_lock(&io->mutex); + + /* reset the task just used to be the next pending task */ + io_pending_next(io, io->index, blockcur_schedule); + + /* get the next task to return to the caller */ + io->index = (io->index + 1) % IO_MAX; + + /* get the position to operate at high level from one task */ + blockcur_caller = io->worker_map[0].task_map[io->index].position; + + /* set the buffer to use */ + *buffer = io->buffer_map[io->index]; + + /* signal all the workers that there is a new pending task */ + pthread_cond_broadcast(&io->not_full); + + pthread_mutex_unlock(&io->mutex); + + return blockcur_caller; +} + +static struct snapraid_task* io_task_next(struct snapraid_io* io, unsigned base, unsigned count, unsigned* pos) +{ + /* the synchronization is protected by the io mutex */ + pthread_mutex_lock(&io->mutex); + + while (1) { + unsigned char* let; + + /* search for a worker that has already finished */ + let = &io->worker_list[0]; + while (1) { + unsigned i = *let; + + /* if we are at the end */ + if (i == io->worker_max) + break; + + /* if it's in range */ + if (base <= i && i < base + count) { + struct snapraid_worker* worker; + + worker = &io->worker_map[i]; + + /* if the worker has finished this index */ + if (io->index != worker->index) { + struct snapraid_task* task; + + task = &worker->task_map[io->index]; + + pthread_mutex_unlock(&io->mutex); + + /* mark the worker as processed */ + /* setting the previous one to point at the next one */ + *let = io->worker_list[i + 1]; + + /* return the position */ + *pos = i - base; + + return task; + } + } + + /* next position to check */ + let = &io->worker_list[i + 1]; + } + + /* if not worker is ready, wait for an event */ + pthread_cond_wait(&io->not_empty, &io->mutex); + } +} + +struct snapraid_task* io_data_next(struct snapraid_io* io, unsigned* pos) +{ + return io_task_next(io, io->data_base, io->data_count, pos); +} + +struct snapraid_task* io_parity_next(struct snapraid_io* io, unsigned* pos) +{ + return io_task_next(io, io->parity_base, io->parity_count, pos); +} + +static void io_worker(struct snapraid_worker* worker, struct snapraid_task* task) +{ + /* if we reached the end */ + if (task->position >= worker->io->block_max) { + task->state = TASK_STATE_EMPTY; + } else { + worker->func(worker, task); + } +} + +static void* io_thread(void* arg) +{ + struct snapraid_worker* worker = arg; + + /* force completion of the first task */ + io_worker(worker, &worker->task_map[0]); + + while (1) { + struct snapraid_task* task; + + /* get the new task */ + task = io_worker_next(worker); + + /* if no task, it means to exit */ + if (!task) + break; + + /* nothing more to do */ + if (task->state == TASK_STATE_EMPTY) + continue; + + assert(task->state == TASK_STATE_READY); + + /* work on the assigned task */ + io_worker(worker, task); + } + + return 0; +} + +void io_start(struct snapraid_io* io, + block_off_t blockstart, block_off_t blockmax, + int (*block_is_enabled)(void* arg,block_off_t), void* blockarg) +{ + unsigned i; + + io->block_start = blockstart; + io->block_max = blockmax; + io->block_is_enabled = block_is_enabled; + io->block_arg = blockarg; + io->block_next = blockstart; + + io->done = 0; + io->index = IO_MAX - 1; + + /* setup the initial pending tasks, except the latest one, */ + /* the latest will be initialized at the fist io_next() call */ + for (i = 0; i < IO_MAX - 1; ++i) { + block_off_t blockcur = io_position_next(io); + + io_pending_next(io, i, blockcur); + } + + /* start the worker threads */ + for (i = 0; i < io->worker_max; ++i) { + struct snapraid_worker* worker = &io->worker_map[i]; + + worker->index = 0; + + if (pthread_create(&worker->thread, 0, io_thread, worker) != 0) { + /* LCOV_EXCL_START */ + log_fatal("Failed to create thread.\n"); + exit(EXIT_FAILURE); + /* LCOV_EXCL_STOP */ + } + } +} + +void io_stop(struct snapraid_io* io) +{ + unsigned i; + + pthread_mutex_lock(&io->mutex); + + /* mark that we are stopping */ + io->done = 1; + + /* signal all the threads to recognize the new state */ + pthread_cond_broadcast(&io->not_full); + + pthread_mutex_unlock(&io->mutex); + + /* wait for all threads to terminate */ + for (i = 0; i < io->worker_max; ++i) { + struct snapraid_worker* worker = &io->worker_map[i]; + void* retval; + + /* wait for thread termination */ + if (pthread_join(worker->thread, &retval) != 0) { + /* LCOV_EXCL_START */ + log_fatal("Failed to join thread.\n"); + exit(EXIT_FAILURE); + /* LCOV_EXCL_STOP */ + } + } +} + diff --git a/cmdline/io.h b/cmdline/io.h new file mode 100644 index 00000000..335bad9e --- /dev/null +++ b/cmdline/io.h @@ -0,0 +1,281 @@ +/* + * Copyright (C) 2016 Andrea Mazzoleni + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see . + */ + +#ifndef __IO_H +#define __IO_H + +#include "state.h" +#include "support.h" +#include "handle.h" +#include "parity.h" + +/** + * Number of read-ahead buffers. + */ +#define IO_MAX 4 + +/** + * State of the task. + */ +#define TASK_STATE_IOERROR_CONTINUE -4 /**< IO error. Continuation requested. */ +#define TASK_STATE_ERROR_CONTINUE -3 /**< Generic error. Continuation requested. */ +#define TASK_STATE_IOERROR -2 /**< IO error. Failure requested. */ +#define TASK_STATE_ERROR -1 /**< Generic error. Failure requested. */ +#define TASK_STATE_EMPTY 0 /**< Nothing to do. */ +#define TASK_STATE_READY 1 /**< Ready to start. */ +#define TASK_STATE_DONE 2 /**< Task completed. */ + +/** + * Task of work. + * + * This represents the minimal element of work that worker threads are + * going to be asked to do. + * + * It consists in reading a block of data from a disk. + * + * Note that the disk to use is defined implicitly in the worker thread. + */ +struct snapraid_task { + int state; /**< State of the task. One of the TASK_STATE_*. */ + char path[PATH_MAX]; /**< Path of the file. */ + struct snapraid_disk* disk; /**< Disk of the file. */ + unsigned char* buffer; /**< Where to read the data. */ + block_off_t position; /**< Parity position to read. */ + + /** + * Result of the task. + */ + struct snapraid_block* block; + struct snapraid_file* file; + block_off_t file_pos; + int read_size; /**< Size of the data read. */ + int is_timestamp_different; /**< Report if file has a changed timestamp. */ +}; + +/** + * Worker for tasks. + * + * This represents a worker thread designated to read data + * from a specific disk. + */ +struct snapraid_worker { + pthread_t thread; + + struct snapraid_io* io; /**< Parent pointer. */ + + void (*func)(struct snapraid_worker*, struct snapraid_task*); + + /** + * Handle to data or parity. + * + * Only one of the two is valid, the other is 0. + */ + struct snapraid_handle* handle; /**< Handle at the file on the disk. */ + struct snapraid_parity_handle* parity_handle; /**< Handle at the parity on the disk. */ + + /** + * Vector of tasks. + * + * It's a ring of tasks reused cycle after cycle. + */ + struct snapraid_task task_map[IO_MAX]; + + /** + * The task in progress by the worker thread. + * + * It's an index inside in the ::task_map vector. + */ + unsigned index; + + /** + * Which buffer base index should be used for destination. + */ + unsigned buffer_skew; + + /** + * Last chunk we accessed. + * + * It's used to optimize access of sequential blocks. + * We cannot use the one in ::disk because it's used + * by the main thread. + */ + struct snapraid_chunk* fs_last; +}; + +/** + * Reader. + * + * This represents the pool of worker threads dedicated to read + * data from the disks. + */ +struct snapraid_io { + /** + * Mutex used to protect the synchronization + * between the io and the workers. + */ + pthread_mutex_t mutex; + + /** + * Condition for (not) empty queue of tasks. + * + * The io waits on this condition when the queue is empty, + * and it's signaled by workers. + */ + pthread_cond_t not_empty; + + /** + * Condition for (not) full queue of tasks. + * + * The workers wait on this condition when the queue is full, + * and it's signaled by the io. + */ + pthread_cond_t not_full; + + struct snapraid_state* state; + + /** + * Base position for workers. + * + * It's the index in the ::worker_map[]. + */ + unsigned data_base; + unsigned data_count; + unsigned parity_base; + unsigned parity_count; + + /** + * Callbacks for workers. + */ + void (*data_reader)(struct snapraid_worker*, struct snapraid_task*); + void (*parity_reader)(struct snapraid_worker*, struct snapraid_task*); + + /** + * Blocks mapping. + * + * This info is used to obtain the sequence of block + * positions to process. + */ + block_off_t block_start; + block_off_t block_max; + block_off_t block_next; + int (*block_is_enabled)(void* arg,block_off_t); + void* block_arg; + + /** + * Buffers for data. + * + * A pool of buffers used to store the data read. + */ + unsigned buffer_max; /**< Number of buffers. */ + void* buffer_alloc_map[IO_MAX]; /**< Allocation map for buffers. */ + void** buffer_map[IO_MAX]; /**< Buffers for data. */ + + /** + * Workers. + * + * A vector of workers, each one representing a different thread. + */ + unsigned worker_max; /**< Number of workers. */ + struct snapraid_worker* worker_map; /**< Vector of workers. */ + + /** + * List of not yet processed workers. + * + * The list has ::worker_max + 1 elements. Each element contains + * the number of the worker to process. + * + * At initialization the list is filled with [0..worker_max]. + * To get the next element to process we use i = list[i + 1]. + * The end is when i == worker_max. + */ + unsigned char* worker_list; + + /** + * Exit condition for all threads. + */ + int done; + + /** + * The task currently used by the caller. + * + * It's a rolling counter, when reaching IO_MAX + * it goes again to 0. + */ + unsigned index; +}; + +/** + * Initialize the InputOutput workers. + * + * \param buffer_max The number of data/parity buffers to allocate. + */ +void io_init(struct snapraid_io* io, struct snapraid_state* state, + unsigned buffer_max, + void (*data_reader)(struct snapraid_worker*, struct snapraid_task*), + struct snapraid_handle* handle_map, unsigned handle_max, + void (*parity_reader)(struct snapraid_worker*, struct snapraid_task*), + struct snapraid_parity_handle* parity_handle_map, unsigned parity_handle_max); + +/** + * Deinitialize te InputOutput workers. + */ +void io_done(struct snapraid_io* io); + +/** + * Start all the worker threads. + */ +void io_start(struct snapraid_io* io, + block_off_t blockstart, block_off_t blockmax, + int (*block_is_enabled)(void* arg,block_off_t), void* blockarg); + +/** + * Stop all the worker threads. + */ +void io_stop(struct snapraid_io* io); + +/** + * Get the next parity position to process. + * + * \param io InputOutput context. + * \param buffer The data buffers to use for this position. + * \return The parity position. + */ +block_off_t io_next(struct snapraid_io* io, void*** buffer); + +/** + * Get the next data block to process. + * + * It must be called exactly ::handle_max times. + * + * \param io InputOutput context. + * \param pos The position of the data block in the ::handle_map vector. + * \return The completed task. + */ +struct snapraid_task* io_data_next(struct snapraid_io* io, unsigned* pos); + +/** + * Get the next parity block to process. + * + * It must be called exactly ::parity_handle_max times. + * + * \param io InputOutput context. + * \param pos The position of the parity block in the ::parity_handle_map vector. + * \return The completed task. + */ +struct snapraid_task* io_parity_next(struct snapraid_io* io, unsigned* pos); + +#endif diff --git a/cmdline/scrub.c b/cmdline/scrub.c index fe10cdd2..0e06d5f2 100644 --- a/cmdline/scrub.c +++ b/cmdline/scrub.c @@ -22,6 +22,7 @@ #include "state.h" #include "parity.h" #include "handle.h" +#include "io.h" #include "raid/raid.h" /****************************************************************************/ @@ -49,8 +50,9 @@ struct snapraid_plan { /** * Check if we have to process the specified block index ::i. */ -static int block_is_enabled(struct snapraid_plan* plan, block_off_t i) +static int block_is_enabled(void* void_plan, block_off_t i) { + struct snapraid_plan* plan = void_plan; time_t blocktime; snapraid_info info; @@ -100,17 +102,179 @@ static int block_is_enabled(struct snapraid_plan* plan, block_off_t i) return 1; } -static int state_scrub_process(struct snapraid_state* state, struct snapraid_parity_handle** parity, block_off_t blockstart, block_off_t blockmax, struct snapraid_plan* plan, time_t now) +static void scrub_data_reader(struct snapraid_worker* worker, struct snapraid_task* task) { + struct snapraid_io* io = worker->io; + struct snapraid_state* state = io->state; + struct snapraid_handle* handle = worker->handle; + struct snapraid_disk* disk = handle->disk; + block_off_t blockcur = task->position; + unsigned char* buffer = task->buffer; + int ret; + + /* if the disk position is not used */ + if (!disk) { + /* use an empty block */ + memset(buffer, 0, state->block_size); + task->state = TASK_STATE_DONE; + return; + } + + /* get the block */ + task->block = fs_par2block_get_ts(disk, &worker->fs_last, blockcur); + + /* if the block is not used */ + if (!block_has_file(task->block)) { + /* use an empty block */ + memset(buffer, 0, state->block_size); + task->state = TASK_STATE_DONE; + return; + } + + /* get the file of this block */ + task->file = fs_par2file_get_ts(disk, &worker->fs_last, blockcur, &task->file_pos); + + /* if the file is different than the current one, close it */ + if (handle->file != 0 && handle->file != task->file) { + /* keep a pointer at the file we are going to close for error reporting */ + struct snapraid_file* report = handle->file; + ret = handle_close(handle); + if (ret == -1) { + /* LCOV_EXCL_START */ + /* This one is really an unexpected error, because we are only reading */ + /* and closing a descriptor should never fail */ + if (errno == EIO) { + log_tag("error:%u:%s:%s: Close EIO error. %s\n", blockcur, disk->name, esc(report->sub), strerror(errno)); + log_fatal("DANGER! Unexpected input/output close error in a data disk, it isn't possible to scrub.\n"); + log_fatal("Ensure that disk '%s' is sane and that file '%s' can be accessed.\n", disk->dir, handle->path); + log_fatal("Stopping at block %u\n", blockcur); + task->state = TASK_STATE_IOERROR; + return; + } + + log_tag("error:%u:%s:%s: Close error. %s\n", blockcur, disk->name, esc(report->sub), strerror(errno)); + log_fatal("WARNING! Unexpected close error in a data disk, it isn't possible to scrub.\n"); + log_fatal("Ensure that file '%s' can be accessed.\n", handle->path); + log_fatal("Stopping at block %u\n", blockcur); + task->state = TASK_STATE_ERROR; + return; + /* LCOV_EXCL_STOP */ + } + } + + ret = handle_open(handle, task->file, state->file_mode, log_error, 0); + if (ret == -1) { + if (errno == EIO) { + /* LCOV_EXCL_START */ + log_tag("error:%u:%s:%s: Open EIO error. %s\n", blockcur, disk->name, esc(task->file->sub), strerror(errno)); + log_fatal("DANGER! Unexpected input/output open error in a data disk, it isn't possible to scrub.\n"); + log_fatal("Ensure that disk '%s' is sane and that file '%s' can be accessed.\n", disk->dir, handle->path); + log_fatal("Stopping at block %u\n", blockcur); + task->state = TASK_STATE_IOERROR; + return; + /* LCOV_EXCL_STOP */ + } + + log_tag("error:%u:%s:%s: Open error. %s\n", blockcur, disk->name, esc(task->file->sub), strerror(errno)); + task->state = TASK_STATE_ERROR_CONTINUE; + return; + } + + /* check if the file is changed */ + if (handle->st.st_size != task->file->size + || handle->st.st_mtime != task->file->mtime_sec + || STAT_NSEC(&handle->st) != task->file->mtime_nsec + /* don't check the inode to support filesystem without persistent inodes */ + ) { + /* report that the block and the file are not synced */ + task->is_timestamp_different = 1; + /* follow */ + } + + /* note that we intentionally don't abort if the file has different attributes */ + /* from the last sync, as we are expected to return errors if running */ + /* in an unsynced array. This is just like the check command. */ + + task->read_size = handle_read(handle, task->file_pos, buffer, state->block_size, log_error, 0); + if (task->read_size == -1) { + if (errno == EIO) { + log_tag("error:%u:%s:%s: Read EIO error at position %u. %s\n", blockcur, disk->name, esc(task->file->sub), task->file_pos, strerror(errno)); +#if 0 + if (io_error >= state->opt.io_error_limit) { + /* LCOV_EXCL_START */ + log_fatal("DANGER! Too many input/output read error in a data disk, it isn't possible to scrub.\n"); + log_fatal("Ensure that disk '%s' is sane and that file '%s' can be accessed.\n", disk->dir, handle->path); + log_fatal("Stopping at block %u\n", blockcur); + task->state = TASK_STATE_IOERROR; + return; + /* LCOV_EXCL_STOP */ + } +#endif + log_error("Input/Output error in file '%s' at position '%u'\n", handle->path, task->file_pos); + task->state = TASK_STATE_IOERROR_CONTINUE; + return; + } + + log_tag("error:%u:%s:%s: Read error at position %u. %s\n", blockcur, disk->name, esc(task->file->sub), task->file_pos, strerror(errno)); + task->state = TASK_STATE_ERROR_CONTINUE; + return; + } + + /* store the path of the opened file */ + pathcpy(task->path, sizeof(task->path), handle->path); + + task->state = TASK_STATE_DONE; +} + +static void scrub_parity_reader(struct snapraid_worker* worker, struct snapraid_task* task) +{ + struct snapraid_io* io = worker->io; + struct snapraid_state* state = io->state; + struct snapraid_parity_handle* parity_handle = worker->parity_handle; + unsigned level = parity_handle->level; + block_off_t blockcur = task->position; + unsigned char* buffer = task->buffer; + int ret; + + /* read the parity */ + ret = parity_read(parity_handle, blockcur, buffer, state->block_size, log_error); + if (ret == -1) { + if (errno == EIO) { + log_tag("parity_error:%u:%s: Read EIO error. %s\n", blockcur, lev_config_name(level), strerror(errno)); +#if 0 + if (io_error >= state->opt.io_error_limit) { + /* LCOV_EXCL_START */ + log_fatal("DANGER! Too many input/output read error in the %s disk, it isn't possible to scrub.\n", lev_name(level)); + log_fatal("Ensure that disk '%s' is sane and can be read.\n", lev_config_name(level)); + log_fatal("Stopping at block %u\n", blockcur); + task->state = TASK_STATE_IOERROR; + return; + /* LCOV_EXCL_STOP */ + } +#endif + + log_error("Input/Output error in parity '%s' at position '%u'\n", lev_config_name(level), blockcur); + task->state = TASK_STATE_IOERROR_CONTINUE; + return; + } + + log_tag("parity_error:%u:%s: Read error. %s\n", blockcur, lev_config_name(level), strerror(errno)); + task->state = TASK_STATE_ERROR_CONTINUE; + return; + } + + task->state = TASK_STATE_DONE; +} + +static int state_scrub_process(struct snapraid_state* state, struct snapraid_parity_handle* parity_handle, block_off_t blockstart, block_off_t blockmax, struct snapraid_plan* plan, time_t now) +{ + struct snapraid_io io; struct snapraid_handle* handle; void* rehandle_alloc; struct snapraid_rehash* rehandle; unsigned diskmax; block_off_t blockcur; - block_off_t blocknext; unsigned j; - void* buffer_alloc; - void** buffer; unsigned buffermax; data_off_t countsize; block_off_t countpos; @@ -133,9 +297,8 @@ static int state_scrub_process(struct snapraid_state* state, struct snapraid_par /* we need disk + 2 for each parity level buffers */ buffermax = diskmax + state->level * 2; - buffer = malloc_nofail_vector_align(diskmax, buffermax, state->block_size, &buffer_alloc); - if (!state->opt.skip_self) - mtest_vector(buffermax, state->block_size, buffer); + /* initialize the io threads */ + io_init(&io, state, buffermax, scrub_data_reader, handle, diskmax, scrub_parity_reader, parity_handle, state->level); error = 0; silent_error = 0; @@ -147,7 +310,6 @@ static int state_scrub_process(struct snapraid_state* state, struct snapraid_par for (blockcur = blockstart; blockcur < blockmax; ++blockcur) { if (!block_is_enabled(plan, blockcur)) continue; - ++countmax; } @@ -164,27 +326,26 @@ static int state_scrub_process(struct snapraid_state* state, struct snapraid_par countsize = 0; countpos = 0; plan->countlast = 0; - state_progress_begin(state, blockstart, blockmax, countmax); - blocknext = blockstart; + /* start all the worker threads */ + io_start(&io, blockstart, blockmax, &block_is_enabled, plan); + + state_progress_begin(state, blockstart, blockmax, countmax); while (1) { + unsigned char* buffer_recov[LEV_MAX]; snapraid_info info; int error_on_this_block; int silent_error_on_this_block; int io_error_on_this_block; int block_is_unsynced; int rehash; + void** buffer; /* go to the next block */ - blockcur = blocknext; + blockcur = io_next(&io, &buffer); if (blockcur >= blockmax) break; - /* find the next block */ - ++blocknext; - while (blocknext < blockmax && !block_is_enabled(plan, blocknext)) - ++blocknext; - /* one more block processed for autosave */ ++autosavedone; --autosavemissing; @@ -206,38 +367,42 @@ static int state_scrub_process(struct snapraid_state* state, struct snapraid_par /* for each disk, process the block */ for (j = 0; j < diskmax; ++j) { + struct snapraid_task* task; int read_size; unsigned char hash[HASH_SIZE]; struct snapraid_block* block; int file_is_unsynced; - struct snapraid_disk* disk = handle[j].disk; + struct snapraid_disk* disk; struct snapraid_file* file; block_off_t file_pos; + unsigned diskcur; /* if the file on this disk is synced */ /* if not, silent errors are assumed as expected error */ file_is_unsynced = 0; + /* until now is CPU */ + state_usage_cpu(state); + + /* get the next task */ + task = io_data_next(&io, &diskcur); + + /* get the task results */ + disk = task->disk; + block = task->block; + file = task->file; + file_pos = task->file_pos; + read_size = task->read_size; + /* by default no rehash in case of "continue" */ - rehandle[j].block = 0; + rehandle[diskcur].block = 0; /* if the disk position is not used */ - if (!disk) { - /* use an empty block */ - memset(buffer[j], 0, state->block_size); + if (!disk) continue; - } - /* if the block is not used */ - block = fs_par2block_get(disk, blockcur); - if (!block_has_file(block)) { - /* use an empty block */ - memset(buffer[j], 0, state->block_size); - continue; - } - - /* get the file of this block */ - file = fs_par2file_get(disk, blockcur, &file_pos); + /* until now is disk */ + state_usage_disk(state, disk); /* if the block is unsynced, errors are expected */ if (block_has_invalid_parity(block)) { @@ -247,112 +412,55 @@ static int state_scrub_process(struct snapraid_state* state, struct snapraid_par /* follow */ } - /* until now is CPU */ - state_usage_cpu(state); - - /* if the file is different than the current one, close it */ - if (handle[j].file != 0 && handle[j].file != file) { - /* keep a pointer at the file we are going to close for error reporting */ - struct snapraid_file* report = handle[j].file; - ret = handle_close(&handle[j]); - if (ret == -1) { - /* LCOV_EXCL_START */ - /* This one is really an unexpected error, because we are only reading */ - /* and closing a descriptor should never fail */ - if (errno == EIO) { - log_tag("error:%u:%s:%s: Close EIO error. %s\n", blockcur, disk->name, esc(report->sub), strerror(errno)); - log_fatal("DANGER! Unexpected input/output close error in a data disk, it isn't possible to scrub.\n"); - log_fatal("Ensure that disk '%s' is sane and that file '%s' can be accessed.\n", disk->dir, handle[j].path); - log_fatal("Stopping at block %u\n", blockcur); - ++io_error; - goto bail; - } - - log_tag("error:%u:%s:%s: Close error. %s\n", blockcur, disk->name, esc(report->sub), strerror(errno)); - log_fatal("WARNING! Unexpected close error in a data disk, it isn't possible to scrub.\n"); - log_fatal("Ensure that file '%s' can be accessed.\n", handle[j].path); - log_fatal("Stopping at block %u\n", blockcur); - ++error; - goto bail; - /* LCOV_EXCL_STOP */ - } - } - - ret = handle_open(&handle[j], file, state->file_mode, log_error, 0); - if (ret == -1) { - if (errno == EIO) { - /* LCOV_EXCL_START */ - log_tag("error:%u:%s:%s: Open EIO error. %s\n", blockcur, disk->name, esc(file->sub), strerror(errno)); - log_fatal("DANGER! Unexpected input/output open error in a data disk, it isn't possible to scrub.\n"); - log_fatal("Ensure that disk '%s' is sane and that file '%s' can be accessed.\n", disk->dir, handle[j].path); - log_fatal("Stopping at block %u\n", blockcur); - ++io_error; - goto bail; - /* LCOV_EXCL_STOP */ - } - - log_tag("error:%u:%s:%s: Open error. %s\n", blockcur, disk->name, esc(file->sub), strerror(errno)); - ++error; - error_on_this_block = 1; + /* if the block is not used */ + if (!block_has_file(block)) continue; - } - /* check if the file is changed */ - if (handle[j].st.st_size != file->size - || handle[j].st.st_mtime != file->mtime_sec - || STAT_NSEC(&handle[j].st) != file->mtime_nsec - /* don't check the inode to support file-system without persistent inodes */ - ) { + /* if the block is unsynced, errors are expected */ + if (task->is_timestamp_different) { /* report that the block and the file are not synced */ block_is_unsynced = 1; file_is_unsynced = 1; /* follow */ } - /* note that we intentionally don't abort if the file has different attributes */ - /* from the last sync, as we are expected to return errors if running */ - /* in an unsynced array. This is just like the check command. */ - - read_size = handle_read(&handle[j], file_pos, buffer[j], state->block_size, log_error, 0); - if (read_size == -1) { - if (errno == EIO) { - log_tag("error:%u:%s:%s: Read EIO error at position %u. %s\n", blockcur, disk->name, esc(file->sub), file_pos, strerror(errno)); - if (io_error >= state->opt.io_error_limit) { - /* LCOV_EXCL_START */ - log_fatal("DANGER! Too many input/output read error in a data disk, it isn't possible to scrub.\n"); - log_fatal("Ensure that disk '%s' is sane and that file '%s' can be accessed.\n", disk->dir, handle[j].path); - log_fatal("Stopping at block %u\n", blockcur); - ++io_error; - goto bail; - /* LCOV_EXCL_STOP */ - } - - log_error("Input/Output error in file '%s' at position '%u'\n", handle[j].path, file_pos); - ++io_error; - io_error_on_this_block = 1; - continue; - } - - log_tag("error:%u:%s:%s: Read error at position %u. %s\n", blockcur, disk->name, esc(file->sub), file_pos, strerror(errno)); + /* handle error conditions */ + if (task->state == TASK_STATE_IOERROR) { + ++io_error; + goto bail; + } + if (task->state == TASK_STATE_ERROR) { + ++error; + goto bail; + } + if (task->state == TASK_STATE_ERROR_CONTINUE) { ++error; error_on_this_block = 1; continue; } - - /* until now is disk */ - state_usage_disk(state, disk); + if (task->state == TASK_STATE_IOERROR_CONTINUE) { + ++io_error; + io_error_on_this_block = 1; + continue; + } + if (task->state != TASK_STATE_DONE) { + /* LCOV_EXCL_START */ + log_fatal("Internal inconsistency in task state\n"); + os_abort(); + /* LCOV_EXCL_STOP */ + } countsize += read_size; /* now compute the hash */ if (rehash) { - memhash(state->prevhash, state->prevhashseed, hash, buffer[j], read_size); + memhash(state->prevhash, state->prevhashseed, hash, buffer[diskcur], read_size); /* compute the new hash, and store it */ - rehandle[j].block = block; - memhash(state->hash, state->hashseed, rehandle[j].hash, buffer[j], read_size); + rehandle[diskcur].block = block; + memhash(state->hash, state->hashseed, rehandle[diskcur].hash, buffer[diskcur], read_size); } else { - memhash(state->hash, state->hashseed, hash, buffer[j], read_size); + memhash(state->hash, state->hashseed, hash, buffer[diskcur], read_size); } if (block_has_updated_hash(block)) { @@ -367,7 +475,7 @@ static int state_scrub_process(struct snapraid_state* state, struct snapraid_par ++error; error_on_this_block = 1; } else { - log_error("Data error in file '%s' at position '%u', diff bits %u\n", handle[j].path, file_pos, diff); + log_error("Data error in file '%s' at position '%u', diff bits %u\n", task->path, file_pos, diff); ++silent_error; silent_error_on_this_block = 1; } @@ -376,52 +484,60 @@ static int state_scrub_process(struct snapraid_state* state, struct snapraid_par } } + /* buffers for parity read and not computed */ + for (l = 0; l < state->level; ++l) + buffer_recov[l] = buffer[diskmax + state->level + l]; + for (; l < LEV_MAX; ++l) + buffer_recov[l] = 0; + + /* until now is CPU */ + state_usage_cpu(state); + + /* read the parity */ + for (l = 0; l < state->level; ++l) { + struct snapraid_task* task; + unsigned levcur; + + task = io_parity_next(&io, &levcur); + + /* until now is parity */ + state_usage_parity(state, levcur); + + /* handle error conditions */ + if (task->state == TASK_STATE_IOERROR) { + ++io_error; + goto bail; + } + if (task->state == TASK_STATE_ERROR) { + ++error; + goto bail; + } + if (task->state == TASK_STATE_ERROR_CONTINUE) { + ++error; + error_on_this_block = 1; + + /* if continuing on error, clear the missing buffer */ + buffer_recov[levcur] = 0; + continue; + } + if (task->state == TASK_STATE_IOERROR_CONTINUE) { + ++io_error; + io_error_on_this_block = 1; + + /* if continuing on error, clear the missing buffer */ + buffer_recov[levcur] = 0; + continue; + } + if (task->state != TASK_STATE_DONE) { + /* LCOV_EXCL_START */ + log_fatal("Internal inconsistency in task state\n"); + os_abort(); + /* LCOV_EXCL_STOP */ + } + } + /* if we have read all the data required and it's correct, proceed with the parity check */ if (!error_on_this_block && !silent_error_on_this_block && !io_error_on_this_block) { - unsigned char* buffer_recov[LEV_MAX]; - - /* until now is CPU */ - state_usage_cpu(state); - - /* buffers for parity read and not computed */ - for (l = 0; l < state->level; ++l) - buffer_recov[l] = buffer[diskmax + state->level + l]; - for (; l < LEV_MAX; ++l) - buffer_recov[l] = 0; - - /* read the parity */ - for (l = 0; l < state->level; ++l) { - ret = parity_read(parity[l], blockcur, buffer_recov[l], state->block_size, log_error); - if (ret == -1) { - buffer_recov[l] = 0; - - if (errno == EIO) { - log_tag("parity_error:%u:%s: Read EIO error. %s\n", blockcur, lev_config_name(l), strerror(errno)); - if (io_error >= state->opt.io_error_limit) { - /* LCOV_EXCL_START */ - log_fatal("DANGER! Too many input/output read error in the %s disk, it isn't possible to scrub.\n", lev_name(l)); - log_fatal("Ensure that disk '%s' is sane and can be read.\n", lev_config_name(l)); - log_fatal("Stopping at block %u\n", blockcur); - ++io_error; - goto bail; - /* LCOV_EXCL_STOP */ - } - - log_error("Input/Output error in parity '%s' at position '%u'\n", lev_config_name(l), blockcur); - ++io_error; - io_error_on_this_block = 1; - continue; - } - - log_tag("parity_error:%u:%s: Read error. %s\n", blockcur, lev_config_name(l), strerror(errno)); - ++error; - error_on_this_block = 1; - continue; - } - - /* until now is parity */ - state_usage_parity(state, l); - } /* compute the parity */ raid_gen(diskmax, state->level, state->block_size, buffer); @@ -538,6 +654,9 @@ static int state_scrub_process(struct snapraid_state* state, struct snapraid_par log_flush(); bail: + /* stop all the worker threads */ + io_stop(&io); + for (j = 0; j < diskmax; ++j) { struct snapraid_file* file = handle[j].file; struct snapraid_disk* disk = handle[j].disk; @@ -553,9 +672,8 @@ bail: } free(handle); - free(buffer_alloc); - free(buffer); free(rehandle_alloc); + io_done(&io); if (state->opt.expect_recoverable) { if (error + silent_error + io_error == 0) @@ -589,8 +707,7 @@ int state_scrub(struct snapraid_state* state, int plan, int olderthan) block_off_t count; time_t recentlimit; int ret; - struct snapraid_parity_handle parity[LEV_MAX]; - struct snapraid_parity_handle* parity_ptr[LEV_MAX]; + struct snapraid_parity_handle parity_handle[LEV_MAX]; struct snapraid_plan ps; time_t* timemap; unsigned error; @@ -722,8 +839,7 @@ int state_scrub(struct snapraid_state* state, int plan, int olderthan) /* open the file for reading */ for (l = 0; l < state->level; ++l) { - parity_ptr[l] = &parity[l]; - ret = parity_open(parity_ptr[l], l, state->parity[l].path, state->file_mode); + ret = parity_open(&parity_handle[l], l, state->parity[l].path, state->file_mode); if (ret == -1) { /* LCOV_EXCL_START */ log_fatal("WARNING! Without an accessible %s file, it isn't possible to scrub.\n", lev_name(l)); @@ -736,14 +852,14 @@ int state_scrub(struct snapraid_state* state, int plan, int olderthan) error = 0; - ret = state_scrub_process(state, parity_ptr, 0, blockmax, &ps, now); + ret = state_scrub_process(state, parity_handle, 0, blockmax, &ps, now); if (ret == -1) { ++error; /* continue, as we are already exiting */ } for (l = 0; l < state->level; ++l) { - ret = parity_close(parity_ptr[l]); + ret = parity_close(&parity_handle[l]); if (ret == -1) { /* LCOV_EXCL_START */ log_fatal("DANGER! Unexpected close error in %s disk.\n", lev_name(l));