[fio.git] / engines / aioring.c

/*
 * aioring engine
 *
 * IO engine using the new native Linux libaio ring interface. See:
 *
 * http://git.kernel.dk/cgit/linux-block/log/?h=aio-poll
 *
 */
#include <stdlib.h>
#include <unistd.h>
#include <errno.h>
#include <libaio.h>
#include <sys/time.h>
#include <sys/resource.h>

#include "../fio.h"
#include "../lib/pow2.h"
#include "../optgroup.h"
#include "../lib/memalign.h"
#include "../lib/fls.h"

#ifdef ARCH_HAVE_AIORING

/*
 * io_setup2(2) flags
 */
#ifndef IOCTX_FLAG_IOPOLL
#define IOCTX_FLAG_IOPOLL	(1 << 0)
#endif
#ifndef IOCTX_FLAG_SCQRING
#define IOCTX_FLAG_SCQRING	(1 << 1)
#endif
#ifndef IOCTX_FLAG_FIXEDBUFS
#define IOCTX_FLAG_FIXEDBUFS	(1 << 2)
#endif
#ifndef IOCTX_FLAG_SQTHREAD
#define IOCTX_FLAG_SQTHREAD	(1 << 3)
#endif
#ifndef IOCTX_FLAG_SQWQ
#define IOCTX_FLAG_SQWQ		(1 << 4)
#endif
#ifndef IOCTX_FLAG_SQPOLL
#define IOCTX_FLAG_SQPOLL	(1 << 5)
#endif


/*
 * io_ring_enter(2) flags
 */
#ifndef IORING_FLAG_GETEVENTS
#define IORING_FLAG_GETEVENTS	(1 << 0)
#endif

typedef uint64_t u64;
typedef uint32_t u32;
typedef uint16_t u16;

#define IORING_SQ_NEED_WAKEUP	(1 << 0)

#define IOEV_RES2_CACHEHIT	(1 << 0)

struct aio_sq_ring {
	union {
		struct {
			u32 head;
			u32 tail;
			u32 nr_events;
			u16 sq_thread_cpu;
			u16 kflags;
			u64 iocbs;
		};
		u32 pad[16];
	};
	u32 array[0];
};

struct aio_cq_ring {
	union {
		struct {
			u32 head;
			u32 tail;
			u32 nr_events;
		};
		struct io_event pad;
	};
	struct io_event events[0];
};

struct aioring_data {
	io_context_t aio_ctx;
	struct io_u **io_us;
	struct io_u **io_u_index;

	struct aio_sq_ring *sq_ring;
	struct iocb *iocbs;
	unsigned sq_ring_mask;

	struct aio_cq_ring *cq_ring;
	struct io_event *events;
	unsigned cq_ring_mask;

	int queued;
	int cq_ring_off;
	unsigned iodepth;

	uint64_t cachehit;
	uint64_t cachemiss;
};

struct aioring_options {
	void *pad;
	unsigned int hipri;
	unsigned int fixedbufs;
	unsigned int sqthread;
	unsigned int sqthread_set;
	unsigned int sqthread_poll;
	unsigned int sqwq;
};

static int fio_aioring_sqthread_cb(void *data,
				   unsigned long long *val)
{
	struct aioring_options *o = data;

	o->sqthread = *val;
	o->sqthread_set = 1;
	return 0;
}

static struct fio_option options[] = {
	{
		.name	= "hipri",
		.lname	= "High Priority",
		.type	= FIO_OPT_STR_SET,
		.off1	= offsetof(struct aioring_options, hipri),
		.help	= "Use polled IO completions",
		.category = FIO_OPT_C_ENGINE,
		.group	= FIO_OPT_G_LIBAIO,
	},
	{
		.name	= "fixedbufs",
		.lname	= "Fixed (pre-mapped) IO buffers",
		.type	= FIO_OPT_STR_SET,
		.off1	= offsetof(struct aioring_options, fixedbufs),
		.help	= "Pre map IO buffers",
		.category = FIO_OPT_C_ENGINE,
		.group	= FIO_OPT_G_LIBAIO,
	},
	{
		.name	= "sqthread",
		.lname	= "Use kernel SQ thread on this CPU",
		.type	= FIO_OPT_INT,
		.cb	= fio_aioring_sqthread_cb,
		.help	= "Offload submission to kernel thread",
		.category = FIO_OPT_C_ENGINE,
		.group	= FIO_OPT_G_LIBAIO,
	},
	{
		.name	= "sqthread_poll",
		.lname	= "Kernel SQ thread should poll",
		.type	= FIO_OPT_STR_SET,
		.off1	= offsetof(struct aioring_options, sqthread_poll),
		.help	= "Used with sqthread, enables kernel side polling",
		.category = FIO_OPT_C_ENGINE,
		.group	= FIO_OPT_G_LIBAIO,
	},
	{
		.name	= "sqwq",
		.lname	= "Offload submission to kernel workqueue",
		.type	= FIO_OPT_STR_SET,
		.off1	= offsetof(struct aioring_options, sqwq),
		.help	= "Offload submission to kernel workqueue",
		.category = FIO_OPT_C_ENGINE,
		.group	= FIO_OPT_G_LIBAIO,
	},
	{
		.name	= NULL,
	},
};

