From 62b7587c3a568bb444d20030c0f1f4f03fd913a4 Mon Sep 17 00:00:00 2001 From: Andrea Mazzoleni Date: Mon, 4 Jan 2016 11:24:04 +0100 Subject: [PATCH] IO functions renaming to prepare for parity write support No functional change. --- cmdline/io.c | 143 +++++++++++++++++++++++++++--------------------- cmdline/io.h | 54 ++++++++++-------- cmdline/scrub.c | 6 +- cmdline/sync.c | 4 +- 4 files changed, 116 insertions(+), 91 deletions(-) diff --git a/cmdline/io.c b/cmdline/io.c index f8a7d422..79ce0e2e 100644 --- a/cmdline/io.c +++ b/cmdline/io.c @@ -28,8 +28,8 @@ void io_init(struct snapraid_io* io, struct snapraid_state* state, unsigned buff unsigned i; pthread_mutex_init(&io->mutex, 0); - pthread_cond_init(&io->not_empty, 0); - pthread_cond_init(&io->not_full, 0); + pthread_cond_init(&io->read_done, 0); + pthread_cond_init(&io->read_sched, 0); io->state = state; @@ -40,21 +40,22 @@ 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->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->reader_max = handle_max + parity_handle_max; + io->reader_map = malloc_nofail(sizeof(struct snapraid_worker) * io->reader_max); + io->reader_list = malloc_nofail(io->reader_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]; + for (i = 0; i < io->reader_max; ++i) { + struct snapraid_worker* worker = &io->reader_map[i]; worker->io = io; if (i < handle_max) { + /* it's a data read */ worker->handle = &handle_map[i]; worker->parity_handle = 0; worker->func = data_reader; @@ -62,6 +63,7 @@ void io_init(struct snapraid_io* io, struct snapraid_state* state, unsigned buff /* data read is put in lower buffer index */ worker->buffer_skew = 0; } else { + /* it's a parity read */ worker->handle = 0; worker->parity_handle = &parity_handle_map[i - handle_max]; worker->func = parity_reader; @@ -81,12 +83,12 @@ void io_done(struct snapraid_io* io) free(io->buffer_alloc_map[i]); } - free(io->worker_map); - free(io->worker_list); + free(io->reader_map); + free(io->reader_list); pthread_mutex_destroy(&io->mutex); - pthread_cond_destroy(&io->not_empty); - pthread_cond_destroy(&io->not_full); + pthread_cond_destroy(&io->read_done); + pthread_cond_destroy(&io->read_sched); } /** @@ -109,14 +111,14 @@ static block_off_t io_position_next(struct snapraid_io* io) } /** - * Setup the next pending task for all workers. + * Setup the next pending task for all readers. */ -static void io_pending_next(struct snapraid_io* io, int index, block_off_t blockcur) +static void io_reader_sched(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]; + for (i = 0; i < io->reader_max; ++i) { + struct snapraid_worker* worker = &io->reader_map[i]; struct snapraid_task* task = &worker->task_map[index]; /* setup the new pending task */ @@ -141,11 +143,11 @@ static void io_pending_next(struct snapraid_io* io, int index, block_off_t block } /** - * Get the next task to work on for a worker. + * Get the next task to work on for a reader. * * This is the synchronization point for workers with the io. */ -static struct snapraid_task* io_worker_next(struct snapraid_worker* worker) +static struct snapraid_task* io_reader_step(struct snapraid_worker* worker) { struct snapraid_io* io = worker->io; struct snapraid_task* task = 0; @@ -157,6 +159,7 @@ static struct snapraid_task* io_worker_next(struct snapraid_worker* worker) unsigned index; /* check if the worker has to exit */ + /* even if there is work to do */ if (io->done) break; @@ -164,20 +167,20 @@ static struct snapraid_task* io_worker_next(struct snapraid_worker* worker) index = (worker->index + 1) % IO_MAX; /* if the queue of pending tasks is not empty */ - if (index != io->index) { + if (index != io->reader_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); + /* notify the io that a new read is complete */ + pthread_cond_signal(&io->read_done); /* return the new task */ break; } - /* otherwise wait for a not_full event */ - pthread_cond_wait(&io->not_full, &io->mutex); + /* otherwise wait for a read_sched event */ + pthread_cond_wait(&io->read_sched, &io->mutex); } pthread_mutex_unlock(&io->mutex); @@ -190,75 +193,85 @@ static struct snapraid_task* io_worker_next(struct snapraid_worker* worker) * * This is the synchronization point for workers with the io. */ -block_off_t io_next(struct snapraid_io* io, void*** buffer) +block_off_t io_read_next(struct snapraid_io* io, void*** buffer) { block_off_t blockcur_schedule; block_off_t blockcur_caller; unsigned i; + /* get the next parity position to process */ blockcur_schedule = io_position_next(io); + /* ensure that all data/parity was read */ + assert(io->reader_list[0] == io->reader_max); + /* setup the list of workers to process */ - for (i = 0; i <= io->worker_max; ++i) - io->worker_list[i] = i; + for (i = 0; i <= io->reader_max; ++i) + io->reader_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); + /* schedule the next read */ + io_reader_sched(io, io->reader_index, blockcur_schedule); /* get the next task to return to the caller */ - io->index = (io->index + 1) % IO_MAX; + io->reader_index = (io->reader_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; + blockcur_caller = io->reader_map[0].task_map[io->reader_index].position; /* set the buffer to use */ - *buffer = io->buffer_map[io->index]; + *buffer = io->buffer_map[io->reader_index]; /* signal all the workers that there is a new pending task */ - pthread_cond_broadcast(&io->not_full); + pthread_cond_broadcast(&io->read_sched); 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) +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 */ pthread_mutex_lock(&io->mutex); while (1) { unsigned char* let; + unsigned index; + + /* get the index the caller is using */ + /* we must ensure that this index has not a read in progress */ + /* to avoid a concurrent access */ + index = io->reader_index; /* search for a worker that has already finished */ - let = &io->worker_list[0]; + let = &io->reader_list[0]; while (1) { unsigned i = *let; /* if we are at the end */ - if (i == io->worker_max) + if (i == io->reader_max) break; /* if it's in range */ if (base <= i && i < base + count) { struct snapraid_worker* worker; - worker = &io->worker_map[i]; + worker = &io->reader_map[i]; /* if the worker has finished this index */ - if (io->index != worker->index) { + if (index != worker->index) { struct snapraid_task* task; - task = &worker->task_map[io->index]; + task = &worker->task_map[io->reader_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]; + *let = io->reader_list[i + 1]; /* return the position */ *pos = i - base; @@ -268,46 +281,47 @@ static struct snapraid_task* io_task_next(struct snapraid_io* io, unsigned base, } /* next position to check */ - let = &io->worker_list[i + 1]; + let = &io->reader_list[i + 1]; } /* if not worker is ready, wait for an event */ - pthread_cond_wait(&io->not_empty, &io->mutex); + pthread_cond_wait(&io->read_done, &io->mutex); } } -struct snapraid_task* io_data_next(struct snapraid_io* io, unsigned* pos) +struct snapraid_task* io_data_read(struct snapraid_io* io, unsigned* pos) { - return io_task_next(io, io->data_base, io->data_count, pos); + return io_task_read(io, io->data_base, io->data_count, pos); } -struct snapraid_task* io_parity_next(struct snapraid_io* io, unsigned* pos) +struct snapraid_task* io_parity_read(struct snapraid_io* io, unsigned* pos) { - return io_task_next(io, io->parity_base, io->parity_count, pos); + return io_task_read(io, io->parity_base, io->parity_count, pos); } -static void io_worker(struct snapraid_worker* worker, struct snapraid_task* task) +static void io_reader_worker(struct snapraid_worker* worker, struct snapraid_task* task) { /* if we reached the end */ if (task->position >= worker->io->block_max) { + /* complete a dummy task */ task->state = TASK_STATE_EMPTY; } else { worker->func(worker, task); } } -static void* io_thread(void* arg) +static void* io_reader_thread(void* arg) { struct snapraid_worker* worker = arg; /* force completion of the first task */ - io_worker(worker, &worker->task_map[0]); + io_reader_worker(worker, &worker->task_map[0]); while (1) { struct snapraid_task* task; /* get the new task */ - task = io_worker_next(worker); + task = io_reader_step(worker); /* if no task, it means to exit */ if (!task) @@ -320,7 +334,7 @@ static void* io_thread(void* arg) assert(task->state == TASK_STATE_READY); /* work on the assigned task */ - io_worker(worker, task); + io_reader_worker(worker, task); } return 0; @@ -339,25 +353,28 @@ void io_start(struct snapraid_io* io, io->block_next = blockstart; io->done = 0; - io->index = IO_MAX - 1; + io->reader_index = IO_MAX - 1; - /* setup the initial pending tasks, except the latest one, */ - /* the latest will be initialized at the fist io_next() call */ + /* setup the initial read pending tasks, except the latest one, */ + /* the latest will be initialized at the fist io_read_next() call */ for (i = 0; i < IO_MAX - 1; ++i) { block_off_t blockcur = io_position_next(io); - io_pending_next(io, i, blockcur); + io_reader_sched(io, i, blockcur); } - /* start the worker threads */ - for (i = 0; i < io->worker_max; ++i) { - struct snapraid_worker* worker = &io->worker_map[i]; + /* setup the lists of workers to process */ + io->reader_list[0] = io->reader_max; + + /* start the reader threads */ + for (i = 0; i < io->reader_max; ++i) { + struct snapraid_worker* worker = &io->reader_map[i]; worker->index = 0; - if (pthread_create(&worker->thread, 0, io_thread, worker) != 0) { + if (pthread_create(&worker->thread, 0, io_reader_thread, worker) != 0) { /* LCOV_EXCL_START */ - log_fatal("Failed to create thread.\n"); + log_fatal("Failed to create reader thread.\n"); exit(EXIT_FAILURE); /* LCOV_EXCL_STOP */ } @@ -374,19 +391,19 @@ void io_stop(struct snapraid_io* io) io->done = 1; /* signal all the threads to recognize the new state */ - pthread_cond_broadcast(&io->not_full); + pthread_cond_broadcast(&io->read_sched); 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]; + /* wait for all readers to terminate */ + for (i = 0; i < io->reader_max; ++i) { + struct snapraid_worker* worker = &io->reader_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"); + log_fatal("Failed to join reader thread.\n"); exit(EXIT_FAILURE); /* LCOV_EXCL_STOP */ } diff --git a/cmdline/io.h b/cmdline/io.h index 9514ada0..94ba58cd 100644 --- a/cmdline/io.h +++ b/cmdline/io.h @@ -134,20 +134,22 @@ struct snapraid_io { pthread_mutex_t mutex; /** - * Condition for (not) empty queue of tasks. + * Condition for a new read is completed. * - * The io waits on this condition when the queue is empty, - * and it's signaled by workers. + * The workers signal this condition when a new read is completed. + * The IO waits on this condition when it's waiting for + * a new read to be completed. */ - pthread_cond_t not_empty; + pthread_cond_t read_done; /** - * Condition for (not) full queue of tasks. + * Condition for a new read scheduled. * - * The workers wait on this condition when the queue is full, - * and it's signaled by the io. + * The workers wait on this condition when they are waiting for a new + * read to process. + * The IO signals this condition when new reads are scheduled. */ - pthread_cond_t not_full; + pthread_cond_t read_sched; struct snapraid_state* state; @@ -191,22 +193,22 @@ struct snapraid_io { /** * Workers. * - * A vector of workers, each one representing a different thread. + * A vector of readers, each one representing a different thread. */ - unsigned worker_max; /**< Number of workers. */ - struct snapraid_worker* worker_map; /**< Vector of workers. */ + unsigned reader_max; /**< Number of workers. */ + struct snapraid_worker* reader_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. + * The list has ::reader_max + 1 elements. Each element contains + * the number of the reader to process. * - * At initialization the list is filled with [0..worker_max]. + * At initialization the list is filled with [0..reader_max]. * To get the next element to process we use i = list[i + 1]. - * The end is when i == worker_max. + * The end is when i == reader_max. */ - unsigned char* worker_list; + unsigned char* reader_list; /** * Exit condition for all threads. @@ -218,8 +220,11 @@ struct snapraid_io { * * 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 read_sched() signal is sent. */ - unsigned index; + unsigned reader_index; }; /** @@ -252,16 +257,19 @@ void io_start(struct snapraid_io* io, void io_stop(struct snapraid_io* io); /** - * Get the next parity position to process. + * Next read position. + * + * This call starts the reading process. + * It must be called before io_data_read() and io_parity_read(). * * \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); +block_off_t io_read_next(struct snapraid_io* io, void*** buffer); /** - * Get the next data block to process. + * Read a data block. * * It must be called exactly ::handle_max times. * @@ -269,10 +277,10 @@ block_off_t io_next(struct snapraid_io* io, void*** buffer); * \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); +struct snapraid_task* io_data_read(struct snapraid_io* io, unsigned* diskcur); /** - * Get the next parity block to process. + * Read a parity block. * * It must be called exactly ::parity_handle_max times. * @@ -280,6 +288,6 @@ struct snapraid_task* io_data_next(struct snapraid_io* io, unsigned* pos); * \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); +struct snapraid_task* io_parity_read(struct snapraid_io* io, unsigned* levcur); #endif diff --git a/cmdline/scrub.c b/cmdline/scrub.c index fccedd69..4b5da24d 100644 --- a/cmdline/scrub.c +++ b/cmdline/scrub.c @@ -319,7 +319,7 @@ static int state_scrub_process(struct snapraid_state* state, struct snapraid_par void** buffer; /* go to the next block */ - blockcur = io_next(&io, &buffer); + blockcur = io_read_next(&io, &buffer); if (blockcur >= blockmax) break; @@ -362,7 +362,7 @@ static int state_scrub_process(struct snapraid_state* state, struct snapraid_par state_usage_cpu(state); /* get the next task */ - task = io_data_next(&io, &diskcur); + task = io_data_read(&io, &diskcur); /* get the task results */ disk = task->disk; @@ -485,7 +485,7 @@ static int state_scrub_process(struct snapraid_state* state, struct snapraid_par struct snapraid_task* task; unsigned levcur; - task = io_parity_next(&io, &levcur); + task = io_parity_read(&io, &levcur); /* until now is parity */ state_usage_parity(state, levcur); diff --git a/cmdline/sync.c b/cmdline/sync.c index e5da297d..e8785314 100644 --- a/cmdline/sync.c +++ b/cmdline/sync.c @@ -715,7 +715,7 @@ static int state_sync_process(struct snapraid_state* state, struct snapraid_pari void** buffer; /* go to the next block */ - blockcur = io_next(&io, &buffer); + blockcur = io_read_next(&io, &buffer); if (blockcur >= blockmax) break; @@ -766,7 +766,7 @@ static int state_sync_process(struct snapraid_state* state, struct snapraid_pari /* until now is CPU */ state_usage_cpu(state); - task = io_data_next(&io, &diskcur); + task = io_data_read(&io, &diskcur); /* get the results */ disk = task->disk;