[fio.git] / engines / posixaio.c

/*
 * posixaio engine
 *
 * IO engine that uses the posix defined aio interface.
 *
 */
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <errno.h>
#include <fcntl.h>

#include "../fio.h"

struct posixaio_data {
	struct io_u **aio_events;
	unsigned int queued;
};

static int fill_timespec(struct timespec *ts)
{
#ifdef CONFIG_CLOCK_GETTIME
#ifdef CONFIG_CLOCK_MONOTONIC
	clockid_t clk = CLOCK_MONOTONIC;
#else
	clockid_t clk = CLOCK_REALTIME;
#endif
	if (!clock_gettime(clk, ts))
		return 0;

	perror("clock_gettime");
	return 1;
#else
	struct timeval tv;

	gettimeofday(&tv, NULL);
	ts->tv_sec = tv.tv_sec;
	ts->tv_nsec = tv.tv_usec * 1000;
	return 0;
#endif
}

static unsigned long long ts_utime_since_now(struct timespec *t)
{
	long long sec, nsec;
	struct timespec now;

	if (fill_timespec(&now))
		return 0;
	
	sec = now.tv_sec - t->tv_sec;
	nsec = now.tv_nsec - t->tv_nsec;
	if (sec > 0 && nsec < 0) {
		sec--;
		nsec += 1000000000;
	}

	sec *= 1000000;
	nsec /= 1000;
	return sec + nsec;
}

static int fio_posixaio_cancel(struct thread_data fio_unused *td,
			       struct io_u *io_u)
{
	struct fio_file *f = io_u->file;
	int r = aio_cancel(f->fd, &io_u->aiocb);

	if (r == AIO_ALLDONE || r == AIO_CANCELED)
		return 0;

	return 1;
}

static int fio_posixaio_prep(struct thread_data fio_unused *td,
			     struct io_u *io_u)
{
	os_aiocb_t *aiocb = &io_u->aiocb;
	struct fio_file *f = io_u->file;

	aiocb->aio_fildes = f->fd;
	aiocb->aio_buf = io_u->xfer_buf;
	aiocb->aio_nbytes = io_u->xfer_buflen;
	aiocb->aio_offset = io_u->offset;
	aiocb->aio_sigevent.sigev_notify = SIGEV_NONE;

	io_u->seen = 0;
	return 0;
}

#define SUSPEND_ENTRIES	8

static int fio_posixaio_getevents(struct thread_data *td, unsigned int min,
				  unsigned int max, struct timespec *t)
{
	struct posixaio_data *pd = td->io_ops->data;
	os_aiocb_t *suspend_list[SUSPEND_ENTRIES];
	struct flist_head *entry;
	struct timespec start;
	int have_timeout = 0;
	int suspend_entries = 0;
	unsigned int r;

	if (t && !fill_timespec(&start))
		have_timeout = 1;
	else
		memset(&start, 0, sizeof(start));

	r = 0;
	memset(suspend_list, 0, sizeof(*suspend_list));
restart:
	flist_for_each(entry, &td->io_u_busylist) {
		struct io_u *io_u = flist_entry(entry, struct io_u, list);
		int err;

		if (io_u->seen)
			continue;

		err = aio_error(&io_u->aiocb);
		if (err == EINPROGRESS) {
			if (suspend_entries < SUSPEND_ENTRIES) {
				suspend_list[suspend_entries] = &io_u->aiocb;
				suspend_entries++;
			}
			continue;
		}

		io_u->seen = 1;
		pd->queued--;
		pd->aio_events[r++] = io_u;

		if (err == ECANCELED)
			io_u->resid = io_u->xfer_buflen;
		else if (!err) {
			ssize_t retval = aio_return(&io_u->aiocb);

			io_u->resid = io_u->xfer_buflen - retval;
		} else
			io_u->error = err;
	}

	if (r >= min)
		return r;

	if (have_timeout) {
		unsigned long long usec;

		usec = (t->tv_sec * 1000000) + (t->tv_nsec / 1000);
		if (ts_utime_since_now(&start) > usec)
			return r;
	}

	/*
	 * must have some in-flight, wait for at least one
	 */
	aio_suspend((const os_aiocb_t * const *)suspend_list,
							suspend_entries, t);
	goto restart;
}