static int io_ring_enter(io_context_t ctx, unsigned int to_submit,
			 unsigned int min_complete, unsigned int flags)
{
	return syscall(__NR_sys_io_ring_enter, ctx, to_submit, min_complete,
			flags);
}

static int fio_aioring_prep(struct thread_data *td, struct io_u *io_u)
{
	struct aioring_data *ld = td->io_ops_data;
	struct fio_file *f = io_u->file;
	struct iocb *iocb;

	iocb = &ld->iocbs[io_u->index];

	if (io_u->ddir == DDIR_READ || io_u->ddir == DDIR_WRITE) {
		if (io_u->ddir == DDIR_READ)
			iocb->aio_lio_opcode = IO_CMD_PREAD;
		else
			iocb->aio_lio_opcode = IO_CMD_PWRITE;
		iocb->aio_reqprio = 0;
		iocb->aio_fildes = f->fd;
		iocb->u.c.buf = io_u->xfer_buf;
		iocb->u.c.nbytes = io_u->xfer_buflen;
		iocb->u.c.offset = io_u->offset;
		iocb->u.c.flags = 0;
	} else if (ddir_sync(io_u->ddir))
		io_prep_fsync(iocb, f->fd);

	iocb->data = io_u;
	return 0;
}

static struct io_u *fio_aioring_event(struct thread_data *td, int event)
{
	struct aioring_data *ld = td->io_ops_data;
	struct io_event *ev;
	struct io_u *io_u;
	unsigned index;

	index = (event + ld->cq_ring_off) & ld->cq_ring_mask;

	ev = &ld->cq_ring->events[index];
	io_u = ev->data;

	if (ev->res != io_u->xfer_buflen) {
		if (ev->res > io_u->xfer_buflen)
			io_u->error = -ev->res;
		else
			io_u->resid = io_u->xfer_buflen - ev->res;
	} else
		io_u->error = 0;

	if (io_u->ddir == DDIR_READ) {
		if (ev->res2 & IOEV_RES2_CACHEHIT)
			ld->cachehit++;
		else
			ld->cachemiss++;
	}

	return io_u;
}

static int fio_aioring_cqring_reap(struct thread_data *td, unsigned int events,
				   unsigned int max)
{
	struct aioring_data *ld = td->io_ops_data;
	struct aio_cq_ring *ring = ld->cq_ring;
	u32 head, reaped = 0;

	head = ring->head;
	do {
		read_barrier();
		if (head == ring->tail)
			break;
		reaped++;
		head++;
	} while (reaped + events < max);

	ring->head = head;
	write_barrier();
	return reaped;
}

static int fio_aioring_getevents(struct thread_data *td, unsigned int min,
				 unsigned int max, const struct timespec *t)
{
	struct aioring_data *ld = td->io_ops_data;
	unsigned actual_min = td->o.iodepth_batch_complete_min == 0 ? 0 : min;
	struct aioring_options *o = td->eo;
	struct aio_cq_ring *ring = ld->cq_ring;
	unsigned events = 0;
	int r;

	ld->cq_ring_off = ring->head;
	do {
		r = fio_aioring_cqring_reap(td, events, max);
		if (r) {
			events += r;
			continue;
		}

		if (!o->sqthread_poll) {
			r = io_ring_enter(ld->aio_ctx, 0, actual_min,
						IORING_FLAG_GETEVENTS);
			if (r < 0) {
				if (errno == EAGAIN)
					continue;
				td_verror(td, errno, "io_ring_enter get");
				break;
			}
		}
	} while (events < min);

	return r < 0 ? r : events;
}

static enum fio_q_status fio_aioring_queue(struct thread_data *td,
					   struct io_u *io_u)
{
	struct aioring_data *ld = td->io_ops_data;
	struct aio_sq_ring *ring = ld->sq_ring;
	unsigned tail, next_tail;

	fio_ro_check(td, io_u);

	if (ld->queued == ld->iodepth)
		return FIO_Q_BUSY;

	if (io_u->ddir == DDIR_TRIM) {
		if (ld->queued)
			return FIO_Q_BUSY;

		do_io_u_trim(td, io_u);
		io_u_mark_submit(td, 1);
		io_u_mark_complete(td, 1);
		return FIO_Q_COMPLETED;
	}

	tail = ring->tail;
	next_tail = tail + 1;
	read_barrier();
	if (next_tail == ring->head)
		return FIO_Q_BUSY;

	ring->array[tail & ld->sq_ring_mask] = io_u->index;
	ring->tail = next_tail;
	write_barrier();

	ld->queued++;
	return FIO_Q_QUEUED;
}

static void fio_aioring_queued(struct thread_data *td, int start, int nr)
{
	struct aioring_data *ld = td->io_ops_data;
	struct timespec now;

	if (!fio_fill_issue_time(td))
		return;

	fio_gettime(&now, NULL);

	while (nr--) {
		int index = ld->sq_ring->array[start & ld->sq_ring_mask];
		struct io_u *io_u = ld->io_u_index[index];

		memcpy(&io_u->issue_time, &now, sizeof(now));
		io_u_queued(td, io_u);

		start++;
	}
}

