[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

#ifndef IOCB_FLAG_HIPRI
#define IOCB_FLAG_HIPRI	(1 << 2)
#endif

/*
 * 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_SUBMIT
#define IORING_FLAG_SUBMIT	(1 << 0)
#endif
#ifndef IORING_FLAG_GETEVENTS
#define IORING_FLAG_GETEVENTS	(1 << 1)
#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 aioring_options *o = td->eo;
	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;
		if (o->hipri)
			iocb->u.c.flags |= IOCB_FLAG_HIPRI;
		else
			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 = 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, IORING_FLAG_SUBMIT);
		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_SUBMIT |
						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);

			if (o->hipri)
				iocb->u.c.flags |= IOCB_FLAG_HIPRI;
		}
	}

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