From 4680b887b581b6397283685e7ebc644ebb5b329b Mon Sep 17 00:00:00 2001
From: Jens Axboe <axboe@suse.de>
Date: Tue, 6 Dec 2005 12:53:48 +0100
Subject: [PATCH] [PATCH] fio: implement syncio on top of the aio framework

---
 Makefile              |   2 +-
 README.fio            |   1 -
 fio-ini.c             |  18 +--
 fio-aio.c => fio-io.c | 201 +++++++++++++++++++++++++------
 fio.c                 | 267 +++++++++---------------------------------
 fio.h                 |  12 +-
 os.h                  |   4 +-
 7 files changed, 235 insertions(+), 270 deletions(-)
 rename fio-aio.c => fio-io.c (58%)

diff --git a/Makefile b/Makefile
index 53f21e9..30eb818 100644
--- a/Makefile
+++ b/Makefile
@@ -8,7 +8,7 @@ all: depend $(PROGS) $(SCRIPTS)
 dops: dops.o
 	$(CC) $(CFLAGS) -o $@ $(filter %.o,$^) -laio
 
-fio: fio.o fio-aio.o fio-ini.o md5.o crc32.o
+fio: fio.o fio-io.o fio-ini.o md5.o crc32.o
 	$(CC) $(CFLAGS) -o $@ $(filter %.o,$^) -lpthread -laio -lm -lrt
 
 sgioread: sgioread.o
diff --git a/README.fio b/README.fio
index b7e2ca9..361e074 100644
--- a/README.fio
+++ b/README.fio
@@ -56,7 +56,6 @@ The <jobs> format is as follows:
 			use crc32 for verifies.
 	stonewall	Wait for preceeding jobs to end before running.
 	numjobs=x	Create 'x' similar entries for this job
-	mmap=x		If 'x', use mmap for data transfers
 
 
 Examples using a job file
diff --git a/fio-ini.c b/fio-ini.c
index de82a88..4a75d65 100644
--- a/fio-ini.c
+++ b/fio-ini.c
@@ -32,7 +32,6 @@
 #define DEF_STONEWALL	(0)
 #define DEF_NUMJOBS	(1)
 #define DEF_USE_THREAD	(0)
-#define DEF_USE_MMAP	(0)
 #define DEF_FILE_SIZE	(1024 * 1024 * 1024UL)
 
 static int repeatable = DEF_RAND_REPEAT;
@@ -142,7 +141,6 @@ static struct thread_data *get_new_job(int global, struct thread_data *parent)
 	td->stonewall = parent->stonewall;
 	td->numjobs = parent->numjobs;
 	td->use_thread = parent->use_thread;
-	td->use_mmap = parent->use_mmap;
 	memcpy(&td->cpumask, &parent->cpumask, sizeof(td->cpumask));
 
 	return td;
@@ -191,7 +189,7 @@ static int add_job(struct thread_data *td, const char *jobname, int prioclass,
 	} else
 		strcpy(td->file_name, jobname);
 
-	sem_init(&td->mutex, 1, 0);
+	sem_init(&td->mutex, 0, 0);
 
 	td->clat_stat.min_val = ULONG_MAX;
 	td->slat_stat.min_val = ULONG_MAX;