static int fio_aioring_commit(struct thread_data *td)
{
	struct aioring_data *ld = td->io_ops_data;
	struct aioring_options *o = td->eo;
	int ret;

	if (!ld->queued)
		return 0;

	/* Nothing to do */
	if (o->sqthread_poll) {
		struct aio_sq_ring *ring = ld->sq_ring;

		if (ring->kflags & IORING_SQ_NEED_WAKEUP)
			io_ring_enter(ld->aio_ctx, ld->queued, 0, 0);
		ld->queued = 0;
		return 0;
	}

	do {
		unsigned start = ld->sq_ring->head;
		long nr = ld->queued;

		ret = io_ring_enter(ld->aio_ctx, nr, 0, IORING_FLAG_GETEVENTS);
		if (ret > 0) {
			fio_aioring_queued(td, start, ret);
			io_u_mark_submit(td, ret);

			ld->queued -= ret;
			ret = 0;
		} else if (!ret) {
			io_u_mark_submit(td, ret);
			continue;
		} else {
			if (errno == EAGAIN) {
				ret = fio_aioring_cqring_reap(td, 0, ld->queued);
				if (ret)
					continue;
				/* Shouldn't happen */
				usleep(1);
				continue;
			}
			td_verror(td, errno, "io_ring_enter sumit");
			break;
		}
	} while (ld->queued);

	return ret;
}

static size_t aioring_cq_size(struct thread_data *td)
{
	return sizeof(struct aio_cq_ring) + 2 * td->o.iodepth * sizeof(struct io_event);
}

static size_t aioring_sq_iocb(struct thread_data *td)
{
	return sizeof(struct iocb) * td->o.iodepth;
}

static size_t aioring_sq_size(struct thread_data *td)
{
	return sizeof(struct aio_sq_ring) + td->o.iodepth * sizeof(u32);
}

static unsigned roundup_pow2(unsigned depth)
{
	return 1UL << __fls(depth - 1);
}

static void fio_aioring_cleanup(struct thread_data *td)
{
	struct aioring_data *ld = td->io_ops_data;

	if (ld) {
		td->ts.cachehit += ld->cachehit;
		td->ts.cachemiss += ld->cachemiss;

		/*
		 * Work-around to avoid huge RCU stalls at exit time. If we
		 * don't do this here, then it'll be torn down by exit_aio().
		 * But for that case we can parallellize the freeing, thus
		 * speeding it up a lot.
		 */
		if (!(td->flags & TD_F_CHILD))
			io_destroy(ld->aio_ctx);
		free(ld->io_u_index);
		free(ld->io_us);
		fio_memfree(ld->sq_ring, aioring_sq_size(td), false);
		fio_memfree(ld->iocbs, aioring_sq_iocb(td), false);
		fio_memfree(ld->cq_ring, aioring_cq_size(td), false);
		free(ld);
	}
}

static int fio_aioring_queue_init(struct thread_data *td)
{
	struct aioring_data *ld = td->io_ops_data;
	struct aioring_options *o = td->eo;
	int flags = IOCTX_FLAG_SCQRING;
	int depth = td->o.iodepth;

	if (o->hipri)
		flags |= IOCTX_FLAG_IOPOLL;
	if (o->sqthread_set) {
		ld->sq_ring->sq_thread_cpu = o->sqthread;
		flags |= IOCTX_FLAG_SQTHREAD;
		if (o->sqthread_poll)
			flags |= IOCTX_FLAG_SQPOLL;
	}
	if (o->sqwq)
		flags |= IOCTX_FLAG_SQWQ;

	if (o->fixedbufs) {
		struct rlimit rlim = {
			.rlim_cur = RLIM_INFINITY,
			.rlim_max = RLIM_INFINITY,
		};

		setrlimit(RLIMIT_MEMLOCK, &rlim);
		flags |= IOCTX_FLAG_FIXEDBUFS;
	}

	return syscall(__NR_sys_io_setup2, depth, flags,
			ld->sq_ring, ld->cq_ring, &ld->aio_ctx);
}

static int fio_aioring_post_init(struct thread_data *td)
{
	struct aioring_data *ld = td->io_ops_data;
	struct aioring_options *o = td->eo;
	struct io_u *io_u;
	struct iocb *iocb;
	int err = 0;

	if (o->fixedbufs) {
		int i;

		for (i = 0; i < td->o.iodepth; i++) {
			io_u = ld->io_u_index[i];
			iocb = &ld->iocbs[i];
			iocb->u.c.buf = io_u->buf;
			iocb->u.c.nbytes = td_max_bs(td);
		}
	}

	err = fio_aioring_queue_init(td);

	if (err) {
		td_verror(td, errno, "io_queue_init");
		return 1;
	}

	return 0;
}

