From 30474cfb80695fe2db66a644b178eedd0574a822 Mon Sep 17 00:00:00 2001 From: Andrea Mazzoleni Date: Thu, 25 Sep 2025 12:46:15 +0200 Subject: [PATCH] Ensure that writer buffers are flushed before synchronizing parity writes --- cmdline/io.c | 42 +++++++++++++++++++++++++++++++++++++++++- cmdline/io.h | 41 ++++++++++++++++++++++++++++++++--------- cmdline/parity.h | 8 ++++++++ cmdline/state.c | 26 ++++++++++++++++++++++++++ cmdline/support.c | 9 ++++++++- cmdline/support.h | 1 + cmdline/sync.c | 38 ++++++++++++++------------------------ 7 files changed, 130 insertions(+), 35 deletions(-) diff --git a/cmdline/io.c b/cmdline/io.c index d8fbcd4d..7e870ba6 100644 --- a/cmdline/io.c +++ b/cmdline/io.c @@ -30,7 +30,7 @@ void (*io_parity_write)(struct snapraid_io* io, unsigned* levcur, unsigned* wait void (*io_write_preset)(struct snapraid_io* io, block_off_t blockcur, int skip) = 0; void (*io_write_next)(struct snapraid_io* io, block_off_t blockcur, int skip, int* writer_error) = 0; void (*io_refresh)(struct snapraid_io* io) = 0; - +void (*io_flush)(struct snapraid_io* io) = 0; /** * Get the next block position to process. @@ -193,6 +193,11 @@ static void io_refresh_mono(struct snapraid_io* io) (void)io; } +static void io_flush_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; @@ -529,6 +534,39 @@ static void io_refresh_thread(struct snapraid_io* io) thread_mutex_unlock(&io->io_mutex); } +static void io_flush_thread(struct snapraid_io* io) +{ + unsigned i; + + while (1) { + int all_done = 1; + + /* the synchronization is protected by the io mutex */ + thread_mutex_lock(&io->io_mutex); + + for (i = 0; i < io->writer_max; ++i) { + struct snapraid_worker* worker = &io->writer_map[i]; + + /* get the next pending task */ + unsigned next_index = (worker->index + 1) % io->io_max; + + /* if the queue of pending tasks is not empty */ + if (next_index != io->writer_index) { + all_done = 0; + break; + } + } + + thread_mutex_unlock(&io->io_mutex); + + if (all_done) + break; + + /* wait for something to complete */ + thread_yield(); + } +} + 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; @@ -983,6 +1021,7 @@ void io_init(struct snapraid_io* io, struct snapraid_state* state, io_write_preset = io_write_preset_thread; io_write_next = io_write_next_thread; io_refresh = io_refresh_thread; + io_flush = io_flush_thread; io_data_read = io_data_read_thread; io_parity_read = io_parity_read_thread; io_parity_write = io_parity_write_thread; @@ -1001,6 +1040,7 @@ void io_init(struct snapraid_io* io, struct snapraid_state* state, io_write_preset = io_write_preset_mono; io_write_next = io_write_next_mono; io_refresh = io_refresh_mono; + io_flush = io_flush_mono; io_data_read = io_data_read_mono; io_parity_read = io_parity_read_mono; io_parity_write = io_parity_write_mono; diff --git a/cmdline/io.h b/cmdline/io.h index 2a6645e3..ef07ae5b 100644 --- a/cmdline/io.h +++ b/cmdline/io.h @@ -83,8 +83,7 @@ struct snapraid_task { /** * Worker for tasks. * - * This represents a worker thread designated to read data - * from a specific disk. + * This represents a worker thread designated to read/write data for a specific disk. */ struct snapraid_worker { #if HAVE_THREAD @@ -139,7 +138,7 @@ struct snapraid_io { struct snapraid_state* state; /** - * Number of read-ahead buffers to use. + * Number of read-ahead and write-cached buffers to use. * * Between IO_MIN and IO_MAX for thread use. * @@ -257,13 +256,23 @@ struct snapraid_io { int done; /** - * The task currently used by the caller. + * The worker task index currently used by the operation in + * progress. It's the position in all the reader task_map[]. + * + * Until the operation is in progress, the data buffer cannot be + * discarded, because it's in use. * * 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. + * When the caller finish with this index, it calls read_sched(), + * and this ::reader_index is increamended, meaning that the data of + * the just finished index is not needed anymore and the buffer is + * available to be used for the following read-ahead. + * + * When this happens a read_sched signal is broadcasted, and all the + * workers can proceed using this now available index to have one more + * read-ahead buffer to load. * * In monothread mode it isn't the task index, * but the worker index. @@ -271,13 +280,22 @@ struct snapraid_io { unsigned reader_index; /** - * The task currently used by the caller. + * The worker task index currently used by the operation in + * progress. It's the position in all the writer task_map[]. + * + * Until the operation is in progress, the data buffer cannot be + * written, because the data is not yet in the buffer. * * 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. + * When the caller finish writing the data in the buffer for this + * index, it calls write_sched() and ::writer_index is incremented, + * meaning the the just finished index is scheduled to be written + * to disk. + * + * When this happens a write_sched signal is broadcasted, and all the + * workers can proceed writing the data in this index. * * In monothread mode it isn't the task index, * but the worker index. @@ -400,5 +418,10 @@ extern void (*io_write_next)(struct snapraid_io* io, block_off_t blockcur, int s */ extern void (*io_refresh)(struct snapraid_io* io); +/** + * Flush all the writes. + */ +extern void (*io_flush)(struct snapraid_io* io); + #endif diff --git a/cmdline/parity.h b/cmdline/parity.h index 41bf6af8..678ca780 100644 --- a/cmdline/parity.h +++ b/cmdline/parity.h @@ -20,6 +20,8 @@ #include "support.h" #include "bw.h" +#include "state.h" +#include "io.h" /****************************************************************************/ /* parity */ @@ -145,5 +147,11 @@ int parity_read(struct snapraid_parity_handle* handle, block_off_t pos, unsigned */ int parity_write(struct snapraid_parity_handle* handle, block_off_t pos, unsigned char* block_buffer, unsigned block_size); +/** + * Flush and sync the parity files. + */ +int state_flush(struct snapraid_state* state, struct snapraid_io* io, struct snapraid_parity_handle* parity_handle, block_off_t blockcur); + + #endif diff --git a/cmdline/state.c b/cmdline/state.c index e4a5e3db..aa1c2c85 100644 --- a/cmdline/state.c +++ b/cmdline/state.c @@ -4915,3 +4915,29 @@ void generate_configuration(const char* path) state_done(&state); } + +/** + * Flush and sync the parity files. + */ +int state_flush(struct snapraid_state* state, struct snapraid_io* io, struct snapraid_parity_handle* parity_handle, block_off_t blockcur) +{ + unsigned l; + + if (io) + io_flush(io); + + /* flush all parity handles to ensure data is written to disk */ + for (l = 0; l < state->level; ++l) { + int ret = parity_sync(&parity_handle[l]); + if (ret == -1) { + /* LCOV_EXCL_START */ + log_tag("parity_error:%u:%s: Sync error\n", blockcur, lev_config_name(l)); + log_fatal("DANGER! Unexpected sync error in %s disk.\n", lev_name(l)); + log_fatal("Ensure that disk '%s' is sane.\n", lev_config_name(l)); + return -1; + /* LCOV_EXCL_STOP */ + } + } + + return 0; +} diff --git a/cmdline/support.c b/cmdline/support.c index 4424a71c..ca792811 100644 --- a/cmdline/support.c +++ b/cmdline/support.c @@ -1741,5 +1741,12 @@ void thread_join(thread_id_t thread, void** retval) } } +void thread_yield(void) +{ +#ifdef __MINGW32__ + Sleep(0); +#else + sched_yield(); +#endif +} #endif - diff --git a/cmdline/support.h b/cmdline/support.h index 0a2ec4e7..f958c7e0 100644 --- a/cmdline/support.h +++ b/cmdline/support.h @@ -439,6 +439,7 @@ void thread_cond_signal_and_unlock(thread_cond_t* cond, thread_mutex_t* mutex); void thread_cond_broadcast_and_unlock(thread_cond_t* cond, thread_mutex_t* mutex); void thread_create(thread_id_t* thread, void* (* func)(void *), void *arg); void thread_join(thread_id_t thread, void** retval); +void thread_yield(void); #endif #endif diff --git a/cmdline/sync.c b/cmdline/sync.c index e9abc609..f45e3da1 100644 --- a/cmdline/sync.c +++ b/cmdline/sync.c @@ -1316,18 +1316,13 @@ static int state_sync_process(struct snapraid_state* state, struct snapraid_pari /* before writing the new content file we ensure that */ /* the parity is really written flushing the disk cache */ - for (l = 0; l < state->level; ++l) { - ret = parity_sync(&parity_handle[l]); - if (ret == -1) { - /* LCOV_EXCL_START */ - log_tag("parity_error:%u:%s: Sync error\n", blockcur, lev_config_name(l)); - log_fatal("DANGER! Unexpected sync error in %s disk.\n", lev_name(l)); - log_fatal("Ensure that disk '%s' is sane.\n", lev_config_name(l)); - log_fatal("Stopping at block %u\n", blockcur); - ++error; - goto bail; - /* LCOV_EXCL_STOP */ - } + ret = state_flush(state, &io, parity_handle, blockcur); + if (ret == -1) { + /* LCOV_EXCL_START */ + log_fatal("Stopping at block %u\n", blockcur); + ++error; + goto bail; + /* LCOV_EXCL_STOP */ } /* now we can safely write the content file */ @@ -1347,18 +1342,13 @@ end: /* before returning we ensure that */ /* the parity is really written flushing the disk cache */ - for (l = 0; l < state->level; ++l) { - ret = parity_sync(&parity_handle[l]); - if (ret == -1) { - /* LCOV_EXCL_START */ - log_tag("parity_error:%u:%s: Sync error\n", blockcur, lev_config_name(l)); - log_fatal("DANGER! Unexpected sync error in %s disk.\n", lev_name(l)); - log_fatal("Ensure that disk '%s' is sane.\n", lev_config_name(l)); - log_fatal("Stopping at block %u\n", blockcur); - ++error; - goto bail; - /* LCOV_EXCL_STOP */ - } + ret = state_flush(state, &io, parity_handle, blockcur); + if (ret == -1) { + /* LCOV_EXCL_START */ + log_fatal("Stopping at block %u\n", blockcur); + ++error; + goto bail; + /* LCOV_EXCL_STOP */ } if (error || silent_error || io_error) {