static struct io_u *fio_posixaio_event(struct thread_data *td, int event)
{
	struct posixaio_data *pd = td->io_ops->data;

	return pd->aio_events[event];
}

static int fio_posixaio_queue(struct thread_data *td,
			      struct io_u *io_u)
{
	struct posixaio_data *pd = td->io_ops->data;
	os_aiocb_t *aiocb = &io_u->aiocb;
	int ret;

	fio_ro_check(td, io_u);

	if (io_u->ddir == DDIR_READ)
		ret = aio_read(aiocb);
	else if (io_u->ddir == DDIR_WRITE)
		ret = aio_write(aiocb);
	else if (io_u->ddir == DDIR_TRIM) {
		if (pd->queued)
			return FIO_Q_BUSY;

		do_io_u_trim(td, io_u);
		return FIO_Q_COMPLETED;
	} else {
#ifdef CONFIG_POSIXAIO_FSYNC
		ret = aio_fsync(O_SYNC, aiocb);
#else
		if (pd->queued)
			return FIO_Q_BUSY;

		do_io_u_sync(td, io_u);
		return FIO_Q_COMPLETED;
#endif
	}
		
	if (ret) {
		/*
		 * At least OSX has a very low limit on the number of pending
		 * IOs, so if it returns EAGAIN, we are out of resources
		 * to queue more. Just return FIO_Q_BUSY to naturally
		 * drop off at this depth.
		 */
		if (errno == EAGAIN)
			return FIO_Q_BUSY;

		io_u->error = errno;
		td_verror(td, io_u->error, "xfer");
		return FIO_Q_COMPLETED;
	}

	pd->queued++;
	return FIO_Q_QUEUED;
}

static void fio_posixaio_cleanup(struct thread_data *td)
{
	struct posixaio_data *pd = td->io_ops->data;

	if (pd) {
		free(pd->aio_events);
		free(pd);
	}
}

static int fio_posixaio_init(struct thread_data *td)
{
	struct posixaio_data *pd = malloc(sizeof(*pd));

	memset(pd, 0, sizeof(*pd));
	pd->aio_events = malloc(td->o.iodepth * sizeof(struct io_u *));
	memset(pd->aio_events, 0, td->o.iodepth * sizeof(struct io_u *));

	td->io_ops->data = pd;
	return 0;
}

static struct ioengine_ops ioengine = {
	.name		= "posixaio",
	.version	= FIO_IOOPS_VERSION,
	.init		= fio_posixaio_init,
	.prep		= fio_posixaio_prep,
	.queue		= fio_posixaio_queue,
	.cancel		= fio_posixaio_cancel,
	.getevents	= fio_posixaio_getevents,
	.event		= fio_posixaio_event,
	.cleanup	= fio_posixaio_cleanup,
	.open_file	= generic_open_file,
	.close_file	= generic_close_file,
	.get_file_size	= generic_get_file_size,
};

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