static int fio_aioring_init(struct thread_data *td)
{
	struct aioring_data *ld;

	ld = calloc(1, sizeof(*ld));

	/* ring depth must be a power-of-2 */
	ld->iodepth = td->o.iodepth;
	td->o.iodepth = roundup_pow2(td->o.iodepth);

	/* io_u index */
	ld->io_u_index = calloc(td->o.iodepth, sizeof(struct io_u *));
	ld->io_us = calloc(td->o.iodepth, sizeof(struct io_u *));

	ld->iocbs = fio_memalign(page_size, aioring_sq_iocb(td), false);
	memset(ld->iocbs, 0, aioring_sq_iocb(td));

	ld->sq_ring = fio_memalign(page_size, aioring_sq_size(td), false);
	memset(ld->sq_ring, 0, aioring_sq_size(td));
	ld->sq_ring->nr_events = td->o.iodepth;
	ld->sq_ring->iocbs = (u64) (uintptr_t) ld->iocbs;
	ld->sq_ring_mask = td->o.iodepth - 1;

	ld->cq_ring = fio_memalign(page_size, aioring_cq_size(td), false);
	memset(ld->cq_ring, 0, aioring_cq_size(td));
	ld->cq_ring->nr_events = td->o.iodepth * 2;
	ld->cq_ring_mask = (2 * td->o.iodepth) - 1;

	td->io_ops_data = ld;
	return 0;
}

static int fio_aioring_io_u_init(struct thread_data *td, struct io_u *io_u)
{
	struct aioring_data *ld = td->io_ops_data;

	ld->io_u_index[io_u->index] = io_u;
	return 0;
}

static struct ioengine_ops ioengine = {
	.name			= "aio-ring",
	.version		= FIO_IOOPS_VERSION,
	.init			= fio_aioring_init,
	.post_init		= fio_aioring_post_init,
	.io_u_init		= fio_aioring_io_u_init,
	.prep			= fio_aioring_prep,
	.queue			= fio_aioring_queue,
	.commit			= fio_aioring_commit,
	.getevents		= fio_aioring_getevents,
	.event			= fio_aioring_event,
	.cleanup		= fio_aioring_cleanup,
	.open_file		= generic_open_file,
	.close_file		= generic_close_file,
	.get_file_size		= generic_get_file_size,
	.options		= options,
	.option_struct_size	= sizeof(struct aioring_options),
};

static void fio_init fio_aioring_register(void)
{
	register_ioengine(&ioengine);
}

static void fio_exit fio_aioring_unregister(void)
{
	unregister_ioengine(&ioengine);
}
#endif
Commit	Line	Data
52885fa2 JA	1	/*
	2	* aioring engine
	3	*
a90cd050 JA	4	* IO engine using the new native Linux libaio ring interface. See:
	5	*
	6	* http://git.kernel.dk/cgit/linux-block/log/?h=aio-poll
52885fa2 JA	7	*
	8	*/
	9	#include <stdlib.h>
	10	#include <unistd.h>
	11	#include <errno.h>
	12	#include <libaio.h>
	13	#include <sys/time.h>
	14	#include <sys/resource.h>
	15
	16	#include "../fio.h"
	17	#include "../lib/pow2.h"
	18	#include "../optgroup.h"
	19	#include "../lib/memalign.h"
b87aa01a	20	#include "../lib/fls.h"
52885fa2	21
a30e63cf JA	22	#ifdef ARCH_HAVE_AIORING
a30e63cf JA	23
52885fa2 JA	24	/*
	25	* io_setup2(2) flags
	26	*/
	27	#ifndef IOCTX_FLAG_IOPOLL
	28	#define IOCTX_FLAG_IOPOLL (1 << 0)
	29	#endif
	30	#ifndef IOCTX_FLAG_SCQRING
	31	#define IOCTX_FLAG_SCQRING (1 << 1)
	32	#endif
	33	#ifndef IOCTX_FLAG_FIXEDBUFS
	34	#define IOCTX_FLAG_FIXEDBUFS (1 << 2)
	35	#endif
	36	#ifndef IOCTX_FLAG_SQTHREAD
	37	#define IOCTX_FLAG_SQTHREAD (1 << 3)
	38	#endif
	39	#ifndef IOCTX_FLAG_SQWQ
	40	#define IOCTX_FLAG_SQWQ (1 << 4)
	41	#endif
771c9901 JA	42	#ifndef IOCTX_FLAG_SQPOLL
	43	#define IOCTX_FLAG_SQPOLL (1 << 5)
	44	#endif
	45
52885fa2 JA	46
	47	/*
	48	* io_ring_enter(2) flags
	49	*/
52885fa2	50	#ifndef IORING_FLAG_GETEVENTS
605d3b6a	51	#define IORING_FLAG_GETEVENTS (1 << 0)
52885fa2 JA	52	#endif
	53
	54	typedef uint64_t u64;
	55	typedef uint32_t u32;
	56	typedef uint16_t u16;
	57
