From 58fece3584e1e5a474641799e1900758e6251fa0 Mon Sep 17 00:00:00 2001 From: Andrea Mazzoleni Date: Thu, 21 May 2026 16:59:57 +0200 Subject: [PATCH] Extend io_init() to support arbitrary combinations of data and parity operations --- cmdline/dry.c | 2 +- cmdline/io.c | 133 ++++++++++++++++++++++++++++++++++-------------- cmdline/io.h | 29 +++++++++-- cmdline/scrub.c | 2 +- cmdline/sync.c | 2 +- 5 files changed, 122 insertions(+), 46 deletions(-) diff --git a/cmdline/dry.c b/cmdline/dry.c index 46bd4774..2db9a14a 100644 --- a/cmdline/dry.c +++ b/cmdline/dry.c @@ -167,7 +167,7 @@ static int state_dry_process(struct snapraid_state* state, struct snapraid_parit buffermax = diskmax + 2 * state->level; /* initialize the io threads */ - io_init(&io, state, state->opt.io_cache, buffermax, dry_data_reader, handle, diskmax, dry_parity_reader, 0, parity_handle, state->level); + io_init(&io, state, state->opt.io_cache, buffermax, handle, diskmax, 0, dry_data_reader, 0, parity_handle, state->level, 0, dry_parity_reader, 0); /* possibly waiting disks */ waiting_mac = diskmax > RAID_PARITY_MAX ? diskmax : RAID_PARITY_MAX; diff --git a/cmdline/io.c b/cmdline/io.c index fa2666b5..d0fa8707 100644 --- a/cmdline/io.c +++ b/cmdline/io.c @@ -911,15 +911,24 @@ static void io_stop_thread(struct snapraid_io* io) 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, + const io_op_t* data_ops, + void (*data_reader)(struct snapraid_worker*, struct snapraid_task*), + void (*data_writer)(struct snapraid_worker*, struct snapraid_task*), + struct snapraid_parity_handle* parity_handle_map, unsigned parity_handle_max, + const io_op_t* parity_ops, 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) + void (*parity_writer)(struct snapraid_worker*, struct snapraid_task*)) { unsigned i; size_t allocated_size; size_t block_size = state->block_size; + unsigned readers_count = 0; + unsigned writers_count = 0; + unsigned r_idx = 0; + unsigned w_idx = 0; + io_op_t data_op_default = IO_OP_NONE; + io_op_t parity_op_default = IO_OP_NONE; io->state = state; @@ -971,13 +980,38 @@ void io_init(struct snapraid_io* io, struct snapraid_state* state, msg_progress("Using %u MiB of memory for %u cached blocks.\n", (unsigned)(allocated_size / 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; + /* resolve implicit fallback directions if ops are NULL */ + if (data_reader && !data_writer) { + data_op_default = IO_OP_READ; } + if (!data_reader && data_writer) { + data_op_default = IO_OP_WRITE; + } + if (parity_reader && !parity_writer) { + parity_op_default = IO_OP_READ; + } + if (!parity_reader && parity_writer) { + parity_op_default = IO_OP_WRITE; + } + + /* count reader and writer workers */ + for (i = 0; i < handle_max; ++i) { + io_op_t op = data_ops ? data_ops[i] : data_op_default; + if (op == IO_OP_READ) + readers_count++; + if (op == IO_OP_WRITE) + writers_count++; + } + for (i = 0; i < parity_handle_max; ++i) { + io_op_t op = parity_ops ? parity_ops[i] : parity_op_default; + if (op == IO_OP_READ) + readers_count++; + if (op == IO_OP_WRITE) + writers_count++; + } + + io->reader_max = readers_count; + io->writer_max = writers_count; io->reader_map = malloc_nofail(sizeof(struct snapraid_worker) * io->reader_max); io->reader_list = malloc_nofail(sizeof(unsigned) * (io->reader_max + 1)); @@ -985,46 +1019,67 @@ void io_init(struct snapraid_io* io, struct snapraid_state* state, io->writer_list = malloc_nofail(sizeof(unsigned) * (io->writer_max + 1)); io->data_base = 0; - io->data_count = handle_max; - io->parity_base = handle_max; - io->parity_count = parity_handle_max; + io->data_count = 0; + for (i = 0; i < handle_max; ++i) { + io_op_t op = data_ops ? data_ops[i] : data_op_default; + if (op == IO_OP_READ) + io->data_count++; + } - for (i = 0; i < io->reader_max; ++i) { - struct snapraid_worker* worker = &io->reader_map[i]; + io->parity_base = io->data_count; + io->parity_count = 0; + for (i = 0; i < parity_handle_max; ++i) { + io_op_t op = parity_ops ? parity_ops[i] : parity_op_default; + if (op == IO_OP_READ) + io->parity_count++; + } - worker->io = io; - - if (i < handle_max) { - /* it's a data read */ + /* configure reader map workers */ + for (i = 0; i < handle_max; ++i) { + io_op_t op = data_ops ? data_ops[i] : data_op_default; + if (op == IO_OP_READ) { + struct snapraid_worker* worker = &io->reader_map[r_idx++]; + worker->io = io; 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->buffer_skew = i - (r_idx - 1); + } + } + for (i = 0; i < parity_handle_max; ++i) { + io_op_t op = parity_ops ? parity_ops[i] : parity_op_default; + if (op == IO_OP_READ) { + struct snapraid_worker* worker = &io->reader_map[r_idx++]; + worker->io = io; worker->handle = 0; - worker->parity_handle = &parity_handle_map[i - handle_max]; + worker->parity_handle = &parity_handle_map[i]; worker->func = parity_reader; - - /* parity read is put after data and computed parity */ - worker->buffer_skew = parity_handle_max; + worker->buffer_skew = (handle_max + parity_handle_max + i) - (r_idx - 1); } } - 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; + /* configure writer map workers */ + for (i = 0; i < handle_max; ++i) { + io_op_t op = data_ops ? data_ops[i] : data_op_default; + if (op == IO_OP_WRITE) { + struct snapraid_worker* worker = &io->writer_map[w_idx++]; + worker->io = io; + worker->handle = &handle_map[i]; + worker->parity_handle = 0; + worker->func = data_writer; + worker->buffer_skew = i - (w_idx - 1); + } + } + for (i = 0; i < parity_handle_max; ++i) { + io_op_t op = parity_ops ? parity_ops[i] : parity_op_default; + if (op == IO_OP_WRITE) { + struct snapraid_worker* worker = &io->writer_map[w_idx++]; + worker->io = io; + worker->handle = 0; + worker->parity_handle = &parity_handle_map[i]; + worker->func = parity_writer; + worker->buffer_skew = (handle_max + i) - (w_idx - 1); + } } #if HAVE_THREAD diff --git a/cmdline/io.h b/cmdline/io.h index b22904c2..771654f1 100644 --- a/cmdline/io.h +++ b/cmdline/io.h @@ -360,19 +360,40 @@ struct snapraid_io { struct snapraid_bw bw; }; +typedef enum { + IO_OP_NONE = 0, /**< Disk is not accessed. */ + IO_OP_READ = 1, /**< Disk is read. */ + IO_OP_WRITE = 2 /**< Disk is written. */ +} io_op_t; + /** - * Initialize the InputOutput workers. + * Initialize the InputOutput workers with granular directions. * + * \param io The InputOutput context to initialize. + * \param state The global program state. * \param io_cache The number of IO buffers for read-ahead and write-behind. 0 for default. * \param buffer_max The number of data/parity buffers to allocate. + * \param handle_map The map of data disk handles. + * \param handle_max The total number of data disk handles. + * \param data_ops Granular operations for each data disk, or NULL for implicit defaults. + * \param data_reader Callback function pointer for data read tasks. + * \param data_writer Callback function pointer for data write tasks. + * \param parity_handle_map The map of parity disk handles. + * \param parity_handle_max The total number of parity levels/handles. + * \param parity_ops Granular operations for each parity level, or NULL for implicit defaults. + * \param parity_reader Callback function pointer for parity read tasks. + * \param parity_writer Callback function pointer for parity write tasks. */ 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, + const io_op_t* data_ops, + void (*data_reader)(struct snapraid_worker*, struct snapraid_task*), + void (*data_writer)(struct snapraid_worker*, struct snapraid_task*), + struct snapraid_parity_handle* parity_handle_map, unsigned parity_handle_max, + const io_op_t* parity_ops, 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); + void (*parity_writer)(struct snapraid_worker*, struct snapraid_task*)); /** * Deinitialize the InputOutput workers. diff --git a/cmdline/scrub.c b/cmdline/scrub.c index 871e3075..80db550d 100644 --- a/cmdline/scrub.c +++ b/cmdline/scrub.c @@ -271,7 +271,7 @@ static int state_scrub_process(struct snapraid_state* state, struct snapraid_par buffermax = diskmax + 2 * state->level; /* initialize the io threads */ - io_init(&io, state, state->opt.io_cache, buffermax, scrub_data_reader, handle, diskmax, scrub_parity_reader, 0, parity_handle, state->level); + io_init(&io, state, state->opt.io_cache, buffermax, handle, diskmax, 0, scrub_data_reader, 0, parity_handle, state->level, 0, scrub_parity_reader, 0); /* possibly waiting disks */ waiting_mac = diskmax > RAID_PARITY_MAX ? diskmax : RAID_PARITY_MAX; diff --git a/cmdline/sync.c b/cmdline/sync.c index 8ccf7ae8..045119f3 100644 --- a/cmdline/sync.c +++ b/cmdline/sync.c @@ -683,7 +683,7 @@ static int state_sync_process(struct snapraid_state* state, struct snapraid_pari buffermax = diskmax + state->level; /* initialize the io threads */ - io_init(&io, state, state->opt.io_cache, buffermax, sync_data_reader, handle, diskmax, 0, sync_parity_writer, parity_handle, state->level); + io_init(&io, state, state->opt.io_cache, buffermax, handle, diskmax, 0, sync_data_reader, 0, parity_handle, state->level, 0, 0, sync_parity_writer); /* allocate the copy buffer */ copy = malloc_nofail_vector_align(diskmax, diskmax, state->block_size, ©_alloc);