@@ -199,11 +197,9 @@ static int add_job(struct thread_data *td, const char *jobname, int prioclass,
 
 	run_str[td->thread_number - 1] = 'P';
 
-	if (td->io_engine != FIO_SYNCIO) {
+	if ((td->io_engine & FIO_SYNCIO) == 0) {
 		if (!td->aio_depth)
 			td->aio_depth = 1;
-		if (td->use_mmap)
-			td->use_mmap = 0;
 	}
 
 	if (td->min_bs == -1U)
@@ -226,7 +222,7 @@ static int add_job(struct thread_data *td, const char *jobname, int prioclass,
 	if (write_bw_log)
 		setup_log(&td->bw_log);
 
-	printf("Client%d (g=%d): rw=%d, prio=%d/%d, seq=%d, odir=%d, mmap=%d, bs=%d-%d, rate=%d, ioengine=%d, aio_depth=%d\n", td->thread_number, td->groupid, td->ddir, prioclass, prio, td->sequential, td->odirect, td->use_mmap, td->min_bs, td->max_bs, td->rate, td->io_engine, td->aio_depth);
+	printf("Client%d (g=%d): rw=%d, prio=%d/%d, seq=%d, odir=%d, bs=%d-%d, rate=%d, ioengine=%d, aio_depth=%d\n", td->thread_number, td->groupid, td->ddir, prioclass, prio, td->sequential, td->odirect, td->min_bs, td->max_bs, td->rate, td->io_engine, td->aio_depth);
 
 	/*
 	 * recurse add identical jobs, clear numjobs and stonewall options
@@ -540,6 +536,9 @@ static int str_ioengine_cb(struct thread_data *td, char *str)
 	} else if (!strncmp(str, "sync", 4)) {
 		td->io_engine = FIO_SYNCIO;
 		return 0;
+	} else if (!strncmp(str, "mmap", 4)) {
+		td->io_engine = FIO_MMAPIO;
+		return 0;
 	}
 
 	fprintf(stderr, "bad ioengine type: %s\n", str);
@@ -679,10 +678,6 @@ int parse_jobs_ini(char *file)
 				fgetpos(f, &off);
 				continue;
 			}
-			if (!check_int(p, "mmap", &td->use_mmap)) {
-				fgetpos(f, &off);
-				continue;
-			}
 			if (!check_int(p, "overwrite", &td->overwrite)) {
 				fgetpos(f, &off);
 				continue;
@@ -804,7 +799,6 @@ static int fill_def_thread(void)
 	def_thread.stonewall = DEF_STONEWALL;
 	def_thread.numjobs = DEF_NUMJOBS;
 	def_thread.use_thread = DEF_USE_THREAD;
-	def_thread.use_mmap = DEF_USE_MMAP;
 #ifdef FIO_HAVE_DISK_UTIL
 	def_thread.do_disk_util = 1;
 #endif
diff --git a/fio-aio.c b/fio-io.c
similarity index 58%
rename from fio-aio.c
rename to fio-io.c
index 040bd08..4e8a880 100644
--- a/fio-aio.c
+++ b/fio-io.c
@@ -2,6 +2,7 @@
 #include <stdlib.h>
 #include <unistd.h>
 #include <errno.h>
+#include <sys/mman.h>
 #include "fio.h"
 #include "os.h"
 
@@ -9,18 +10,25 @@
 
 #define ev_to_iou(ev)	(struct io_u *) ((unsigned long) (ev)->obj)
 
+static int fio_io_sync(struct thread_data *td)
+{
+	return fsync(td->fd);
+}
+
 struct libaio_data {
 	io_context_t aio_ctx;
 	struct io_event *aio_events;
 };
 
-static void fio_libaio_io_prep(struct thread_data *td, struct io_u *io_u,
+static int fio_libaio_io_prep(struct thread_data *td, struct io_u *io_u,
 			       int read)
 {
 	if (read)
 		io_prep_pread(&io_u->iocb, td->fd, io_u->buf, io_u->buflen, io_u->offset);
 	else
 		io_prep_pwrite(&io_u->iocb, td->fd, io_u->buf, io_u->buflen, io_u->offset);
+
+	return 0;
 }
 
 static struct io_u *fio_libaio_event(struct thread_data *td, int event)
@@ -75,6 +83,20 @@ static int fio_libaio_cancel(struct thread_data *td, struct io_u *io_u)
 	return io_cancel(ld->aio_ctx, &io_u->iocb, ld->aio_events);
 }
 
+static void fio_libaio_cleanup(struct thread_data *td)
+{
+	struct libaio_data *ld = td->aio_data;
+
+	if (ld) {
+		io_destroy(ld->aio_ctx);
+		if (ld->aio_events)
+			free(ld->aio_events);
+
+		free(ld);
+		td->aio_data = NULL;
+	}
+}
+
 int fio_libaio_init(struct thread_data *td)
 {
 	struct libaio_data *ld = malloc(sizeof(*ld));
@@ -90,26 +112,14 @@ int fio_libaio_init(struct thread_data *td)
 	td->io_getevents = fio_libaio_getevents;
 	td->io_event = fio_libaio_event;
 	td->io_cancel = fio_libaio_cancel;
+	td->io_cleanup = fio_libaio_cleanup;
+	td->io_sync = fio_io_sync;
 
 	ld->aio_events = malloc(td->aio_depth * sizeof(struct io_event));
 	td->aio_data = ld;
 	return 0;
 }
 
-void fio_libaio_cleanup(struct thread_data *td)
-{
-	struct libaio_data *ld = td->aio_data;
-
-	if (ld) {
-		io_destroy(ld->aio_ctx);
-		if (ld->aio_events)
-			free(ld->aio_events);
-
-		free(ld);
-		td->aio_data = NULL;
-	}
-}
-
 #else /* FIO_HAVE_LIBAIO */
 
 int fio_libaio_init(struct thread_data *td)
@@ -117,10 +127,6 @@ int fio_libaio_init(struct thread_data *td)
 	return EINVAL;
 }
 
-void fio_libaio_cleanup(struct thread_data *td)
-{
-}
-
 #endif /* FIO_HAVE_LIBAIO */
 
 #ifdef FIO_HAVE_POSIXAIO
@@ -139,7 +145,7 @@ static int fio_posixaio_cancel(struct thread_data *td, struct io_u *io_u)
 	return 1;
 }
 
-static void fio_posixaio_prep(struct thread_data *td, struct io_u *io_u,
+static int fio_posixaio_prep(struct thread_data *td, struct io_u *io_u,
 			      int read)
 {
 	struct aiocb *aiocb = &io_u->aiocb;
@@ -149,12 +155,8 @@ static void fio_posixaio_prep(struct thread_data *td, struct io_u *io_u,
 	aiocb->aio_nbytes = io_u->buflen;
 	aiocb->aio_offset = io_u->offset;
 
-	if (read)
-		aiocb->aio_lio_opcode = LIO_READ;
-	else
-		aiocb->aio_lio_opcode = LIO_WRITE;
-
 	io_u->seen = 0;
+	return 0;
 }
 
 static int fio_posixaio_getevents(struct thread_data *td, int min, int max,
@@ -202,11 +204,28 @@ static struct io_u *fio_posixaio_event(struct thread_data *td, int event)
 static int fio_posixaio_queue(struct thread_data *td, struct io_u *io_u)
 {
 	struct aiocb *aiocb = &io_u->aiocb;
+	int ret;
 
-	if (aiocb->aio_lio_opcode == LIO_READ)
-		return aio_read(aiocb);
+	if (io_u->ddir == DDIR_READ)
+		ret = aio_read(aiocb);
 	else
-		return aio_write(aiocb);
+		ret = aio_write(aiocb);
+
+	if (ret)
+		return errno;
+
+	return 0;
+}
+
+static void fio_posixaio_cleanup(struct thread_data *td)
+{
+	struct posixaio_data *pd = td->aio_data;
+
+	if (pd) {
+		free(pd->aio_events);
+		free(pd);
+		td->aio_data = NULL;
+	}
 }
 
 int fio_posixaio_init(struct thread_data *td)
@@ -220,31 +239,135 @@ int fio_posixaio_init(struct thread_data *td)
 	td->io_getevents = fio_posixaio_getevents;
 	td->io_event = fio_posixaio_event;
 	td->io_cancel = fio_posixaio_cancel;
+	td->io_cleanup = fio_posixaio_cleanup;
+	td->io_sync = fio_io_sync;
 
 	td->aio_data = pd;
 	return 0;
 }
 
-void fio_posixaio_cleanup(struct thread_data *td)
+#else /* FIO_HAVE_POSIXAIO */
+
+int fio_posixaio_init(struct thread_data *td)
 {
-	struct posixaio_data *pd = td->aio_data;
+	return EINVAL;
+}
 
-	if (pd) {
-		free(pd->aio_events);
-		free(pd);
-		td->aio_data = NULL;
+#endif /* FIO_HAVE_POSIXAIO */
+
+struct syncio_data {
+	struct io_u *last_io_u;
+};
+
+static int fio_syncio_getevents(struct thread_data *td, int min, int max,
+				struct timespec *t)
+{
+	if (list_empty(&td->io_u_busylist))
+		return 0;
+
+	return 1;
+}
+
+static struct io_u *fio_syncio_event(struct thread_data *td, int event)
+{
+	struct syncio_data *sd = td->aio_data;
+
+	return sd->last_io_u;
+}
+
+static int fio_syncio_prep(struct thread_data *td, struct io_u *io_u,
+			   int read)
+{
+	if (td->cur_off != io_u->offset) {
+		if (lseek(td->fd, io_u->offset, SEEK_SET) == -1) {
+			td_verror(td, errno);
+			return 1;
+		}
 	}
+
+	return 0;
 }
 
-#else /* FIO_HAVE_POSIXAIO */
+static int fio_syncio_queue(struct thread_data *td, struct io_u *io_u)
+{
+	struct syncio_data *sd = td->aio_data;
+	int ret;
 
-int fio_posixaio_init(struct thread_data *td)
+	if (io_u->ddir == DDIR_READ)
+		ret = read(td->fd, io_u->buf, io_u->buflen);
+	else
+		ret = write(td->fd, io_u->buf, io_u->buflen);
+
+	if (ret < 0)
+		return errno;
+
+	sd->last_io_u = io_u;
+	return 0;
+}
+
+static void fio_syncio_cleanup(struct thread_data *td)
 {
-	return EINVAL;
+	free(td->aio_data);
+	td->aio_data = NULL;
 }
 
-void fio_posixaio_cleanup(struct thread_data *td)
+int fio_syncio_init(struct thread_data *td)
 {
+	struct syncio_data *sd = malloc(sizeof(*sd));
+
+	td->io_prep = fio_syncio_prep;
+	td->io_queue = fio_syncio_queue;
+	td->io_getevents = fio_syncio_getevents;
+	td->io_event = fio_syncio_event;
+	td->io_cancel = NULL;
+	td->io_cleanup = fio_syncio_cleanup;
+	td->io_sync = fio_io_sync;
+
+	sd->last_io_u = NULL;
+	td->aio_data = sd;
+	return 0;
 }
 
-#endif /* FIO_HAVE_POSIXAIO */
+static int fio_mmapio_queue(struct thread_data *td, struct io_u *io_u)
+{
+	unsigned long long real_off = io_u->offset - td->file_offset;
+	struct syncio_data *sd = td->aio_data;
+
+	if (io_u->ddir == DDIR_READ)
+		memcpy(io_u->buf, td->mmap + real_off, io_u->buflen);
+	else
+		memcpy(td->mmap + real_off, io_u->buf, io_u->buflen);
+
+	/*
+	 * not really direct, but should drop the pages from the cache
+	 */
+	if (td->odirect) {
+		msync(td->mmap + real_off, io_u->buflen, MS_SYNC);
+		madvise(td->mmap + real_off, io_u->buflen,  MADV_DONTNEED);
+	}
+
+	sd->last_io_u = io_u;
+	return 0;
+}
+
+static int fio_mmapio_sync(struct thread_data *td)
+{
+	return msync(td->mmap, td->file_size, MS_SYNC);
+}
+
+int fio_mmapio_init(struct thread_data *td)
+{
+	struct syncio_data *sd = malloc(sizeof(*sd));
+
+	td->io_prep = NULL;
+	td->io_queue = fio_mmapio_queue;
+	td->io_getevents = fio_syncio_getevents;
+	td->io_event = fio_syncio_event;
+	td->io_cancel = NULL;
+	td->io_cleanup = fio_syncio_cleanup;
+	td->io_sync = fio_mmapio_sync;
+
+	sd->last_io_u = NULL;
+	td->aio_data = sd;
+	return 0;
+}
diff --git a/fio.c b/fio.c
index afe63ab..3fd2859 100644
--- a/fio.c
+++ b/fio.c
@@ -567,6 +567,19 @@ static struct io_u *__get_io_u(struct thread_data *td)
 	return io_u;
 }
 
+static int td_io_prep(struct thread_data *td, struct io_u *io_u, int read)
+{
+	if (td->io_prep && td->io_prep(td, io_u, read))
+		return 1;
+
+	if (read)
+		io_u->ddir = DDIR_READ;
+	else
+		io_u->ddir = DDIR_WRITE;
+
+	return 0;
+}
+
 static struct io_u *get_io_u(struct thread_data *td)
 {
 	struct io_u *io_u;
@@ -602,8 +615,10 @@ static struct io_u *get_io_u(struct thread_data *td)
 	if (td->verify != VERIFY_NONE)
 		populate_io_u(td, io_u);
 
-	if (td->io_prep)
-		td->io_prep(td, io_u, td_read(td));
+	if (td_io_prep(td, io_u, td_read(td))) {
+		put_io_u(td, io_u);
+		return NULL;
+	}
 
 	gettimeofday(&io_u->start_time, NULL);
 	return io_u;
@@ -680,182 +695,12 @@ static void log_io_piece(struct thread_data *td, struct io_u *io_u)
 	list_add(&ipo->list, entry);
 }
 
-static void do_sync_verify(struct thread_data *td)
-{
-	struct timeval t;
-	struct io_u *io_u = NULL;
-	int ret;
-
-	td_set_runstate(td, TD_VERIFYING);
-
-	io_u = __get_io_u(td);
-
-	if (!td->odirect) {
-		if (!td->use_mmap) {
-			if (fadvise(td->fd, td->file_offset, td->io_size, POSIX_FADV_DONTNEED) < 0) {
-				td_verror(td, errno);
-				goto out;
-			}
-		} else {
-			if (madvise(td->mmap, td->io_size, MADV_DONTNEED)) {
-				td_verror(td, errno);
-				goto out;
-			}
-		}
-	}
-
-	do {
-		if (td->terminate)
-			break;
-
-		gettimeofday(&t, NULL);
-		if (runtime_exceeded(td, &t))
-			break;
-
-		if (get_next_verify(td, &io_u->offset, &io_u->buflen))
-			break;
-
-		if (td->cur_off != io_u->offset) {
-			if (lseek(td->fd, io_u->offset, SEEK_SET) == -1) {
-				td_verror(td, errno);
-				break;
-			}
-		}
-
-		ret = read(td->fd, io_u->buf, io_u->buflen);
-		if (ret < (int) io_u->buflen) {
-			if (ret == -1) {
-				td_verror(td, errno);
-				break;
-			} else if (!ret)
-				break;
-			else
-				io_u->buflen = ret;
-		}
-
-		if (verify_io_u(io_u))
-			break;
-
-		td->cur_off = io_u->offset + io_u->buflen;
-	} while (1);
-
-out:
-	td_set_runstate(td, TD_RUNNING);
-	put_io_u(td, io_u);
-}
-
-static int __do_sync_mmap(struct thread_data *td, struct io_u *io_u)
-{
-	unsigned long long real_off = io_u->offset - td->file_offset;
-
-	if (td_read(td))
-		memcpy(io_u->buf, td->mmap + real_off, io_u->buflen);
-	else
-		memcpy(td->mmap + real_off, io_u->buf, io_u->buflen);
-	
-	/*
-	 * not really direct, but should drop the pages from the cache
-	 */
-	if (td->odirect) {
-		msync(td->mmap + real_off, io_u->buflen, MS_SYNC);
-		madvise(td->mmap + real_off, io_u->buflen,  MADV_DONTNEED);
-	}
-
-	return io_u->buflen;
-}
-
-static int __do_sync_rw(struct thread_data *td, struct io_u *io_u)
-{
-	if (td->cur_off != io_u->offset) {
-		if (lseek(td->fd, io_u->offset, SEEK_SET) == -1) {
-			td_verror(td, errno);
-			return 1;
-		}
-	}
-
-	if (td_read(td))
-		return read(td->fd, io_u->buf, io_u->buflen);
-	else
-		return write(td->fd, io_u->buf, io_u->buflen);
-}
-
-static void sync_td(struct thread_data *td)
-{
-	if (!td->use_mmap)
-		fsync(td->fd);
-	else
-		msync(td->mmap, td->file_size, MS_SYNC);
-}
-
-static void do_sync_io(struct thread_data *td)
+static int sync_td(struct thread_data *td)
 {
-	unsigned long msec, usec;
-	struct io_u *io_u = NULL;
-	struct timeval e;
-
-	while (td->this_io_bytes < td->io_size) {
-		int ret;
-
-		if (td->terminate)
-			break;
-
-		io_u = get_io_u(td);
-		if (!io_u)
-			break;
-
-		if (!td->use_mmap)
-			ret = __do_sync_rw(td, io_u);
-		else
-			ret = __do_sync_mmap(td, io_u);
+	if (td->io_sync)
+		return td->io_sync(td);
 
-		if (ret < (int) io_u->buflen) {
-			if (ret == -1)
-				td_verror(td, errno);
-			break;
-		}
-
-		if (td_write(td))
-			log_io_piece(td, io_u);
-
-		td->io_blocks++;
-		td->io_bytes += io_u->buflen;
-		td->this_io_bytes += io_u->buflen;
-		td->cur_off = io_u->offset + io_u->buflen;
-
-		gettimeofday(&e, NULL);
-
-		usec = utime_since(&io_u->start_time, &e);
-
-		rate_throttle(td, usec, io_u->buflen);
-
-		if (check_min_rate(td, &e)) {
-			td_verror(td, ENOMEM);
-			break;
-		}
-
-		msec = usec / 1000;
-		add_clat_sample(td, msec);
-		add_bw_sample(td);
-
-		if (runtime_exceeded(td, &e))
-			break;
-
-		put_io_u(td, io_u);
-		io_u = NULL;
-
-		if (td->thinktime)
-			usec_sleep(td, td->thinktime);
-
-		if (should_fsync(td) && td->fsync_blocks &&
-		    (td->io_blocks % td->fsync_blocks) == 0)
-			sync_td(td);
-	}
-
-	if (io_u)
-		put_io_u(td, io_u);
-
-	if (should_fsync(td))
-		sync_td(td);
+	return 0;
 }
 
 static int io_u_getevents(struct thread_data *td, int min, int max,
@@ -919,12 +764,14 @@ static void cleanup_pending_aio(struct thread_data *td)
 	/*
 	 * now cancel remaining active events
 	 */
-	list_for_each_safe(entry, n, &td->io_u_busylist) {
-		io_u = list_entry(entry, struct io_u, list);
+	if (td->io_cancel) {
+		list_for_each_safe(entry, n, &td->io_u_busylist) {
+			io_u = list_entry(entry, struct io_u, list);
 
-		r = td->io_cancel(td, io_u);
-		if (!r)
-			put_io_u(td, io_u);
+			r = td->io_cancel(td, io_u);
+			if (!r)
+				put_io_u(td, io_u);
+		}
 	}
 
 	if (td->cur_depth) {
@@ -934,7 +781,7 @@ static void cleanup_pending_aio(struct thread_data *td)
 	}
 }
 
-static int async_do_verify(struct thread_data *td, struct io_u **io_u)
+static int do_io_u_verify(struct thread_data *td, struct io_u **io_u)
 {
 	struct io_u *v_io_u = *io_u;
 	int ret = 0;
@@ -948,7 +795,7 @@ static int async_do_verify(struct thread_data *td, struct io_u **io_u)
 	return ret;
 }
 
-static void do_async_verify(struct thread_data *td)
+static void do_verify(struct thread_data *td)
 {
 	struct timeval t;
 	struct io_u *io_u, *v_io_u = NULL;
@@ -973,7 +820,10 @@ static void do_async_verify(struct thread_data *td)
 			break;
 		}
 
-		td->io_prep(td, io_u, 1);
+		if (td_io_prep(td, io_u, 1)) {
+			put_io_u(td, io_u);
+			break;
+		}
 
 		ret = io_u_queue(td, io_u);
 		if (ret) {
@@ -986,7 +836,7 @@ static void do_async_verify(struct thread_data *td)
 		 * we have one pending to verify, do that while the next
 		 * we are doing io on the next one
 		 */
-		if (async_do_verify(td, &v_io_u))
+		if (do_io_u_verify(td, &v_io_u))
 			break;
 
 		ret = io_u_getevents(td, 1, 1, NULL);
@@ -1003,12 +853,12 @@ static void do_async_verify(struct thread_data *td)
 		/*
 		 * if we can't submit more io, we need to verify now
 		 */
-		if (queue_full(td) && async_do_verify(td, &v_io_u))
+		if (queue_full(td) && do_io_u_verify(td, &v_io_u))
 			break;
 
 	} while (1);
 
-	async_do_verify(td, &v_io_u);
+	do_io_u_verify(td, &v_io_u);
 
 	if (td->cur_depth)
 		cleanup_pending_aio(td);
@@ -1016,7 +866,7 @@ static void do_async_verify(struct thread_data *td)
 	td_set_runstate(td, TD_RUNNING);
 }
 
-static void do_async_io(struct thread_data *td)
+static void do_io(struct thread_data *td)
 {
 	struct timeval s, e;
 	unsigned long usec;
@@ -1086,27 +936,29 @@ static void do_async_io(struct thread_data *td)
 
 		if (should_fsync(td) && td->fsync_blocks &&
 		    (td->io_blocks % td->fsync_blocks) == 0)
-			fsync(td->fd);
+			sync_td(td);
 	}
 
 	if (td->cur_depth)
 		cleanup_pending_aio(td);
 
 	if (should_fsync(td))
-		fsync(td->fd);
+		sync_td(td);
 }
 
-static void cleanup_aio(struct thread_data *td)
+static void cleanup_io(struct thread_data *td)
 {
-	if (td->io_engine == FIO_LIBAIO)
-		fio_libaio_cleanup(td);
-	else if (td->io_engine == FIO_POSIXAIO)
-		fio_posixaio_cleanup(td);
+	if (td->io_cleanup)
+		td->io_cleanup(td);
 }
 
-static int init_aio(struct thread_data *td)
+static int init_io(struct thread_data *td)
 {
-	if (td->io_engine == FIO_LIBAIO)
+	if (td->io_engine == FIO_SYNCIO)
+		return fio_syncio_init(td);
+	else if (td->io_engine == FIO_MMAPIO)
+		return fio_mmapio_init(td);
+	else if (td->io_engine == FIO_LIBAIO)
 		return fio_libaio_init(td);
 	else if (td->io_engine == FIO_POSIXAIO)
 		return fio_posixaio_init(td);
@@ -1149,7 +1001,7 @@ static int init_io_u(struct thread_data *td)
 	int i, max_units;
 	char *p;
 
-	if (td->io_engine == FIO_SYNCIO)
+	if (td->io_engine & FIO_SYNCIO)
 		max_units = 1;
 	else
 		max_units = td->aio_depth;
@@ -1695,7 +1547,7 @@ static void disk_util_timer_arm(void)
 
 static void clear_io_state(struct thread_data *td)
 {
-	if (td->io_engine == FIO_SYNCIO)
+	if (td->io_engine & FIO_SYNCIO)
 		lseek(td->fd, SEEK_SET, 0);
 
 	td->cur_off = 0;
@@ -1738,7 +1590,7 @@ static void *thread_main(void *data)
 		goto err;
 	}
 
-	if ((td->io_engine != FIO_SYNCIO) && init_aio(td))
+	if (init_io(td))
 		goto err;
 
 	if (td->ioprio) {
@@ -1770,10 +1622,7 @@ static void *thread_main(void *data)
 		clear_io_state(td);
 		prune_io_piece_log(td);
 
-		if (td->io_engine == FIO_SYNCIO)
-			do_sync_io(td);
-		else
-			do_async_io(td);
+		do_io(td);
 
 		if (td->error || td->terminate)
 			break;
@@ -1786,10 +1635,7 @@ static void *thread_main(void *data)
 
 		clear_io_state(td);
 
-		if (td->io_engine == FIO_SYNCIO)
-			do_sync_verify(td);
-		else
-			do_async_verify(td);
+		do_verify(td);
 
 		if (td->error || td->terminate)
 			break;
@@ -1812,8 +1658,7 @@ err:
 	}
 	if (td->mmap)
 		munmap(td->mmap, td->file_size);
-	if (td->io_engine != FIO_SYNCIO)
-		cleanup_aio(td);
+	cleanup_io(td);
 	cleanup_io_u(td);
 	if (ret) {
 		sem_post(&startup_sem);
@@ -2050,7 +1895,7 @@ static void run_threads(void)
 
 			td_set_runstate(td, TD_CREATED);
 			check_str_update(td, nr_running, t_rate, m_rate);
-			sem_init(&startup_sem, 1, 1);
+			sem_init(&startup_sem, 0, 1);
 			todo--;
 			nr_started++;
 
diff --git a/fio.h b/fio.h
index 7ff0ea0..3f2fc84 100644
--- a/fio.h
+++ b/fio.h
@@ -59,6 +59,7 @@ struct io_u {
 	unsigned long long offset;
 
 	unsigned char seen;
+	unsigned char ddir;
 
 	struct list_head list;
 };
@@ -143,11 +144,13 @@ struct thread_data {
 	unsigned long long cur_off;
 
 	void *aio_data;
-	void (*io_prep)(struct thread_data *, struct io_u *, int);
+	int (*io_prep)(struct thread_data *, struct io_u *, int);
 	int (*io_queue)(struct thread_data *, struct io_u *);
 	int (*io_getevents)(struct thread_data *, int, int, struct timespec *);
 	struct io_u *(*io_event)(struct thread_data *, int);
 	int (*io_cancel)(struct thread_data *, struct io_u *);
+	void (*io_cleanup)(struct thread_data *);
+	int (*io_sync)(struct thread_data *);
 	unsigned int aio_depth;
 
 	unsigned int cur_depth;
@@ -249,9 +252,10 @@ enum {
 };
 
 enum {
-	FIO_SYNCIO = 0,
-	FIO_LIBAIO,
-	FIO_POSIXAIO,
+	FIO_SYNCIO = 1 << 0,
+	FIO_MMAPIO = 1 << 1 | FIO_SYNCIO,
+	FIO_LIBAIO = 1 << 2,
+	FIO_POSIXAIO = 1 << 3,
 };
 
 #define td_read(td)		((td)->ddir == DDIR_READ)
diff --git a/os.h b/os.h
index adae4fe..1ba0b24 100644
--- a/os.h
+++ b/os.h
@@ -36,8 +36,8 @@
 
 struct thread_data;
 extern int fio_libaio_init(struct thread_data *);
-extern void fio_libaio_cleanup(struct thread_data *);
 extern int fio_posixaio_init(struct thread_data *);
-extern void fio_posixaio_cleanup(struct thread_data *);
+extern int fio_syncio_init(struct thread_data *);
+extern int fio_mmapio_init(struct thread_data *);
 
 #endif
-- 
2.25.1