diff --git a/cmdline/io.c b/cmdline/io.c index a84e9579..c2f0b8e9 100644 --- a/cmdline/io.c +++ b/cmdline/io.c @@ -17,6 +17,7 @@ void (*io_write_preset)(struct snapraid_io* io, block_off_t blockcur, int skip) 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, int* writer_error) = 0; +void (*io_quiesce)(struct snapraid_io* io) = 0; /** * Get the next block position to process. @@ -218,6 +219,11 @@ static void io_flush_mono(struct snapraid_io* io, int* writer_error) io_writer_error_get(io, writer_error); } +static void io_quiesce_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; @@ -316,6 +322,9 @@ static struct snapraid_task* io_reader_step(struct snapraid_worker* worker) /* the synchronization is protected by the io mutex */ thread_mutex_lock(&io->io_mutex); + /* acknowledge completion of the previous task */ + worker->busy = 0; + while (1) { unsigned next_index; @@ -344,6 +353,7 @@ static struct snapraid_task* io_reader_step(struct snapraid_worker* worker) /* get the new working task */ worker->index = next_index; task = &worker->task_map[worker->index]; + worker->busy = 1; /* if the just completed task is at this index */ if (done_index == waiting_index) { @@ -608,6 +618,44 @@ static void io_flush_thread(struct snapraid_io* io, int* writer_error) } } +static void io_quiesce_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->reader_max; ++i) { + struct snapraid_worker* worker = &io->reader_map[i]; + unsigned next_index = (worker->index + 1) % io->io_max; + + /* + * busy covers the task already dequeued and executing outside + * the mutex. A different next index identifies scheduled work + * not yet dequeued. Completed but unconsumed tasks need neither. + * When stopping, pending tasks cannot start and don't need draining. + */ + if (worker->busy || (!io->done && next_index != io->reader_index)) { + all_done = 0; + break; + } + } + + if (all_done) { + thread_mutex_unlock(&io->io_mutex); + return; + } + + thread_mutex_unlock(&io->io_mutex); + + /* wait for a reader 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; @@ -902,6 +950,7 @@ static void io_start_thread(struct snapraid_io* io, struct snapraid_worker* worker = &io->reader_map[i]; worker->index = 0; + worker->busy = 1; thread_create(&worker->thread, io_reader_thread, worker); } @@ -1136,6 +1185,7 @@ void io_init(struct snapraid_io* io, struct snapraid_state* state, io_write_next = io_write_next_thread; io_refresh = io_refresh_thread; io_flush = io_flush_thread; + io_quiesce = io_quiesce_thread; io_data_read = io_data_read_thread; io_parity_read = io_parity_read_thread; io_parity_write = io_parity_write_thread; @@ -1155,6 +1205,7 @@ void io_init(struct snapraid_io* io, struct snapraid_state* state, io_write_next = io_write_next_mono; io_refresh = io_refresh_mono; io_flush = io_flush_mono; + io_quiesce = io_quiesce_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 f6565835..9b9c6703 100644 --- a/cmdline/io.h +++ b/cmdline/io.h @@ -74,7 +74,7 @@ struct snapraid_task { struct snapraid_worker { #if HAVE_THREAD thread_id_t thread; /**< Thread context for the worker. */ - int busy; /**< A task is currently being processed. */ + int busy; /**< A read or write task is currently being processed. */ #endif /** @@ -509,5 +509,12 @@ extern void (*io_refresh)(struct snapraid_io* io); */ extern void (*io_flush)(struct snapraid_io* io, int* writer_error); +/** + * Wait for all scheduled reads to complete. + * + * Completed read-ahead results are retained for later consumption. + */ +extern void (*io_quiesce)(struct snapraid_io* io); + #endif diff --git a/cmdline/parity.h b/cmdline/parity.h index 0b9750d8..77feb161 100644 --- a/cmdline/parity.h +++ b/cmdline/parity.h @@ -134,9 +134,13 @@ 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. + * Complete all pending I/O and sync the parity files. + * + * This waits for all parity writes, reports their errors, and flushes the + * parity data to disk. It also waits for all scheduled read-ahead to complete + * without consuming the results, which remain available to the caller. */ -int state_flush(struct snapraid_state* state, struct snapraid_io* io, struct snapraid_parity_handle* parity_handle, block_off_t blockcur); +int state_barrier(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 ff1c923d..439a72d5 100644 --- a/cmdline/state.c +++ b/cmdline/state.c @@ -5872,9 +5872,12 @@ void generate_configuration(const char* path) /** - * Flush and sync the parity files. + * Establish a durable I/O barrier: drain asynchronous writes, collect their + * errors, sync parity only if all writes succeeded, then quiesce read-ahead. + * A failed writer invalidates the whole barrier because parity must not be + * treated as coherent, or recorded in content state, after a failed write. */ -int state_flush(struct snapraid_state* state, struct snapraid_io* io, struct snapraid_parity_handle* parity_handle, block_off_t blockcur) +int state_barrier(struct snapraid_state* state, struct snapraid_io* io, struct snapraid_parity_handle* parity_handle, block_off_t blockcur) { unsigned l; @@ -5904,6 +5907,10 @@ int state_flush(struct snapraid_state* state, struct snapraid_io* io, struct sna } } + /* complete all read-ahead without consuming the results */ + if (io) + io_quiesce(io); + return 0; } diff --git a/cmdline/sync.c b/cmdline/sync.c index 3a35add1..24a8e119 100644 --- a/cmdline/sync.c +++ b/cmdline/sync.c @@ -1362,7 +1362,7 @@ static int state_sync_process(struct snapraid_state* state, struct snapraid_pari state_progress_stop(state); /* before spinning down flush all the caches */ - ret = state_flush(state, &io, parity_handle, blockcur); + ret = state_barrier(state, &io, parity_handle, blockcur); if (ret == -1) { /* LCOV_EXCL_START */ log_fatal(errno, "Stopping at block %" PRIu64 "\n", blockcur); @@ -1399,7 +1399,7 @@ 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 */ - ret = state_flush(state, &io, parity_handle, blockcur); + ret = state_barrier(state, &io, parity_handle, blockcur); if (ret == -1) { /* LCOV_EXCL_START */ log_fatal(EIO, "Stopping at block %" PRIu64 "\n", blockcur); @@ -1425,7 +1425,7 @@ end: * Before returning we ensure that * the parity is really written flushing the disk cache */ - ret = state_flush(state, &io, parity_handle, blockcur); + ret = state_barrier(state, &io, parity_handle, blockcur); if (ret == -1) { /* LCOV_EXCL_START */ log_fatal(errno, "Stopping at block %" PRIu64 "\n", blockcur);