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));