static void fio_exit fio_posixaio_unregister(void)
{
	unregister_ioengine(&ioengine);
}
Commit	Line	Data
	1	/*
	2	* posixaio engine
	3	*
	4	* IO engine that uses the posix defined aio interface.
	5	*
	6	*/
	7	#include <stdio.h>
	8	#include <stdlib.h>
	9	#include <unistd.h>
	10	#include <errno.h>
	11	#include <fcntl.h>
	12
	13	#include "../fio.h"
	14
	15	struct posixaio_data {
	16	struct io_u **aio_events;
	17	unsigned int queued;
	18	};
	19
	20	static int fill_timespec(struct timespec *ts)
	21	{
	22	#ifdef CONFIG_CLOCK_GETTIME
	23	#ifdef CONFIG_CLOCK_MONOTONIC
	24	clockid_t clk = CLOCK_MONOTONIC;
	25	#else
	26	clockid_t clk = CLOCK_REALTIME;
	27	#endif
	28	if (!clock_gettime(clk, ts))
	29	return 0;
	30
	31	perror("clock_gettime");
	32	return 1;
	33	#else
	34	struct timeval tv;
	35
	36	gettimeofday(&tv, NULL);
	37	ts->tv_sec = tv.tv_sec;
	38	ts->tv_nsec = tv.tv_usec * 1000;
	39	return 0;
	40	#endif
	41	}
	42
	43	static unsigned long long ts_utime_since_now(struct timespec *t)
	44	{
	45	long long sec, nsec;
	46	struct timespec now;
	47
	48	if (fill_timespec(&now))
	49	return 0;
	50
	51	sec = now.tv_sec - t->tv_sec;
	52	nsec = now.tv_nsec - t->tv_nsec;
	53	if (sec > 0 && nsec < 0) {
	54	sec--;
	55	nsec += 1000000000;
	56	}
	57
	58	sec *= 1000000;
	59	nsec /= 1000;
	60	return sec + nsec;
	61	}
	62
	63	static int fio_posixaio_cancel(struct thread_data fio_unused *td,
	64	struct io_u *io_u)
	65	{
	66	struct fio_file *f = io_u->file;
	67	int r = aio_cancel(f->fd, &io_u->aiocb);
	68
	69	if (r == AIO_ALLDONE \|\| r == AIO_CANCELED)
	70	return 0;
	71
	72	return 1;
	73	}
	74
	75	static int fio_posixaio_prep(struct thread_data fio_unused *td,
	76	struct io_u *io_u)
	77	{
	78	os_aiocb_t *aiocb = &io_u->aiocb;
	79	struct fio_file *f = io_u->file;
	80
	81	aiocb->aio_fildes = f->fd;
	82	aiocb->aio_buf = io_u->xfer_buf;
	83	aiocb->aio_nbytes = io_u->xfer_buflen;
	84	aiocb->aio_offset = io_u->offset;
	85	aiocb->aio_sigevent.sigev_notify = SIGEV_NONE;
	86
	87	io_u->seen = 0;
	88	return 0;
	89	}
	90
	91	#define SUSPEND_ENTRIES 8
	92
	93	static int fio_posixaio_getevents(struct thread_data *td, unsigned int min,
	94	unsigned int max, struct timespec *t)
	95	{
	96	struct posixaio_data *pd = td->io_ops->data;
	97	os_aiocb_t *suspend_list[SUSPEND_ENTRIES];
	98	struct flist_head *entry;
	99	struct timespec start;
	100	int have_timeout = 0;
	101	int suspend_entries = 0;
	102	unsigned int r;
	103
	104	if (t && !fill_timespec(&start))
	105	have_timeout = 1;
	106	else
	107	memset(&start, 0, sizeof(start));
	108
	109	r = 0;
	110	memset(suspend_list, 0, sizeof(*suspend_list));
	111	restart:
	112	flist_for_each(entry, &td->io_u_busylist) {
	113	struct io_u *io_u = flist_entry(entry, struct io_u, list);
	114	int err;
	115
	116	if (io_u->seen)
	117	continue;
	118
	119	err = aio_error(&io_u->aiocb);
	120	if (err == EINPROGRESS) {
	121	if (suspend_entries < SUSPEND_ENTRIES) {
	122	suspend_list[suspend_entries] = &io_u->aiocb;
	123	suspend_entries++;
	124	}
	125	continue;
	126	}
	127
	128	io_u->seen = 1;
	129	pd->queued--;
	130	pd->aio_events[r++] = io_u;
	131
	132	if (err == ECANCELED)
	133	io_u->resid = io_u->xfer_buflen;
	134	else if (!err) {
	135	ssize_t retval = aio_return(&io_u->aiocb);
	136
	137	io_u->resid = io_u->xfer_buflen - retval;
	138	} else
	139	io_u->error = err;
	140	}
	141
	142	if (r >= min)
	143	return r;
	144
	145	if (have_timeout) {
	146	unsigned long long usec;
	147
	148	usec = (t->tv_sec * 1000000) + (t->tv_nsec / 1000);
	149	if (ts_utime_since_now(&start) > usec)
	150	return r;
	151	}
	152
	153	/*
	154	* must have some in-flight, wait for at least one
	155	*/
	156	aio_suspend((const os_aiocb_t * const *)suspend_list,
	157	suspend_entries, t);
	158	goto restart;
	159	}
	160
	161	static struct io_u fio_posixaio_event(struct thread_data td, int event)
	162	{
	163	struct posixaio_data *pd = td->io_ops->data;
	164
	165	return pd->aio_events[event];
	166	}
	167
	168	static int fio_posixaio_queue(struct thread_data *td,
	169	struct io_u *io_u)
	170	{
	171	struct posixaio_data *pd = td->io_ops->data;
	172	os_aiocb_t *aiocb = &io_u->aiocb;
	173	int ret;
	174
	175	fio_ro_check(td, io_u);
	176
	177	if (io_u->ddir == DDIR_READ)
	178	ret = aio_read(aiocb);
	179	else if (io_u->ddir == DDIR_WRITE)
	180	ret = aio_write(aiocb);
	181	else if (io_u->ddir == DDIR_TRIM) {
	182	if (pd->queued)
	183	return FIO_Q_BUSY;
	184
	185	do_io_u_trim(td, io_u);
	186	return FIO_Q_COMPLETED;
	187	} else {
	188	#ifdef CONFIG_POSIXAIO_FSYNC
	189	ret = aio_fsync(O_SYNC, aiocb);
	190	#else
	191	if (pd->queued)
	192	return FIO_Q_BUSY;
	193
	194	do_io_u_sync(td, io_u);
	195	return FIO_Q_COMPLETED;
	196	#endif
	197	}
	198
	199	if (ret) {
	200	/*
	201	* At least OSX has a very low limit on the number of pending
	202	* IOs, so if it returns EAGAIN, we are out of resources
	203	* to queue more. Just return FIO_Q_BUSY to naturally
	204	* drop off at this depth.
	205	*/
	206	if (errno == EAGAIN)
	207	return FIO_Q_BUSY;
	208
	209	io_u->error = errno;
	210	td_verror(td, io_u->error, "xfer");
	211	return FIO_Q_COMPLETED;
	212	}
	213
	214	pd->queued++;
	215	return FIO_Q_QUEUED;
	216	}
	217
	218	static void fio_posixaio_cleanup(struct thread_data *td)
	219	{
	220	struct posixaio_data *pd = td->io_ops->data;
	221
	222	if (pd) {
	223	free(pd->aio_events);
	224	free(pd);
	225	}
	226	}
	227
	228	static int fio_posixaio_init(struct thread_data *td)
	229	{
	230	struct posixaio_data pd = malloc(sizeof(pd));
	231
	232	memset(pd, 0, sizeof(*pd));
	233	pd->aio_events = malloc(td->o.iodepth * sizeof(struct io_u *));
	234	memset(pd->aio_events, 0, td->o.iodepth * sizeof(struct io_u *));
	235
	236	td->io_ops->data = pd;
	237	return 0;
	238	}
	239
	240	static struct ioengine_ops ioengine = {
	241	.name = "posixaio",
	242	.version = FIO_IOOPS_VERSION,
	243	.init = fio_posixaio_init,
	244	.prep = fio_posixaio_prep,
	245	.queue = fio_posixaio_queue,
	246	.cancel = fio_posixaio_cancel,
	247	.getevents = fio_posixaio_getevents,
	248	.event = fio_posixaio_event,
	249	.cleanup = fio_posixaio_cleanup,
	250	.open_file = generic_open_file,
	251	.close_file = generic_close_file,
	252	.get_file_size = generic_get_file_size,
	253	};
	254
	255	static void fio_init fio_posixaio_register(void)
	256	{
	257	register_ioengine(&ioengine);
	258	}
	259
	260	static void fio_exit fio_posixaio_unregister(void)
	261	{
	262	unregister_ioengine(&ioengine);
	263	}