Add the new --test-io-cache option to control the cache size

HISTORY

@@ -4,11 +4,13 @@ SnapRAID HISTORY
 10.0 2016/01
 ===========
   * Increases the speed of the 'sync' and 'scrub' commands with a new
-    read-ahead implementation using threads. We have one thread for each
-    disk, dedicated exclusively to read data or parity in advance,
-    and to write parity later.
-    This ensures to keep the disks always busy, maximizing the data
-    throughput.
+    multithread implementation. It uses one thread for each disk, dedicated
+    exclusively to read-ahead data or parity early and to write-behind
+    parity later. This ensures to keep the disks always busy, maximizing
+    the data throughput.
+    You can control the number of blocks to cache with the option
+    --test-io-cache=NUMBER, where the number is between 3 and 128.
+    The default is 32.
 
 9.2 2016/01
 ===========

Makefile.am

@@ -230,6 +230,8 @@ if HAVE_THREAD_CHECKER
 # Other commands that uses threads
 	$(TESTENV) ./snapraid$(EXEEXT) $(CHECKFLAGS) -c $(CONF) test-rewrite
 	$(TESTENV) ./snapraid$(EXEEXT) $(CHECKFLAGS) -c $(CONF) scrub -p full
+	$(TESTENV) ./snapraid$(EXEEXT) $(CHECKFLAGS) -c $(CONF) scrub -p full -test-io-cache 3
+	$(TESTENV) ./snapraid$(EXEEXT) $(CHECKFLAGS) -c $(CONF) scrub -p full -test-io-cache 128
 else
 #### COMMAND LINE ####
 	$(MSG) Help

cmdline/dry.c

@@ -186,7 +186,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, buffermax, dry_data_reader, handle, diskmax, dry_parity_reader, 0, parity_handle, state->level);
+	io_init(&io, state, state->opt.io_cache, buffermax, dry_data_reader, handle, diskmax, dry_parity_reader, 0, parity_handle, state->level);
 
 	error = 0;
 	io_error = 0;

cmdline/io.c

@@ -38,7 +38,8 @@
  * http://stackoverflow.com/questions/4544234/calling-pthread-cond-signal-without-locking-mutex/4544494#4544494
  */
 
