Extend io_init() to support arbitrary combinations of data and parity operations

This commit is contained in:
Andrea Mazzoleni 2026-05-21 16:59:57 +02:00
parent 185df4bd2e
commit 58fece3584
5 changed files with 122 additions and 46 deletions

View file

@ -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;

View file

@ -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

View file

@ -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.

View file

@ -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;

View file

@ -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, &copy_alloc);