4cdbc048 JA	58	#define IORING_SQ_NEED_WAKEUP (1 << 0)
4cdbc048 JA	59
96563db9 JA	60	#define IOEV_RES2_CACHEHIT (1 << 0)
96563db9 JA	61
52885fa2 JA	62	struct aio_sq_ring {
	63	union {
	64	struct {
	65	u32 head;
	66	u32 tail;
	67	u32 nr_events;
	68	u16 sq_thread_cpu;
4cdbc048	69	u16 kflags;
52885fa2 JA	70	u64 iocbs;
	71	};
	72	u32 pad[16];
	73	};
	74	u32 array[0];
	75	};
	76
	77	struct aio_cq_ring {
	78	union {
	79	struct {
	80	u32 head;
	81	u32 tail;
	82	u32 nr_events;
	83	};
	84	struct io_event pad;
	85	};
	86	struct io_event events[0];
	87	};
	88
	89	struct aioring_data {
	90	io_context_t aio_ctx;
	91	struct io_u **io_us;
	92	struct io_u **io_u_index;
	93
	94	struct aio_sq_ring *sq_ring;
	95	struct iocb *iocbs;
b87aa01a	96	unsigned sq_ring_mask;
52885fa2 JA	97
	98	struct aio_cq_ring *cq_ring;
	99	struct io_event *events;
b87aa01a	100	unsigned cq_ring_mask;
52885fa2 JA	101
	102	int queued;
	103	int cq_ring_off;
b87aa01a	104	unsigned iodepth;
96563db9 JA	105
	106	uint64_t cachehit;
	107	uint64_t cachemiss;
52885fa2 JA	108	};
	109
	110	struct aioring_options {
	111	void *pad;
	112	unsigned int hipri;
	113	unsigned int fixedbufs;
a90cd050 JA	114	unsigned int sqthread;
a90cd050 JA	115	unsigned int sqthread_set;
771c9901	116	unsigned int sqthread_poll;
a90cd050	117	unsigned int sqwq;
52885fa2 JA	118	};
52885fa2 JA	119
a90cd050 JA	120	static int fio_aioring_sqthread_cb(void *data,
	121	unsigned long long *val)
	122	{
	123	struct aioring_options *o = data;
	124
	125	o->sqthread = *val;
	126	o->sqthread_set = 1;
	127	return 0;
	128	}
	129
52885fa2 JA	130	static struct fio_option options[] = {
	131	{
	132	.name = "hipri",
	133	.lname = "High Priority",
	134	.type = FIO_OPT_STR_SET,
	135	.off1 = offsetof(struct aioring_options, hipri),
	136	.help = "Use polled IO completions",
	137	.category = FIO_OPT_C_ENGINE,
	138	.group = FIO_OPT_G_LIBAIO,
	139	},
	140	{
	141	.name = "fixedbufs",
	142	.lname = "Fixed (pre-mapped) IO buffers",
	143	.type = FIO_OPT_STR_SET,
	144	.off1 = offsetof(struct aioring_options, fixedbufs),
	145	.help = "Pre map IO buffers",
	146	.category = FIO_OPT_C_ENGINE,
	147	.group = FIO_OPT_G_LIBAIO,
	148	},
a90cd050 JA	149	{
	150	.name = "sqthread",
	151	.lname = "Use kernel SQ thread on this CPU",
	152	.type = FIO_OPT_INT,
	153	.cb = fio_aioring_sqthread_cb,
	154	.help = "Offload submission to kernel thread",
	155	.category = FIO_OPT_C_ENGINE,
	156	.group = FIO_OPT_G_LIBAIO,
	157	},
771c9901 JA	158	{
	159	.name = "sqthread_poll",
	160	.lname = "Kernel SQ thread should poll",
	161	.type = FIO_OPT_STR_SET,
	162	.off1 = offsetof(struct aioring_options, sqthread_poll),
	163	.help = "Used with sqthread, enables kernel side polling",
	164	.category = FIO_OPT_C_ENGINE,
	165	.group = FIO_OPT_G_LIBAIO,
	166	},
a90cd050 JA	167	{
	168	.name = "sqwq",
	169	.lname = "Offload submission to kernel workqueue",
	170	.type = FIO_OPT_STR_SET,
	171	.off1 = offsetof(struct aioring_options, sqwq),
	172	.help = "Offload submission to kernel workqueue",
	173	.category = FIO_OPT_C_ENGINE,
	174	.group = FIO_OPT_G_LIBAIO,
	175	},
52885fa2 JA	176	{
	177	.name = NULL,
	178	},
	179	};
	180
