Quiesce read-ahead before thermal spin-down

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;

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
 

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

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

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