diff --git a/HISTORY b/HISTORY index 63975ff6..ac027d99 100644 --- a/HISTORY +++ b/HISTORY @@ -7,6 +7,8 @@ SnapRAID HISTORY file patterns containing real characters matching the globbing '*?[]' ones. In Unix it was already possible to do the same escaping with '\'. + * Added the --test-io-cache=1 option to disable the multithread IO + mode. 10.0 2016/02 ============ diff --git a/cmdline/io.c b/cmdline/io.c index 0bed26c6..d4af5fb5 100644 --- a/cmdline/io.c +++ b/cmdline/io.c @@ -26,125 +26,9 @@ #error Pthread library is required for InputOutput! #endif -void io_init(struct snapraid_io* io, struct snapraid_state* state, - unsigned io_cache, 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*), - void (*parity_writer)(struct snapraid_worker*, struct snapraid_task*), - struct snapraid_parity_handle* parity_handle_map, unsigned parity_handle_max) +static inline int io_thread(struct snapraid_io* io) { - unsigned i; - size_t allocated; - - thread_mutex_init(&io->io_mutex, 0); - thread_cond_init(&io->read_done, 0); - thread_cond_init(&io->read_sched, 0); - thread_cond_init(&io->write_done, 0); - thread_cond_init(&io->write_sched, 0); - - io->state = state; - if (io_cache == 0) { - /* default is 8 MiB of cache */ - /* this seems to be a good tradeoff between speed and memory usage */ - io->io_max = 8 * 1024 * 1024 / state->block_size; - if (io->io_max < IO_MIN) - io->io_max = IO_MIN; - if (io->io_max > IO_MAX) - io->io_max = IO_MAX; - } else { - io->io_max = io_cache; - } - - assert(io->io_max >= IO_MIN && io->io_max <= IO_MAX); - - io->buffer_max = buffer_max; - allocated = 0; - for (i = 0; i < io->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]); - allocated += state->block_size * buffer_max; - } - - msg_progress("Using %u MiB of memory for %u blocks of IO cache.\n", (unsigned)(allocated / MEBI), io->io_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; - io->parity_base = handle_max; - io->parity_count = parity_handle_max; - - 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; - - /* 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; - - /* parity read is put after data and computed parity */ - 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) -{ - unsigned i; - - for (i = 0; i < io->io_max; ++i) { - free(io->buffer_map[i]); - free(io->buffer_alloc_map[i]); - } - - free(io->reader_map); - free(io->reader_list); - free(io->writer_map); - free(io->writer_list); - - thread_mutex_destroy(&io->io_mutex); - thread_cond_destroy(&io->read_done); - thread_cond_destroy(&io->read_sched); - thread_cond_destroy(&io->write_done); - thread_cond_destroy(&io->write_sched); + return io->io_max > 1; } /** @@ -248,6 +132,147 @@ static void io_writer_sched_empty(struct snapraid_io* io, int task_index, block_ } } +/*****************************************************************************/ +/* mono thread */ + +static block_off_t io_read_next_mono(struct snapraid_io* io, void*** buffer) +{ + block_off_t blockcur_schedule; + + /* reset the index */ + io->reader_index = 0; + + blockcur_schedule = io_position_next(io); + + /* schedule the next read */ + io_reader_sched(io, 0, blockcur_schedule); + + /* set the buffer to use */ + *buffer = io->buffer_map[0]; + + return blockcur_schedule; +} + +static void io_write_preset_mono(struct snapraid_io* io, block_off_t blockcur, int skip) +{ + unsigned i; + + /* reset the index */ + io->writer_index = 0; + + /* clear errors */ + for (i = 0; i < IO_WRITER_ERROR_MAX; ++i) + io->writer_error[i] = 0; + + if (skip) { + /* skip the next write */ + io_writer_sched_empty(io, 0, blockcur); + } else { + /* schedule the next write */ + io_writer_sched(io, 0, blockcur); + } +} + +static void io_write_next_mono(struct snapraid_io* io, block_off_t blockcur, int skip, int* writer_error) +{ + unsigned i; + + (void)blockcur; + (void)skip; + + /* report errors */ + for (i = 0; i < IO_WRITER_ERROR_MAX; ++i) + writer_error[i] = io->writer_error[i]; +} + +static void io_refresh_mono(struct snapraid_io* io) +{ + (void)io; +} + +static struct snapraid_task* io_task_read_mono(struct snapraid_io* io, unsigned base, unsigned count, unsigned* pos, unsigned* waiting_map, unsigned* waiting_mac) +{ + struct snapraid_worker* worker; + struct snapraid_task* task; + unsigned i; + + /* get the next task */ + i = io->reader_index++; + + assert(base <= i && i < base + count); + + worker = &io->reader_map[i]; + task = &worker->task_map[0]; + + /* do the work */ + if (task->state != TASK_STATE_EMPTY) + worker->func(worker, task); + + /* return the position */ + *pos = i - base; + + /* store the waiting index */ + waiting_map[0] = i - base; + *waiting_mac = 1; + + return task; +} + +static struct snapraid_task* io_data_read_mono(struct snapraid_io* io, unsigned* pos, unsigned* waiting_map, unsigned* waiting_mac) +{ + return io_task_read_mono(io, io->data_base, io->data_count, pos, waiting_map, waiting_mac); +} + +static struct snapraid_task* io_parity_read_mono(struct snapraid_io* io, unsigned* pos, unsigned* waiting_map, unsigned* waiting_mac) +{ + return io_task_read_mono(io, io->parity_base, io->parity_count, pos, waiting_map, waiting_mac); +} + +static void io_parity_write_mono(struct snapraid_io* io, unsigned* pos, unsigned* waiting_map, unsigned* waiting_mac) +{ + struct snapraid_worker* worker; + struct snapraid_task* task; + unsigned i; + + /* get the next task */ + i = io->writer_index++; + + worker = &io->writer_map[i]; + task = &worker->task_map[0]; + + io->writer_error[i] = 0; + + /* do the work */ + if (task->state != TASK_STATE_EMPTY) + worker->func(worker, task); + + /* return the position */ + *pos = i; + + /* store the waiting index */ + waiting_map[0] = i; + *waiting_mac = 1; +} + +static void io_start_mono(struct snapraid_io* io, + block_off_t blockstart, block_off_t blockmax, + int (*block_is_enabled)(void* arg, block_off_t), void* blockarg) +{ + io->block_start = blockstart; + io->block_max = blockmax; + io->block_is_enabled = block_is_enabled; + io->block_arg = blockarg; + io->block_next = blockstart; +} + +static void io_stop_mono(struct snapraid_io* io) +{ + (void)io; +} + +/*****************************************************************************/ +/* multi thread */ + /** * Get the next task to work on for a reader. * @@ -331,7 +356,7 @@ static struct snapraid_task* io_writer_step(struct snapraid_worker* worker, int /* if the queue of pending tasks is not empty */ if (next_index != io->writer_index) { struct snapraid_task* task; - + /* the index that the IO may be waiting for */ unsigned waiting_index = (io->writer_index + 1) % io->io_max; @@ -371,7 +396,7 @@ static struct snapraid_task* io_writer_step(struct snapraid_worker* worker, int * * This is the synchronization point for workers with the io. */ -block_off_t io_read_next(struct snapraid_io* io, void*** buffer) +static block_off_t io_read_next_thread(struct snapraid_io* io, void*** buffer) { block_off_t blockcur_schedule; block_off_t blockcur_caller; @@ -408,7 +433,14 @@ block_off_t io_read_next(struct snapraid_io* io, void*** buffer) return blockcur_caller; } -void io_write_next(struct snapraid_io* io, block_off_t blockcur, int skip, int* writer_error) +static void io_write_preset_thread(struct snapraid_io* io, block_off_t blockcur, int skip) +{ + (void)io; + (void)blockcur; + (void)skip; +} + +static void io_write_next_thread(struct snapraid_io* io, block_off_t blockcur, int skip, int* writer_error) { unsigned i; @@ -446,7 +478,7 @@ void io_write_next(struct snapraid_io* io, block_off_t blockcur, int skip, int* thread_cond_broadcast_and_unlock(&io->write_sched, &io->io_mutex); } -void io_refresh(struct snapraid_io* io) +static void io_refresh_thread(struct snapraid_io* io) { unsigned i; @@ -492,7 +524,7 @@ void io_refresh(struct snapraid_io* io) thread_mutex_unlock(&io->io_mutex); } -static struct snapraid_task* io_task_read(struct snapraid_io* io, unsigned base, unsigned count, unsigned* pos, unsigned* waiting_map, unsigned* waiting_mac) +static struct snapraid_task* io_task_read_thread(struct snapraid_io* io, unsigned base, unsigned count, unsigned* pos, unsigned* waiting_map, unsigned* waiting_mac) { unsigned waiting_cycle; @@ -570,17 +602,17 @@ static struct snapraid_task* io_task_read(struct snapraid_io* io, unsigned base, } } -struct snapraid_task* io_data_read(struct snapraid_io* io, unsigned* pos, unsigned* waiting_map, unsigned* waiting_mac) +static struct snapraid_task* io_data_read_thread(struct snapraid_io* io, unsigned* pos, unsigned* waiting_map, unsigned* waiting_mac) { - return io_task_read(io, io->data_base, io->data_count, pos, waiting_map, waiting_mac); + return io_task_read_thread(io, io->data_base, io->data_count, pos, waiting_map, waiting_mac); } -struct snapraid_task* io_parity_read(struct snapraid_io* io, unsigned* pos, unsigned* waiting_map, unsigned* waiting_mac) +static struct snapraid_task* io_parity_read_thread(struct snapraid_io* io, unsigned* pos, unsigned* waiting_map, unsigned* waiting_mac) { - return io_task_read(io, io->parity_base, io->parity_count, pos, waiting_map, waiting_mac); + return io_task_read_thread(io, io->parity_base, io->parity_count, pos, waiting_map, waiting_mac); } -void io_parity_write(struct snapraid_io* io, unsigned* pos, unsigned* waiting_map, unsigned* waiting_mac) +static void io_parity_write_thread(struct snapraid_io* io, unsigned* pos, unsigned* waiting_map, unsigned* waiting_mac) { unsigned waiting_cycle; @@ -729,7 +761,7 @@ static void* io_writer_thread(void* arg) return 0; } -void io_start(struct snapraid_io* io, +static void io_start_thread(struct snapraid_io* io, block_off_t blockstart, block_off_t blockmax, int (*block_is_enabled)(void* arg, block_off_t), void* blockarg) { @@ -781,7 +813,7 @@ void io_start(struct snapraid_io* io, } } -void io_stop(struct snapraid_io* io) +static void io_stop_thread(struct snapraid_io* io) { unsigned i; @@ -815,3 +847,151 @@ void io_stop(struct snapraid_io* io) } } +/*****************************************************************************/ +/* global */ + +void io_init(struct snapraid_io* io, struct snapraid_state* state, + unsigned io_cache, 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*), + void (*parity_writer)(struct snapraid_worker*, struct snapraid_task*), + struct snapraid_parity_handle* parity_handle_map, unsigned parity_handle_max) +{ + unsigned i; + size_t allocated; + + io->state = state; + if (io_cache == 0) { + /* default is 8 MiB of cache */ + /* this seems to be a good tradeoff between speed and memory usage */ + io->io_max = 8 * 1024 * 1024 / state->block_size; + if (io->io_max < IO_MIN) + io->io_max = IO_MIN; + if (io->io_max > IO_MAX) + io->io_max = IO_MAX; + } else { + io->io_max = io_cache; + } + + assert(io->io_max == 1 || (io->io_max >= IO_MIN && io->io_max <= IO_MAX)); + + io->buffer_max = buffer_max; + allocated = 0; + for (i = 0; i < io->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]); + allocated += state->block_size * buffer_max; + } + + msg_progress("Using %u MiB of memory for %u blocks of IO cache.\n", (unsigned)(allocated / MEBI), io->io_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; + io->parity_base = handle_max; + io->parity_count = parity_handle_max; + + 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; + + /* 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; + + /* parity read is put after data and computed parity */ + 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; + } + + if (io_thread(io)) { + io_read_next = io_read_next_thread; + io_write_preset = io_write_preset_thread; + io_write_next = io_write_next_thread; + io_refresh = io_refresh_thread; + io_data_read = io_data_read_thread; + io_parity_read = io_parity_read_thread; + io_parity_write = io_parity_write_thread; + io_start = io_start_thread; + io_stop = io_stop_thread; + + thread_mutex_init(&io->io_mutex, 0); + thread_cond_init(&io->read_done, 0); + thread_cond_init(&io->read_sched, 0); + thread_cond_init(&io->write_done, 0); + thread_cond_init(&io->write_sched, 0); + } else { + io_read_next = io_read_next_mono; + io_write_preset = io_write_preset_mono; + io_write_next = io_write_next_mono; + io_refresh = io_refresh_mono; + io_data_read = io_data_read_mono; + io_parity_read = io_parity_read_mono; + io_parity_write = io_parity_write_mono; + io_start = io_start_mono; + io_stop = io_stop_mono; + } +} + +void io_done(struct snapraid_io* io) +{ + unsigned i; + + for (i = 0; i < io->io_max; ++i) { + free(io->buffer_map[i]); + free(io->buffer_alloc_map[i]); + } + + free(io->reader_map); + free(io->reader_list); + free(io->writer_map); + free(io->writer_list); + + if (io_thread(io)) { + thread_mutex_destroy(&io->io_mutex); + thread_cond_destroy(&io->read_done); + thread_cond_destroy(&io->read_sched); + thread_cond_destroy(&io->write_done); + thread_cond_destroy(&io->write_sched); + } +} + diff --git a/cmdline/io.h b/cmdline/io.h index b6d6bed3..82ec7448 100644 --- a/cmdline/io.h +++ b/cmdline/io.h @@ -140,6 +140,10 @@ struct snapraid_io { /** * Number of read-ahead buffers to use. + * + * Between IO_MIN and IO_MAX for thread use. + * + * If equal to 1, it means to work without any thread. */ unsigned io_max; @@ -261,6 +265,9 @@ struct snapraid_io { * * When the caller finish with the current index, * it's incremented, and a read_sched() signal is sent. + * + * In monothread mode it isn't the task index, + * but the worker index. */ unsigned reader_index; @@ -272,6 +279,9 @@ struct snapraid_io { * * When the caller finish with the current index, * it's incremented, and a write_sched() signal is sent. + * + * In monothread mode it isn't the task index, + * but the worker index. */ unsigned writer_index; @@ -303,14 +313,14 @@ void io_done(struct snapraid_io* io); /** * Start all the worker threads. */ -void io_start(struct snapraid_io* io, +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); +void (*io_stop)(struct snapraid_io* io); /** * Next read position. @@ -322,7 +332,7 @@ void io_stop(struct snapraid_io* io); * \param buffer The data buffers to use for this position. * \return The parity position. */ -block_off_t io_read_next(struct snapraid_io* io, void*** buffer); +block_off_t (*io_read_next)(struct snapraid_io* io, void*** buffer); /** * Read a data block. @@ -330,10 +340,10 @@ block_off_t io_read_next(struct snapraid_io* io, void*** buffer); * 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. + * \param diskcur The position of the data block in the ::handle_map vector. * \return The completed task. */ -struct snapraid_task* io_data_read(struct snapraid_io* io, unsigned* diskcur, unsigned* waiting_map, unsigned* waiting_mac); +struct snapraid_task* (*io_data_read)(struct snapraid_io* io, unsigned* diskcur, unsigned* waiting_map, unsigned* waiting_mac); /** * Read a parity block. @@ -341,10 +351,10 @@ struct snapraid_task* io_data_read(struct snapraid_io* io, unsigned* diskcur, un * 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. + * \param levcur The position of the parity block in the ::parity_handle_map vector. * \return The completed task. */ -struct snapraid_task* io_parity_read(struct snapraid_io* io, unsigned* levcur, unsigned* waiting_map, unsigned* waiting_mac); +struct snapraid_task* (*io_parity_read)(struct snapraid_io* io, unsigned* levcur, unsigned* waiting_map, unsigned* waiting_mac); /** * Write of a parity block. @@ -352,9 +362,21 @@ struct snapraid_task* io_parity_read(struct snapraid_io* io, unsigned* levcur, u * 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. + * \param levcur The position of the parity block in the ::parity_handle_map vector. */ -void io_parity_write(struct snapraid_io* io, unsigned* levcur, unsigned* waiting_map, unsigned* waiting_mac); +void (*io_parity_write)(struct snapraid_io* io, unsigned* levcur, unsigned* waiting_map, unsigned* waiting_mac); + +/** + * Preset the write position. + * + * This call starts the write process. + * It must be called before 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_preset)(struct snapraid_io* io, block_off_t blockcur, int skip); /** * Next write position. @@ -367,12 +389,12 @@ void io_parity_write(struct snapraid_io* io, unsigned* levcur, unsigned* waiting * \param skip Skip the writes, in case parity doesn't need to be updated. * \param writer_error Return the number of errors. Vector of IO_WRITER_ERROR_MAX elements. */ -void io_write_next(struct snapraid_io* io, block_off_t blockcur, int skip, int* writer_error); +void (*io_write_next)(struct snapraid_io* io, block_off_t blockcur, int skip, int* writer_error); /** * Refresh the number of cached blocks for all data and parity disks. */ -void io_refresh(struct snapraid_io* io); +void (*io_refresh)(struct snapraid_io* io); #endif diff --git a/cmdline/mkstream.c b/cmdline/mkstream.c index aadc7eb9..0a2b5a64 100644 --- a/cmdline/mkstream.c +++ b/cmdline/mkstream.c @@ -18,6 +18,7 @@ #include "portable.h" #include "stream.h" +#include "support.h" #define STREAM_MAX 8 #define BUFFER_MAX 64 diff --git a/cmdline/snapraid.c b/cmdline/snapraid.c index aba00ae3..eeffb5fe 100644 --- a/cmdline/snapraid.c +++ b/cmdline/snapraid.c @@ -803,7 +803,7 @@ int main(int argc, char* argv[]) break; case OPT_TEST_IO_CACHE : opt.io_cache = atoi(optarg); - if (opt.io_cache < IO_MIN || opt.io_cache > IO_MAX) { + if (opt.io_cache != 1 && (opt.io_cache < IO_MIN || opt.io_cache > IO_MAX)) { /* LCOV_EXCL_START */ log_fatal("The IO cache should be between %u and %u.\n", IO_MIN, IO_MAX); exit(EXIT_FAILURE); diff --git a/cmdline/sync.c b/cmdline/sync.c index fcc27e38..dec7c702 100644 --- a/cmdline/sync.c +++ b/cmdline/sync.c @@ -1221,6 +1221,9 @@ static int state_sync_process(struct snapraid_state* state, struct snapraid_pari /* This because we want to keep track of the time usage */ state_usage_misc(state); + /* write start */ + io_write_preset(&io, blockcur, !parity_going_to_be_updated); + /* write the parity */ for (l = 0; l < state->level; ++l) { unsigned levcur;