Ensure that writer buffers are flushed before synchronizing parity writes

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;

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
 

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
 

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

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
-

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

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