52885fa2 JA	181	static int io_ring_enter(io_context_t ctx, unsigned int to_submit,
	182	unsigned int min_complete, unsigned int flags)
	183	{
52885fa2 JA	184	return syscall(__NR_sys_io_ring_enter, ctx, to_submit, min_complete,
52885fa2 JA	185	flags);
52885fa2 JA	186	}
	187
	188	static int fio_aioring_prep(struct thread_data td, struct io_u io_u)
	189	{
	190	struct aioring_data *ld = td->io_ops_data;
	191	struct fio_file *f = io_u->file;
52885fa2 JA	192	struct iocb *iocb;
	193
	194	iocb = &ld->iocbs[io_u->index];
	195
e3970057 JA	196	if (io_u->ddir == DDIR_READ \|\| io_u->ddir == DDIR_WRITE) {
e3970057 JA	197	if (io_u->ddir == DDIR_READ)
52885fa2	198	iocb->aio_lio_opcode = IO_CMD_PREAD;
e3970057	199	else
52885fa2	200	iocb->aio_lio_opcode = IO_CMD_PWRITE;
e3970057 JA	201	iocb->aio_reqprio = 0;
	202	iocb->aio_fildes = f->fd;
	203	iocb->u.c.buf = io_u->xfer_buf;
	204	iocb->u.c.nbytes = io_u->xfer_buflen;
	205	iocb->u.c.offset = io_u->offset;
ac4f3d4e	206	iocb->u.c.flags = 0;
52885fa2 JA	207	} else if (ddir_sync(io_u->ddir))
	208	io_prep_fsync(iocb, f->fd);
	209
	210	iocb->data = io_u;
	211	return 0;
	212	}
	213
	214	static struct io_u fio_aioring_event(struct thread_data td, int event)
	215	{
	216	struct aioring_data *ld = td->io_ops_data;
	217	struct io_event *ev;
	218	struct io_u *io_u;
b87aa01a	219	unsigned index;
52885fa2	220
b87aa01a	221	index = (event + ld->cq_ring_off) & ld->cq_ring_mask;
52885fa2 JA	222
	223	ev = &ld->cq_ring->events[index];
	224	io_u = ev->data;
	225
	226	if (ev->res != io_u->xfer_buflen) {
	227	if (ev->res > io_u->xfer_buflen)
	228	io_u->error = -ev->res;
	229	else
	230	io_u->resid = io_u->xfer_buflen - ev->res;
	231	} else
	232	io_u->error = 0;
	233
96563db9 JA	234	if (io_u->ddir == DDIR_READ) {
	235	if (ev->res2 & IOEV_RES2_CACHEHIT)
	236	ld->cachehit++;
	237	else
	238	ld->cachemiss++;
	239	}
	240
52885fa2 JA	241	return io_u;
	242	}
	243
	244	static int fio_aioring_cqring_reap(struct thread_data *td, unsigned int events,
	245	unsigned int max)
	246	{
	247	struct aioring_data *ld = td->io_ops_data;
	248	struct aio_cq_ring *ring = ld->cq_ring;
	249	u32 head, reaped = 0;
	250
	251	head = ring->head;
	252	do {
	253	read_barrier();
	254	if (head == ring->tail)
	255	break;
	256	reaped++;
	257	head++;
52885fa2 JA	258	} while (reaped + events < max);
	259
	260	ring->head = head;
	261	write_barrier();
	262	return reaped;
	263	}
	264
	265	static int fio_aioring_getevents(struct thread_data *td, unsigned int min,
	266	unsigned int max, const struct timespec *t)
	267	{
	268	struct aioring_data *ld = td->io_ops_data;
	269	unsigned actual_min = td->o.iodepth_batch_complete_min == 0 ? 0 : min;
771c9901	270	struct aioring_options *o = td->eo;
52885fa2	271	struct aio_cq_ring *ring = ld->cq_ring;
b87aa01a JA	272	unsigned events = 0;
b87aa01a JA	273	int r;
52885fa2 JA	274
	275	ld->cq_ring_off = ring->head;
	276	do {
	277	r = fio_aioring_cqring_reap(td, events, max);
	278	if (r) {
	279	events += r;
	280	continue;
	281	}
	282
771c9901 JA	283	if (!o->sqthread_poll) {
	284	r = io_ring_enter(ld->aio_ctx, 0, actual_min,
	285	IORING_FLAG_GETEVENTS);
	286	if (r < 0) {
	287	if (errno == EAGAIN)
	288	continue;
	289	td_verror(td, errno, "io_ring_enter get");
	290	break;
	291	}
52885fa2 JA	292	}
	293	} while (events < min);
	294
	295	return r < 0 ? r : events;
	296	}
	297
	298	static enum fio_q_status fio_aioring_queue(struct thread_data *td,
	299	struct io_u *io_u)
	300	{
	301	struct aioring_data *ld = td->io_ops_data;
	302	struct aio_sq_ring *ring = ld->sq_ring;
	303	unsigned tail, next_tail;
	304
	305	fio_ro_check(td, io_u);
	306
b87aa01a	307	if (ld->queued == ld->iodepth)
52885fa2 JA	308	return FIO_Q_BUSY;
52885fa2 JA	309
52885fa2 JA	310	if (io_u->ddir == DDIR_TRIM) {
	311	if (ld->queued)
	312	return FIO_Q_BUSY;
	313
	314	do_io_u_trim(td, io_u);
	315	io_u_mark_submit(td, 1);
	316	io_u_mark_complete(td, 1);
	317	return FIO_Q_COMPLETED;
	318	}
	319
	320	tail = ring->tail;
	321	next_tail = tail + 1;
52885fa2 JA	322	read_barrier();
	323	if (next_tail == ring->head)
	324	return FIO_Q_BUSY;
	325
b87aa01a	326	ring->array[tail & ld->sq_ring_mask] = io_u->index;
52885fa2 JA	327	ring->tail = next_tail;
	328	write_barrier();
	329
	330	ld->queued++;
	331	return FIO_Q_QUEUED;
	332	}
	333
	334	static void fio_aioring_queued(struct thread_data *td, int start, int nr)
	335	{
	336	struct aioring_data *ld = td->io_ops_data;
	337	struct timespec now;
	338
	339	if (!fio_fill_issue_time(td))
	340	return;
	341
	342	fio_gettime(&now, NULL);
	343
	344	while (nr--) {
b87aa01a	345	int index = ld->sq_ring->array[start & ld->sq_ring_mask];
f8289afc	346	struct io_u *io_u = ld->io_u_index[index];
52885fa2 JA	347
	348	memcpy(&io_u->issue_time, &now, sizeof(now));
	349	io_u_queued(td, io_u);
	350
	351	start++;
52885fa2 JA	352	}
	353	}
	354
	355	static int fio_aioring_commit(struct thread_data *td)
	356	{
	357	struct aioring_data *ld = td->io_ops_data;
771c9901	358	struct aioring_options *o = td->eo;
52885fa2 JA	359	int ret;
	360
	361	if (!ld->queued)
	362	return 0;
	363
771c9901 JA	364	/* Nothing to do */
771c9901 JA	365	if (o->sqthread_poll) {
4cdbc048 JA	366	struct aio_sq_ring *ring = ld->sq_ring;
	367
	368	if (ring->kflags & IORING_SQ_NEED_WAKEUP)
605d3b6a	369	io_ring_enter(ld->aio_ctx, ld->queued, 0, 0);
771c9901 JA	370	ld->queued = 0;
	371	return 0;
	372	}
	373
52885fa2	374	do {
b87aa01a	375	unsigned start = ld->sq_ring->head;
52885fa2 JA	376	long nr = ld->queued;
52885fa2 JA	377
605d3b6a	378	ret = io_ring_enter(ld->aio_ctx, nr, 0, IORING_FLAG_GETEVENTS);
52885fa2 JA	379	if (ret > 0) {
	380	fio_aioring_queued(td, start, ret);
	381	io_u_mark_submit(td, ret);
	382
	383	ld->queued -= ret;
	384	ret = 0;
a90cd050 JA	385	} else if (!ret) {
a90cd050 JA	386	io_u_mark_submit(td, ret);
52885fa2	387	continue;
a90cd050 JA	388	} else {
	389	if (errno == EAGAIN) {
	390	ret = fio_aioring_cqring_reap(td, 0, ld->queued);
	391	if (ret)
	392	continue;
	393	/* Shouldn't happen */
	394	usleep(1);
	395	continue;
52885fa2	396	}
a90cd050	397	td_verror(td, errno, "io_ring_enter sumit");
52885fa2	398	break;
a90cd050	399	}
52885fa2 JA	400	} while (ld->queued);
	401
	402	return ret;
	403	}
	404
	405	static size_t aioring_cq_size(struct thread_data *td)
	406	{
	407	return sizeof(struct aio_cq_ring) + 2 * td->o.iodepth * sizeof(struct io_event);
	408	}
	409
	410	static size_t aioring_sq_iocb(struct thread_data *td)
	411	{
	412	return sizeof(struct iocb) * td->o.iodepth;
	413	}
	414
	415	static size_t aioring_sq_size(struct thread_data *td)
	416	{
	417	return sizeof(struct aio_sq_ring) + td->o.iodepth * sizeof(u32);
	418	}
	419
