[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"

#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;

struct aio_sq_ring {
	union {
		struct {
			u32 head;
			u32 tail;
			u32 nr_events;
			u16 sq_thread_cpu;
			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;

	struct aio_cq_ring *cq_ring;
	struct io_event *events;

	int queued;
	int cq_ring_off;
};

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) {
		if (o->fixedbufs) {
			iocb->aio_fildes = f->fd;
			iocb->aio_lio_opcode = IO_CMD_PREAD;
			iocb->u.c.offset = io_u->offset;
		} else {
			io_prep_pread(iocb, f->fd, io_u->xfer_buf, io_u->xfer_buflen, io_u->offset);
			if (o->hipri)
				iocb->u.c.flags |= IOCB_FLAG_HIPRI;
		}
	} else if (io_u->ddir == DDIR_WRITE) {
		if (o->fixedbufs) {
			iocb->aio_fildes = f->fd;
			iocb->aio_lio_opcode = IO_CMD_PWRITE;
			iocb->u.c.offset = io_u->offset;
		} else {
			io_prep_pwrite(iocb, f->fd, io_u->xfer_buf, io_u->xfer_buflen, io_u->offset);
			if (o->hipri)
				iocb->u.c.flags |= IOCB_FLAG_HIPRI;
		}
	} 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;
	int index;

	index = event + ld->cq_ring_off;
	if (index >= ld->cq_ring->nr_events)
		index -= ld->cq_ring->nr_events;

	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;

	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++;
		if (head == ring->nr_events)
			head = 0;
	} 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;
	int r, events = 0;

	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 == td->o.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;
	if (next_tail == ring->nr_events)
		next_tail = 0;
	read_barrier();
	if (next_tail == ring->head)
		return FIO_Q_BUSY;

	ring->array[tail] = 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];
		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++;
		if (start == ld->sq_ring->nr_events)
			start = 0;
	}
}

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) {
		ld->queued = 0;
		return 0;
	}

	do {
		int 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 void fio_aioring_cleanup(struct thread_data *td)
{
	struct aioring_data *ld = td->io_ops_data;

	if (ld) {
		/* Bump depth to match init depth */
		td->o.iodepth++;

		/*
		 * 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;
	} else 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, -err, "io_queue_init");
		return 1;
	}

	/* Adjust depth back again */
	td->o.iodepth--;
	return 0;
}

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

	if (o->sqthread_set && o->sqwq) {
		log_err("fio: aioring sqthread and sqwq are mutually exclusive\n");
		return 1;
	}

	/* ring needs an extra entry, add one to achieve QD set */
	td->o.iodepth++;

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

	/* 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->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;

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