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

Add support for multithread parity writing in sync

Browse source
This commit is contained in:
Andrea Mazzoleni 2016-01-04 11:28:34 +01:00
parent e69e8640ce
commit a3121c31c5
4 changed files with 412 additions and 64 deletions
Download patch file Download diff file Expand all files Collapse all files

275
cmdline/io.c
View file

@ -23,6 +23,7 @@ void io_init(struct snapraid_io* io, struct snapraid_state* state, unsigned buff
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*),
void (*parity_writer)(struct snapraid_worker*, struct snapraid_task*),
struct snapraid_parity_handle* parity_handle_map, unsigned parity_handle_max)
{
unsigned i;
@ -30,6 +31,8 @@ void io_init(struct snapraid_io* io, struct snapraid_state* state, unsigned buff
pthread_mutex_init(&io->mutex, 0);
pthread_cond_init(&io->read_done, 0);
pthread_cond_init(&io->read_sched, 0);
pthread_cond_init(&io->write_done, 0);
pthread_cond_init(&io->write_sched, 0);
io->state = state;
@ -40,9 +43,18 @@ void io_init(struct snapraid_io* io, struct snapraid_state* state, unsigned buff
mtest_vector(io->buffer_max, state->block_size, io->buffer_map[i]);
}
io->reader_max = handle_max + parity_handle_max;
if (parity_writer) {
io->reader_max = handle_max;
io->writer_max = parity_handle_max;
} else {
io->reader_max = handle_max + parity_handle_max;
io->writer_max = 0;
}
io->reader_map = malloc_nofail(sizeof(struct snapraid_worker) * io->reader_max);
io->reader_list = malloc_nofail(io->reader_max + 1);
io->writer_map = malloc_nofail(sizeof(struct snapraid_worker) * io->writer_max);
io->writer_list = malloc_nofail(io->writer_max + 1);
io->data_base = 0;
io->data_count = handle_max;
@ -72,6 +84,20 @@ void io_init(struct snapraid_io* io, struct snapraid_state* state, unsigned buff
worker->buffer_skew = parity_handle_max;
}
}
for (i = 0; i < io->writer_max; ++i) {
struct snapraid_worker* worker = &io->writer_map[i];
worker->io = io;
/* it's a parity write */
worker->handle = 0;
worker->parity_handle = &parity_handle_map[i];
worker->func = parity_writer;
/* parity to write is put after data */
worker->buffer_skew = handle_max;
}
}
void io_done(struct snapraid_io* io)
@ -85,10 +111,14 @@ void io_done(struct snapraid_io* io)
free(io->reader_map);
free(io->reader_list);
free(io->writer_map);
free(io->writer_list);
pthread_mutex_destroy(&io->mutex);
pthread_cond_destroy(&io->read_done);
pthread_cond_destroy(&io->read_sched);
pthread_cond_destroy(&io->write_done);
pthread_cond_destroy(&io->write_sched);
}
/**
@ -142,6 +172,57 @@ static void io_reader_sched(struct snapraid_io* io, int index, block_off_t block
}
}
/**
* Setup the next pending task for all writers.
*/
static void io_writer_sched(struct snapraid_io* io, int index, block_off_t blockcur)
{
unsigned i;
for (i = 0; i < io->writer_max; ++i) {
struct snapraid_worker* worker = &io->writer_map[i];
struct snapraid_task* task = &worker->task_map[index];
/* setup the new pending task */
task->state = TASK_STATE_READY;
task->path[0] = 0;
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;
}
}
/**
* Setup an empty next pending task for all writers.
*/
static void io_writer_sched_empty(struct snapraid_io* io, int index, block_off_t blockcur)
{
unsigned i;
for (i = 0; i < io->writer_max; ++i) {
struct snapraid_worker* worker = &io->writer_map[i];
struct snapraid_task* task = &worker->task_map[index];
/* setup the new pending task */
task->state = TASK_STATE_EMPTY;
task->path[0] = 0;
task->disk = 0;
task->buffer = 0;
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 reader.
*
@ -188,6 +269,52 @@ static struct snapraid_task* io_reader_step(struct snapraid_worker* worker)
return task;
}
/**
* Get the next task to work on for a writer.
*
* This is the synchronization point for workers with the io.
*/
static struct snapraid_task* io_writer_step(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;
/* get the next pending task */
index = (worker->index + 1) % IO_MAX;
/* if the queue of pending tasks is not empty */
if (index != io->writer_index) {
/* get the new working task */
worker->index = index;
task = &worker->task_map[worker->index];
/* notify the io that a new writer is complete */
pthread_cond_signal(&io->write_done);
/* return the new task */
break;
}
/* check if the worker has to exit */
/* but only if there is no work to do */
if (io->done)
break;
/* otherwise wait for a write_sched event */
pthread_cond_wait(&io->write_sched, &io->mutex);
}
pthread_mutex_unlock(&io->mutex);
return task;
}
/**
* Get the next block position to operate on.
*
@ -232,6 +359,39 @@ block_off_t io_read_next(struct snapraid_io* io, void*** buffer)
return blockcur_caller;
}
void io_write_next(struct snapraid_io* io, unsigned blockcur, int skip)
{
unsigned i;
/* ensure that all parity was written */
assert(io->writer_list[0] == io->writer_max);
/* setup the list of workers to process */
for (i = 0; i <= io->writer_max; ++i)
io->writer_list[i] = i;
/* the synchronization is protected by the io mutex */
pthread_mutex_lock(&io->mutex);
if (skip) {
/* skip the next write */
io_writer_sched_empty(io, io->writer_index, blockcur);
} else {
/* schedule the next write */
io_writer_sched(io, io->writer_index, blockcur);
}
assert(io->writer_index == io->reader_index);
/* get the next task to return to the caller */
io->writer_index = (io->writer_index + 1) % IO_MAX;
/* signal all the workers that there is a new pending task */
pthread_cond_broadcast(&io->write_sched);
pthread_mutex_unlock(&io->mutex);
}
static struct snapraid_task* io_task_read(struct snapraid_io* io, unsigned base, unsigned count, unsigned* pos)
{
/* the synchronization is protected by the io mutex */
@ -299,6 +459,60 @@ struct snapraid_task* io_parity_read(struct snapraid_io* io, unsigned* pos)
return io_task_read(io, io->parity_base, io->parity_count, pos);
}
void io_parity_write(struct snapraid_io* io, unsigned* pos)
{
/* the synchronization is protected by the io mutex */
pthread_mutex_lock(&io->mutex);
while (1) {
unsigned char* let;
unsigned index;
/* get the next index the caller is going to use */
/* we must ensure that this index has not a write in progress */
/* to avoid a concurrent access */
/* note that we are already sure that a write is not in progress */
/* at the index the caller is using at now */
index = (io->writer_index + 1) % IO_MAX;
/* search for a worker that has already finished */
let = &io->writer_list[0];
while (1) {
unsigned i = *let;
struct snapraid_worker* worker;
/* if we are at the end */
if (i == io->writer_max)
break;
worker = &io->writer_map[i];
/* the two indexes cannot be equal */
assert(io->writer_index != worker->index);
/* if the worker has finished this index */
if (index != worker->index) {
pthread_mutex_unlock(&io->mutex);
/* mark the worker as processed */
/* setting the previous one to point at the next one */
*let = io->writer_list[i + 1];
/* return the position */
*pos = i;
return;
}
/* next position to check */
let = &io->writer_list[i + 1];
}
/* if not worker is ready, wait for an event */
pthread_cond_wait(&io->write_done, &io->mutex);
}
}
static void io_reader_worker(struct snapraid_worker* worker, struct snapraid_task* task)
{
/* if we reached the end */
@ -340,6 +554,33 @@ static void* io_reader_thread(void* arg)
return 0;
}
static void* io_writer_thread(void* arg)
{
struct snapraid_worker* worker = arg;
while (1) {
struct snapraid_task* task;
/* get the new task */
task = io_writer_step(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 */
worker->func(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)
@ -354,6 +595,7 @@ void io_start(struct snapraid_io* io,
io->done = 0;
io->reader_index = IO_MAX - 1;
io->writer_index = 0;
/* setup the initial read pending tasks, except the latest one, */
/* the latest will be initialized at the fist io_read_next() call */
@ -365,6 +607,8 @@ void io_start(struct snapraid_io* io,
/* setup the lists of workers to process */
io->reader_list[0] = io->reader_max;
for (i = 0; i <= io->writer_max; ++i)
io->writer_list[i] = i;
/* start the reader threads */
for (i = 0; i < io->reader_max; ++i) {
@ -379,6 +623,20 @@ void io_start(struct snapraid_io* io,
/* LCOV_EXCL_STOP */
}
}
/* start the writer threads */
for (i = 0; i < io->writer_max; ++i) {
struct snapraid_worker* worker = &io->writer_map[i];
worker->index = IO_MAX - 1;
if (pthread_create(&worker->thread, 0, io_writer_thread, worker) != 0) {
/* LCOV_EXCL_START */
log_fatal("Failed to create writer thread.\n");
exit(EXIT_FAILURE);
/* LCOV_EXCL_STOP */
}
}
}
void io_stop(struct snapraid_io* io)
@ -392,6 +650,7 @@ void io_stop(struct snapraid_io* io)
/* signal all the threads to recognize the new state */
pthread_cond_broadcast(&io->read_sched);
pthread_cond_broadcast(&io->write_sched);
pthread_mutex_unlock(&io->mutex);
@ -408,5 +667,19 @@ void io_stop(struct snapraid_io* io)
/* LCOV_EXCL_STOP */
}
}
/* wait for all writers to terminate */
for (i = 0; i < io->writer_max; ++i) {
struct snapraid_worker* worker = &io->writer_map[i];
void* retval;
/* wait for thread termination */
if (pthread_join(worker->thread, &retval) != 0) {
/* LCOV_EXCL_START */
log_fatal("Failed to join writer thread.\n");
exit(EXIT_FAILURE);
/* LCOV_EXCL_STOP */
}
}
}

56
cmdline/io.h
View file

@ -151,6 +151,24 @@ struct snapraid_io {
*/
pthread_cond_t read_sched;
/**
* Condition for a new write is completed.
*
* The workers signal this condition when a new write is completed.
* The IO waits on this condition when it's waiting for
* a new write to be completed.
*/
pthread_cond_t write_done;
/**
* Condition for a new write scheduled.
*
* The workers wait on this condition when they are waiting for a new
* write to process.
* The IO signals this condition when new writes are scheduled.
*/
pthread_cond_t write_sched;
struct snapraid_state* state;
/**
@ -168,6 +186,7 @@ struct snapraid_io {
*/
void (*data_reader)(struct snapraid_worker*, struct snapraid_task*);
void (*parity_reader)(struct snapraid_worker*, struct snapraid_task*);
void (*parity_writer)(struct snapraid_worker*, struct snapraid_task*);
/**
* Blocks mapping.
@ -197,6 +216,8 @@ struct snapraid_io {
*/
unsigned reader_max; /**< Number of workers. */
struct snapraid_worker* reader_map; /**< Vector of workers. */
unsigned writer_max; /**< Number of workers. */
struct snapraid_worker* writer_map; /**< Vector of workers. */
/**
* List of not yet processed workers.
@ -209,6 +230,7 @@ struct snapraid_io {
* The end is when i == reader_max.
*/
unsigned char* reader_list;
unsigned char* writer_list;
/**
* Exit condition for all threads.
@ -225,6 +247,17 @@ struct snapraid_io {
* it's incremented, and a read_sched() signal is sent.
*/
unsigned reader_index;
/**
* The task currently used by the caller.
*
* It's a rolling counter, when reaching IO_MAX
* it goes again to 0.
*
* When the caller finish with the current index,
* it's incremented, and a write_sched() signal is sent.
*/
unsigned writer_index;
};
/**
@ -237,6 +270,7 @@ void io_init(struct snapraid_io* io, struct snapraid_state* state,
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*),
void (*parity_writer)(struct snapraid_worker*, struct snapraid_task*),
struct snapraid_parity_handle* parity_handle_map, unsigned parity_handle_max);
/**
@ -290,4 +324,26 @@ struct snapraid_task* io_data_read(struct snapraid_io* io, unsigned* diskcur);
*/
struct snapraid_task* io_parity_read(struct snapraid_io* io, unsigned* levcur);
/**
* Write of a parity block.
*
* 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.
*/
void io_parity_write(struct snapraid_io* io, unsigned* levcur);
/**
* Next write position.
*
* This call ends the write process.
* It must be called after io_parity_write().
*
* \param io InputOutput context.
* \param blockcur The parity position to write.
* \param skip Skip the writes, in case parity doesn't need to be updated.
*/
void io_write_next(struct snapraid_io* io, unsigned blockcur, int skip);
#endif

2
cmdline/scrub.c
View file

@ -275,7 +275,7 @@ static int state_scrub_process(struct snapraid_state* state, struct snapraid_par
buffermax = diskmax + state->level * 2;
/* initialize the io threads */
io_init(&io, state, buffermax, scrub_data_reader, handle, diskmax, scrub_parity_reader, parity_handle, state->level);
io_init(&io, state, buffermax, scrub_data_reader, handle, diskmax, scrub_parity_reader, 0, parity_handle, state->level);
error = 0;
silent_error = 0;

143
cmdline/sync.c
View file

@ -615,6 +615,39 @@ static void sync_data_reader(struct snapraid_worker* worker, struct snapraid_tas
task->state = TASK_STATE_DONE;
}
static void sync_parity_writer(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;
/* write parity */
ret = parity_write(parity_handle, blockcur, buffer, state->block_size);
if (ret == -1) {
/* LCOV_EXCL_START */
if (errno == EIO) {
log_tag("parity_error:%u:%s: Write EIO error. %s\n", blockcur, lev_config_name(level), strerror(errno));
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: Write error. %s\n", blockcur, lev_config_name(level), strerror(errno));
log_fatal("WARNING! Unexpected write error in the %s disk, it isn't possible to sync.\n", lev_name(level));
log_fatal("Ensure that disk '%s' has some free space available.\n", lev_config_name(level));
log_fatal("Stopping at block %u\n", blockcur);
task->state = TASK_STATE_ERROR;
return;
/* LCOV_EXCL_STOP */
}
task->state = TASK_STATE_DONE;
}
static int state_sync_process(struct snapraid_state* state, struct snapraid_parity_handle* parity_handle, block_off_t blockstart, block_off_t blockmax)
{
struct snapraid_io io;
@ -660,7 +693,7 @@ static int state_sync_process(struct snapraid_state* state, struct snapraid_pari
buffermax = 2 * diskmax + state->level;
/* initialize the io threads */
io_init(&io, state, buffermax, sync_data_reader, handle, diskmax, 0, 0, 0);
io_init(&io, state, buffermax, sync_data_reader, handle, diskmax, 0, sync_parity_writer, parity_handle, state->level);
/* fill up the zero buffer */
zero = malloc_nofail_align(state->block_size, &zero_alloc);
@ -710,6 +743,7 @@ static int state_sync_process(struct snapraid_state* state, struct snapraid_pari
int io_error_on_this_block;
int fixed_error_on_this_block;
int parity_needs_to_be_updated;
int parity_going_to_be_updated;
snapraid_info info;
int rehash;
void** buffer;
@ -738,14 +772,18 @@ static int state_sync_process(struct snapraid_state* state, struct snapraid_pari
/* if we have to use the old hash */
rehash = info_get_rehash(info);
/* it could happens that all the blocks are EMPTY/BLK and CHG but with the hash */
/* if the parity requires to be updated */
/* It could happens that all the blocks are EMPTY/BLK and CHG but with the hash */
/* still matching because the specific CHG block was not modified. */
/* In such case, we can avoid to update parity, because it would be the same as before */
/* Note that CHG/DELETED blocks already present in the content file loaded */
/* have the hash cleared (::clear_past_hash flag), and then they won't never match the hash. */
/* We are treating only CHG blocks created at runtime. */
/* In such case, we can avoid to update parity, because it would be the same as before */
parity_needs_to_be_updated = state->opt.force_parity_update;
/* if the parity is going to be updated */
parity_going_to_be_updated = 0;
/* if the block is marked as bad, we force the parity update */
/* because the bad block may be the result of a wrong parity */
if (info_get_bad(info))
@ -1078,70 +1116,33 @@ static int state_sync_process(struct snapraid_state* state, struct snapraid_pari
/* compute the parity */
raid_gen(diskmax, state->level, state->block_size, buffer);
/* until now is CPU */
state_usage_cpu(state);
/* write the parity */
for (l = 0; l < state->level; ++l) {
ret = parity_write(&parity_handle[l], blockcur, buffer[diskmax + l], state->block_size);
if (ret == -1) {
/* LCOV_EXCL_START */
if (errno == EIO) {
log_tag("parity_error:%u:%s: Write EIO error. %s\n", blockcur, lev_config_name(l), strerror(errno));
if (io_error >= state->opt.io_error_limit) {
log_fatal("DANGER! Unexpected input/output write error in the %s disk, it isn't possible to sync.\n", lev_name(l));
log_fatal("Ensure that disk '%s' is sane and can be written.\n", lev_config_name(l));
log_fatal("Stopping at block %u\n", blockcur);
++io_error;
goto bail;
}
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: Write error. %s\n", blockcur, lev_config_name(l), strerror(errno));
log_fatal("WARNING! Unexpected write error in the %s disk, it isn't possible to sync.\n", lev_name(l));
log_fatal("Ensure that disk '%s' has some free space available.\n", lev_config_name(l));
log_fatal("Stopping at block %u\n", blockcur);
++error;
goto bail;
/* LCOV_EXCL_STOP */
}
/* until now is parity */
state_usage_parity(state, l);
}
/* mark that the parity is going to be written */
parity_going_to_be_updated = 1;
}
/* if no error in parity write */
if (!io_error_on_this_block) {
/* for each disk, mark the blocks as processed */
for (j = 0; j < diskmax; ++j) {
struct snapraid_block* block;
/* for each disk, mark the blocks as processed */
for (j = 0; j < diskmax; ++j) {
struct snapraid_block* block;
if (!handle[j].disk)
continue;
if (!handle[j].disk)
continue;
block = fs_par2block_get(handle[j].disk, blockcur);
block = fs_par2block_get(handle[j].disk, blockcur);
if (block == BLOCK_EMPTY) {
/* nothing to do */
continue;
}
/* if it's a deleted block */
if (block_state_get(block) == BLOCK_STATE_DELETED) {
/* the parity is now updated without this block, so it's now empty */
fs_deallocate(handle[j].disk, blockcur);
continue;
}
/* now all the blocks have the hash and the parity computed */
block_state_set(block, BLOCK_STATE_BLK);
if (block == BLOCK_EMPTY) {
/* nothing to do */
continue;
}
/* if it's a deleted block */
if (block_state_get(block) == BLOCK_STATE_DELETED) {
/* the parity is now updated without this block, so it's now empty */
fs_deallocate(handle[j].disk, blockcur);
continue;
}
/* now all the blocks have the hash and the parity computed */
block_state_set(block, BLOCK_STATE_BLK);
}
/* we update the info block only if we really have updated the parity */
@ -1151,7 +1152,6 @@ static int state_sync_process(struct snapraid_state* state, struct snapraid_pari
/* because has no sense to refresh the time for data that we know bad */
if (parity_needs_to_be_updated
&& !silent_error_on_this_block
&& !io_error_on_this_block
) {
/* if rehash is neeed */
if (rehash) {
@ -1176,6 +1176,25 @@ static int state_sync_process(struct snapraid_state* state, struct snapraid_pari
info_set(&state->infoarr, blockcur, info_set_bad(info));
}
/* finally schedule parity write */
/* Note that the calls to io_parity_write() are mandatory */
/* even if the parity doesn't need to be updated */
/* This because we want to keep track of the time usage */
state_usage_cpu(state);
/* write the parity */
for (l = 0; l < state->level; ++l) {
unsigned levcur;
io_parity_write(&io, &levcur);
/* until now is parity */
state_usage_parity(state, levcur);
}
/* write finished */
io_write_next(&io, blockcur, !parity_going_to_be_updated);
/* mark the state as needing write */
state->need_write = 1;