b87aa01a JA	420	static unsigned roundup_pow2(unsigned depth)
	421	{
	422	return 1UL << __fls(depth - 1);
	423	}
	424
52885fa2 JA	425	static void fio_aioring_cleanup(struct thread_data *td)
	426	{
	427	struct aioring_data *ld = td->io_ops_data;
	428
	429	if (ld) {
96563db9 JA	430	td->ts.cachehit += ld->cachehit;
	431	td->ts.cachemiss += ld->cachemiss;
	432
52885fa2 JA	433	/*
	434	* Work-around to avoid huge RCU stalls at exit time. If we
	435	* don't do this here, then it'll be torn down by exit_aio().
	436	* But for that case we can parallellize the freeing, thus
	437	* speeding it up a lot.
	438	*/
	439	if (!(td->flags & TD_F_CHILD))
	440	io_destroy(ld->aio_ctx);
	441	free(ld->io_u_index);
	442	free(ld->io_us);
	443	fio_memfree(ld->sq_ring, aioring_sq_size(td), false);
	444	fio_memfree(ld->iocbs, aioring_sq_iocb(td), false);
	445	fio_memfree(ld->cq_ring, aioring_cq_size(td), false);
	446	free(ld);
	447	}
	448	}
	449
	450	static int fio_aioring_queue_init(struct thread_data *td)
	451	{
52885fa2 JA	452	struct aioring_data *ld = td->io_ops_data;
	453	struct aioring_options *o = td->eo;
	454	int flags = IOCTX_FLAG_SCQRING;
	455	int depth = td->o.iodepth;
	456
	457	if (o->hipri)
	458	flags \|= IOCTX_FLAG_IOPOLL;
a90cd050 JA	459	if (o->sqthread_set) {
	460	ld->sq_ring->sq_thread_cpu = o->sqthread;
	461	flags \|= IOCTX_FLAG_SQTHREAD;
771c9901 JA	462	if (o->sqthread_poll)
771c9901 JA	463	flags \|= IOCTX_FLAG_SQPOLL;
f635f1fb JA	464	}
f635f1fb JA	465	if (o->sqwq)
a90cd050 JA	466	flags \|= IOCTX_FLAG_SQWQ;
a90cd050 JA	467
52885fa2 JA	468	if (o->fixedbufs) {
	469	struct rlimit rlim = {
	470	.rlim_cur = RLIM_INFINITY,
	471	.rlim_max = RLIM_INFINITY,
	472	};
	473
	474	setrlimit(RLIMIT_MEMLOCK, &rlim);
	475	flags \|= IOCTX_FLAG_FIXEDBUFS;
	476	}
	477
	478	return syscall(__NR_sys_io_setup2, depth, flags,
	479	ld->sq_ring, ld->cq_ring, &ld->aio_ctx);
52885fa2 JA	480	}
	481
	482	static int fio_aioring_post_init(struct thread_data *td)
	483	{
	484	struct aioring_data *ld = td->io_ops_data;
	485	struct aioring_options *o = td->eo;
	486	struct io_u *io_u;
	487	struct iocb *iocb;
	488	int err = 0;
	489
	490	if (o->fixedbufs) {
	491	int i;
	492
	493	for (i = 0; i < td->o.iodepth; i++) {
	494	io_u = ld->io_u_index[i];
	495	iocb = &ld->iocbs[i];
	496	iocb->u.c.buf = io_u->buf;
	497	iocb->u.c.nbytes = td_max_bs(td);
52885fa2 JA	498	}
	499	}
	500
	501	err = fio_aioring_queue_init(td);
d63a472d	502
52885fa2	503	if (err) {
d63a472d	504	td_verror(td, errno, "io_queue_init");
52885fa2 JA	505	return 1;
	506	}
	507
	508	return 0;
	509	}
	510
	511	static int fio_aioring_init(struct thread_data *td)
	512	{
	513	struct aioring_data *ld;
	514
52885fa2 JA	515	ld = calloc(1, sizeof(*ld));
52885fa2 JA	516
b87aa01a JA	517	/* ring depth must be a power-of-2 */
	518	ld->iodepth = td->o.iodepth;
	519	td->o.iodepth = roundup_pow2(td->o.iodepth);
	520
52885fa2 JA	521	/* io_u index */
	522	ld->io_u_index = calloc(td->o.iodepth, sizeof(struct io_u *));
	523	ld->io_us = calloc(td->o.iodepth, sizeof(struct io_u *));
	524
	525	ld->iocbs = fio_memalign(page_size, aioring_sq_iocb(td), false);
	526	memset(ld->iocbs, 0, aioring_sq_iocb(td));
	527
	528	ld->sq_ring = fio_memalign(page_size, aioring_sq_size(td), false);
	529	memset(ld->sq_ring, 0, aioring_sq_size(td));
	530	ld->sq_ring->nr_events = td->o.iodepth;
1f90e9bb	531	ld->sq_ring->iocbs = (u64) (uintptr_t) ld->iocbs;
b87aa01a	532	ld->sq_ring_mask = td->o.iodepth - 1;
52885fa2 JA	533
	534	ld->cq_ring = fio_memalign(page_size, aioring_cq_size(td), false);
	535	memset(ld->cq_ring, 0, aioring_cq_size(td));
	536	ld->cq_ring->nr_events = td->o.iodepth * 2;
b87aa01a	537	ld->cq_ring_mask = (2 * td->o.iodepth) - 1;
52885fa2 JA	538
	539	td->io_ops_data = ld;
	540	return 0;
	541	}
	542
	543	static int fio_aioring_io_u_init(struct thread_data td, struct io_u io_u)
	544	{
	545	struct aioring_data *ld = td->io_ops_data;
	546
	547	ld->io_u_index[io_u->index] = io_u;
	548	return 0;
	549	}
	550
	551	static struct ioengine_ops ioengine = {
	552	.name = "aio-ring",
	553	.version = FIO_IOOPS_VERSION,
	554	.init = fio_aioring_init,
	555	.post_init = fio_aioring_post_init,
	556	.io_u_init = fio_aioring_io_u_init,
	557	.prep = fio_aioring_prep,
	558	.queue = fio_aioring_queue,
	559	.commit = fio_aioring_commit,
	560	.getevents = fio_aioring_getevents,
	561	.event = fio_aioring_event,
	562	.cleanup = fio_aioring_cleanup,
	563	.open_file = generic_open_file,
	564	.close_file = generic_close_file,
	565	.get_file_size = generic_get_file_size,
	566	.options = options,
	567	.option_struct_size = sizeof(struct aioring_options),
	568	};
	569
	570	static void fio_init fio_aioring_register(void)
	571	{
52885fa2	572	register_ioengine(&ioengine);
52885fa2 JA	573	}
	574
	575	static void fio_exit fio_aioring_unregister(void)
	576	{
52885fa2	577	unregister_ioengine(&ioengine);
52885fa2	578	}
1f90e9bb	579	#endif