-void io_init(struct snapraid_io* io, struct snapraid_state* state, unsigned buffer_max,
+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,
 	void (*parity_reader)(struct snapraid_worker*, struct snapraid_task*),
@@ -55,10 +56,16 @@ void io_init(struct snapraid_io* io, struct snapraid_state* state, unsigned buff
 	pthread_cond_init(&io->write_sched, 0);
 
 	io->state = state;
+	if (io_cache == 0)
+		io->io_max = 32;
+	else
+		io->io_max = io_cache;
+
+	assert(io->io_max >= IO_MIN && io->io_max <= IO_MAX);
 
 	io->buffer_max = buffer_max;
 	allocated = 0;
-	for (i = 0; i < IO_MAX; ++i) {
+	for (i = 0; i < io->io_max; ++i) {
 		io->buffer_map[i] = malloc_nofail_vector_align(handle_max, buffer_max, state->block_size, &io->buffer_alloc_map[i]);
 		if (!state->opt.skip_self)
 			mtest_vector(io->buffer_max, state->block_size, io->buffer_map[i]);
@@ -128,7 +135,7 @@ void io_done(struct snapraid_io* io)
 {
 	unsigned i;
 
-	for (i = 0; i < IO_MAX; ++i) {
+	for (i = 0; i < io->io_max; ++i) {
 		free(io->buffer_map[i]);
 		free(io->buffer_alloc_map[i]);
 	}
@@ -269,7 +276,7 @@ static struct snapraid_task* io_reader_step(struct snapraid_worker* worker)
 		}
 
 		/* get the next pending task */
-		next_index = (worker->index + 1) % IO_MAX;
+		next_index = (worker->index + 1) % io->io_max;
 
 		/* if the queue of pending tasks is not empty */
 		if (next_index != io->reader_index) {
@@ -331,14 +338,14 @@ static struct snapraid_task* io_writer_step(struct snapraid_worker* worker, int
 		unsigned next_index;
 
 		/* get the next pending task */
-		next_index = (worker->index + 1) % IO_MAX;
+		next_index = (worker->index + 1) % io->io_max;
 
 		/* if the queue of pending tasks is not empty */
 		if (next_index != io->writer_index) {
 			struct snapraid_task* task;
 		
 			/* the index that the IO may be waiting for */
-			unsigned waiting_index = (io->writer_index + 1) % IO_MAX;
+			unsigned waiting_index = (io->writer_index + 1) % io->io_max;
 
 			/* the index that worker just completed */
 			unsigned done_index = worker->index;
@@ -406,7 +413,7 @@ block_off_t io_read_next(struct snapraid_io* io, void*** buffer)
 	io_reader_sched(io, io->reader_index, blockcur_schedule);
 
 	/* set the index for the tasks to return to the caller */
-	io->reader_index = (io->reader_index + 1) % IO_MAX;
+	io->reader_index = (io->reader_index + 1) % io->io_max;
 
 	/* get the position to operate at high level from one task */
 	blockcur_caller = io->reader_map[0].task_map[io->reader_index].position;
@@ -462,7 +469,7 @@ void io_write_next(struct snapraid_io* io, unsigned blockcur, int skip, int* wri
 	assert(io->writer_index == io->reader_index);
 
 	/* set the index to be used for the next write */
-	io->writer_index = (io->writer_index + 1) % IO_MAX;
+	io->writer_index = (io->writer_index + 1) % io->io_max;
 
 #ifndef CHECKER
 	/* without the thread checker unlock before signaling, */
@@ -559,7 +566,7 @@ void io_parity_write(struct snapraid_io* io, unsigned* pos)
 		/* to avoid a concurrent access */
 		/* note that we are already sure that a write is not in progress */
 		/* at the index the IO is using at now */
-		busy_index = (io->writer_index + 1) % IO_MAX;
+		busy_index = (io->writer_index + 1) % io->io_max;
 
 		/* search for a worker that has already finished */
 		let = &io->writer_list[0];
@@ -687,7 +694,7 @@ void io_start(struct snapraid_io* io,
 	io->block_next = blockstart;
 
 	io->done = 0;
-	io->reader_index = IO_MAX - 1;
+	io->reader_index = io->io_max - 1;
 	io->writer_index = 0;
 
 	/* clear writer errors */
@@ -696,7 +703,7 @@ void io_start(struct snapraid_io* io,
 
 	/* setup the initial read pending tasks, except the latest one, */
 	/* the latest will be initialized at the fist io_read_next() call */
-	for (i = 0; i < IO_MAX - 1; ++i) {
+	for (i = 0; i < io->io_max - 1; ++i) {
 		block_off_t blockcur = io_position_next(io);
 
 		io_reader_sched(io, i, blockcur);
@@ -725,7 +732,7 @@ void io_start(struct snapraid_io* io,
 	for (i = 0; i < io->writer_max; ++i) {
 		struct snapraid_worker* worker = &io->writer_map[i];
 
-		worker->index = IO_MAX - 1;
+		worker->index = io->io_max - 1;
 
 		if (pthread_create(&worker->thread, 0, io_writer_thread, worker) != 0) {
 			/* LCOV_EXCL_START */

cmdline/io.h

@@ -39,7 +39,8 @@
  * 64  - 524 MiB/s, CPU 65%, speed 137%
  * 128 - 525 MiB/s, CPU 66%, speed 138%
  */
-#define IO_MAX 32
+#define IO_MIN 3 /* required by writers, readers can work also with 2 */
+#define IO_MAX 128
 
 /**
  * State of the task.
@@ -133,6 +134,13 @@ struct snapraid_worker {
  * data from the disks.
  */
 struct snapraid_io {
+	struct snapraid_state* state;
+
+	/**
+	 * Number of read-ahead buffers to use.
+	 */
+	unsigned io_max;
+
 	/**
 	 * Mutex used to protect the synchronization
 	 * between the io and the workers.
@@ -175,8 +183,6 @@ struct snapraid_io {
 	 */
 	pthread_cond_t write_sched;
 
-	struct snapraid_state* state;
-
 	/**
 	 * Base position for workers.
 	 *
@@ -246,7 +252,7 @@ struct snapraid_io {
 	/**
 	 * The task currently used by the caller.
 	 *
-	 * It's a rolling counter, when reaching IO_MAX
+	 * It's a rolling counter, when reaching ::io_max
 	 * it goes again to 0.
 	 *
 	 * When the caller finish with the current index,
@@ -257,7 +263,7 @@ struct snapraid_io {
 	/**
 	 * The task currently used by the caller.
 	 *
-	 * It's a rolling counter, when reaching IO_MAX
+	 * It's a rolling counter, when reaching ::io_max
 	 * it goes again to 0.
 	 *
 	 * When the caller finish with the current index,
@@ -274,10 +280,11 @@ struct snapraid_io {
 /**
  * Initialize the InputOutput workers.
  *
+ * \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.
  */
 void io_init(struct snapraid_io* io, struct snapraid_state* state,
-	unsigned buffer_max,
+	unsigned io_cache, unsigned buffer_max,
 	void (*data_reader)(struct snapraid_worker*, struct snapraid_task*),
 	struct snapraid_handle* handle_map, unsigned handle_max,
 	void (*parity_reader)(struct snapraid_worker*, struct snapraid_task*),

cmdline/scrub.c

@@ -275,7 +275,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, buffermax, scrub_data_reader, handle, diskmax, scrub_parity_reader, 0, parity_handle, state->level);
+	io_init(&io, state, state->opt.io_cache, buffermax, scrub_data_reader, handle, diskmax, scrub_parity_reader, 0, parity_handle, state->level);
 
 	error = 0;
 	silent_error = 0;

cmdline/snapraid.c

@@ -23,6 +23,7 @@
 #include "import.h"
 #include "search.h"
 #include "state.h"
+#include "io.h"
 #include "raid/raid.h"
 
 /****************************************************************************/
@@ -249,6 +250,7 @@ void config(char* conf, size_t conf_size, const char* argv0)
 #define OPT_TEST_FAKE_UUID 287
 #define OPT_TEST_MATCH_FIRST_UUID 288
 #define OPT_TEST_FORCE_PARITY_UPDATE 289
+#define OPT_TEST_IO_CACHE 290
 
 #if HAVE_GETOPT_LONG
 struct option long_options[] = {
@@ -376,6 +378,9 @@ struct option long_options[] = {
 	/* Force parity update even if all the data is not changed */
 	{ "test-force-parity-update", 0, 0, OPT_TEST_FORCE_PARITY_UPDATE },
 
+	/* Number of read-ahead buffers */
+	{ "test-io-cache", 1, 0, OPT_TEST_IO_CACHE },
+
 	{ 0, 0, 0, 0 }
 };
 #endif
@@ -756,6 +761,15 @@ int main(int argc, char* argv[])
 		case OPT_TEST_FORCE_PARITY_UPDATE :
 			opt.force_parity_update = 1;
 			break;
+		case OPT_TEST_IO_CACHE :
+			opt.io_cache = atoi(optarg);
+			if (opt.io_cache < IO_MIN || opt.io_cache > IO_MAX) {
+				/* LCOV_EXCL_START */
+				log_fatal("The IO cache should be between %u and %u.\n", IO_MIN, IO_MAX);
+				exit(EXIT_FAILURE);
+				/* LCOV_EXCL_STOP */
+			}
+			break;
 		default :
 			/* LCOV_EXCL_START */
 			log_fatal("Unknown option '%c'\n", (char)c);

cmdline/state.h

@@ -108,6 +108,7 @@ struct snapraid_option {
 	int fake_uuid; /**< Set fakes UUID for testing. */
 	int match_first_uuid; /**< Force the matching of the first UUID. */
 	int force_parity_update; /**< Force parity update even if data is not changed. */
+	unsigned io_cache; /**< Number of IO buffers to use. 0 for default. */
 };
 
 /**

cmdline/sync.c

@@ -695,7 +695,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, buffermax, sync_data_reader, handle, diskmax, 0, sync_parity_writer, parity_handle, state->level);
+	io_init(&io, state, state->opt.io_cache, buffermax, sync_data_reader, handle, diskmax, 0, sync_parity_writer, parity_handle, state->level);
 
 	/* allocate the copy buffer */
 	copy = malloc_nofail_vector_align(diskmax, diskmax, state->block_size, &copy_alloc);