rt/snapraid
1
0
Fork 0
mirror of https://github.com/amadvance/snapraid.git synced 2026-10-26 19:01:57 -04:00
Code Issues Projects Releases Packages Wiki Activity Actions

Add support for multithread data and parity reading in scrub

Browse source
This commit is contained in:
Andrea Mazzoleni 2015-12-30 11:08:20 +01:00
parent 3c00248aa6
commit 51af8931b7
4 changed files with 967 additions and 172 deletions
Download patch file Download diff file Expand all files Collapse all files

2
Makefile.am
View file

@ -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 \

396
cmdline/io.c Normal file
View file

@ -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 <http://www.gnu.org/licenses/>.
*/
#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 */
}
}
}

281
cmdline/io.h Normal file
View file

@ -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 <http://www.gnu.org/licenses/>.
*/
#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

460
cmdline/scrub.c
View file

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