From a3121c31c58b60fb41ebe4b05abdfbaf0be457a1 Mon Sep 17 00:00:00 2001 From: Andrea Mazzoleni Date: Mon, 4 Jan 2016 11:28:34 +0100 Subject: [PATCH] Add support for multithread parity writing in sync --- cmdline/io.c | 275 +++++++++++++++++++++++++++++++++++++++++++++++- cmdline/io.h | 56 ++++++++++ cmdline/scrub.c | 2 +- cmdline/sync.c | 143 ++++++++++++++----------- 4 files changed, 412 insertions(+), 64 deletions(-) diff --git a/cmdline/io.c b/cmdline/io.c index 79ce0e2e..4e1c7b77 100644 --- a/cmdline/io.c +++ b/cmdline/io.c @@ -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 */ + } + } } diff --git a/cmdline/io.h b/cmdline/io.h index 94ba58cd..da240a87 100644 --- a/cmdline/io.h +++ b/cmdline/io.h @@ -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 diff --git a/cmdline/scrub.c b/cmdline/scrub.c index 4b5da24d..800a3c1a 100644 --- a/cmdline/scrub.c +++ b/cmdline/scrub.c @@ -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; diff --git a/cmdline/sync.c b/cmdline/sync.c index e8785314..6f4b7c79 100644 --- a/cmdline/sync.c +++ b/cmdline/sync.c @@ -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;