[fio.git] / engines / syslet-rw.c

/*
 * syslet engine
 *
 * IO engine that does regular pread(2)/pwrite(2) to transfer data, but
 * with syslets to make the execution async.
 *
 */
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <errno.h>
#include <assert.h>
#include <asm/unistd.h>

#include "../fio.h"
#include "../os.h"

#ifdef FIO_HAVE_SYSLET

struct syslet_data {
	struct io_u **events;
	unsigned int nr_events;
	
	struct async_head_user ahu;
	struct syslet_uatom **ring;

	struct syslet_uatom *head, *tail;
};

static void fio_syslet_complete_atom(struct thread_data *td,
				     struct syslet_uatom *atom)
{
	struct syslet_data *sd = td->io_ops->data;
	struct syslet_uatom *last;
	struct io_u *io_u;

	/*
	 * complete from the beginning of the sequence up to (and
	 * including) this atom
	 */
	last = atom;
	io_u = atom->private;
	atom = io_u->req.head;

	/*
	 * now complete in right order
	 */
	do {
		long ret;

		io_u = atom->private;
		ret = *atom->ret_ptr;
		if (ret >= 0)
			io_u->resid = io_u->xfer_buflen - ret;
		else if (ret < 0)
			io_u->error = ret;

		assert(sd->nr_events < td->iodepth);
		sd->events[sd->nr_events++] = io_u;

		if (atom == last)
			break;

		atom = atom->next;
	} while (1);

	assert(!last->next);
}

/*
 * Inspect the ring to see if we have completed events
 */
static void fio_syslet_complete(struct thread_data *td)
{
	struct syslet_data *sd = td->io_ops->data;

	do {
		struct syslet_uatom *atom;

		atom = sd->ring[sd->ahu.user_ring_idx];
		if (!atom)
			break;

		sd->ring[sd->ahu.user_ring_idx] = NULL;
		if (++sd->ahu.user_ring_idx == td->iodepth)
			sd->ahu.user_ring_idx = 0;

		fio_syslet_complete_atom(td, atom);
	} while (1);
}

static int fio_syslet_getevents(struct thread_data *td, int min,
				int fio_unused max,
				struct timespec fio_unused *t)
{
	struct syslet_data *sd = td->io_ops->data;
	long ret;

	do {
		fio_syslet_complete(td);

		/*
		 * do we have enough immediate completions?
		 */
		if (sd->nr_events >= (unsigned int) min)
			break;

		/*
		 * OK, we need to wait for some events...
		 */
		ret = async_wait(1, sd->ahu.user_ring_idx, &sd->ahu);
		if (ret < 0)
			return -errno;
	} while (1);

	ret = sd->nr_events;
	sd->nr_events = 0;
	return ret;
}

static struct io_u *fio_syslet_event(struct thread_data *td, int event)
{
	struct syslet_data *sd = td->io_ops->data;

	return sd->events[event];
}

static void init_atom(struct syslet_uatom *atom, int nr, void *arg0,
		      void *arg1, void *arg2, void *arg3, void *ret_ptr,
		      unsigned long flags, void *priv)
{
	atom->flags = flags;
	atom->nr = nr;
	atom->ret_ptr = ret_ptr;
	atom->next = NULL;
	atom->arg_ptr[0] = arg0;
	atom->arg_ptr[1] = arg1;
	atom->arg_ptr[2] = arg2;
	atom->arg_ptr[3] = arg3;
	atom->arg_ptr[4] = atom->arg_ptr[5] = NULL;
	atom->private = priv;
}

/*
 * Use seek atom for sync
 */
static void fio_syslet_prep_sync(struct io_u *io_u, struct fio_file *f)
{
	init_atom(&io_u->req.atom, __NR_fsync, &f->fd, NULL, NULL, NULL,
		  &io_u->req.ret, 0, io_u);
}

static void fio_syslet_prep_rw(struct io_u *io_u, struct fio_file *f)
{
	int nr;

	/*
	 * prepare rw
	 */
	if (io_u->ddir == DDIR_READ)
		nr = __NR_pread64;
	else
		nr = __NR_pwrite64;

	init_atom(&io_u->req.atom, nr, &f->fd, &io_u->xfer_buf,
		  &io_u->xfer_buflen, &io_u->offset, &io_u->req.ret, 0, io_u);
}

static int fio_syslet_prep(struct thread_data fio_unused *td, struct io_u *io_u)
{
	struct fio_file *f = io_u->file;

	if (io_u->ddir == DDIR_SYNC)
		fio_syslet_prep_sync(io_u, f);
	else
		fio_syslet_prep_rw(io_u, f);

	return 0;
}

static void cachemiss_thread_start(void)
{
	while (1)
		async_thread(NULL, NULL);
}

#define THREAD_STACK_SIZE (16384)

static unsigned long thread_stack_alloc()
{
	return (unsigned long) malloc(THREAD_STACK_SIZE) + THREAD_STACK_SIZE;
}

static void fio_syslet_queued(struct thread_data *td, struct syslet_data *sd)
{
	struct syslet_uatom *atom;
	struct timeval now;

	fio_gettime(&now, NULL);

	atom = sd->head;
	while (atom) {
		struct io_u *io_u = atom->private;

		memcpy(&io_u->issue_time, &now, sizeof(now));
		io_u_queued(td, io_u);
		atom = atom->next;
	}
}

static int fio_syslet_commit(struct thread_data *td)
{
	struct syslet_data *sd = td->io_ops->data;
	struct syslet_uatom *done;

	if (!sd->head)
		return 0;

	assert(!sd->tail->next);

	if (!sd->ahu.new_thread_stack)
		sd->ahu.new_thread_stack = thread_stack_alloc();

	fio_syslet_queued(td, sd);

	/*
	 * On sync completion, the atom is returned. So on NULL return
	 * it's queued asynchronously.
	 */
	done = async_exec(sd->head, &sd->ahu);

	sd->head = sd->tail = NULL;

	if (done)
		fio_syslet_complete_atom(td, done);

	return 0;
}

static int fio_syslet_queue(struct thread_data *td, struct io_u *io_u)
{
	struct syslet_data *sd = td->io_ops->data;

	if (sd->tail) {
		sd->tail->next = &io_u->req.atom;
		sd->tail = &io_u->req.atom;
	} else
		sd->head = sd->tail = &io_u->req.atom;

	io_u->req.head = sd->head;
	return FIO_Q_QUEUED;
}

static int async_head_init(struct syslet_data *sd, unsigned int depth)
{
	unsigned long ring_size;

	memset(&sd->ahu, 0, sizeof(struct async_head_user));

	ring_size = sizeof(struct syslet_uatom *) * depth;
	sd->ring = malloc(ring_size);
	memset(sd->ring, 0, ring_size);

	sd->ahu.user_ring_idx = 0;
	sd->ahu.completion_ring = sd->ring;
	sd->ahu.ring_size_bytes = ring_size;
	sd->ahu.head_stack = thread_stack_alloc();
	sd->ahu.head_eip = (unsigned long) cachemiss_thread_start;
	sd->ahu.new_thread_eip = (unsigned long) cachemiss_thread_start;

	return 0;
}

static void async_head_exit(struct syslet_data *sd)
{
	free(sd->ring);
}

static void fio_syslet_cleanup(struct thread_data *td)
{
	struct syslet_data *sd = td->io_ops->data;

	if (sd) {
		async_head_exit(sd);
		free(sd->events);
		free(sd);
		td->io_ops->data = NULL;
	}
}

static int fio_syslet_init(struct thread_data *td)
{
	struct syslet_data *sd;


	sd = malloc(sizeof(*sd));
	memset(sd, 0, sizeof(*sd));
	sd->events = malloc(sizeof(struct io_u *) * td->iodepth);
	memset(sd->events, 0, sizeof(struct io_u *) * td->iodepth);

	/*
	 * This will handily fail for kernels where syslet isn't available
	 */
	if (async_head_init(sd, td->iodepth)) {
		free(sd->events);
		free(sd);
		return 1;
	}

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

static struct ioengine_ops ioengine = {
	.name		= "syslet-rw",
	.version	= FIO_IOOPS_VERSION,
	.init		= fio_syslet_init,
	.prep		= fio_syslet_prep,
	.queue		= fio_syslet_queue,
	.commit		= fio_syslet_commit,
	.getevents	= fio_syslet_getevents,
	.event		= fio_syslet_event,
	.cleanup	= fio_syslet_cleanup,
	.open_file	= generic_open_file,
	.close_file	= generic_close_file,
};

#else /* FIO_HAVE_SYSLET */

/*
 * When we have a proper configure system in place, we simply wont build
 * and install this io engine. For now install a crippled version that
 * just complains and fails to load.
 */
static int fio_syslet_init(struct thread_data fio_unused *td)
{
	fprintf(stderr, "fio: syslet not available\n");
	return 1;
}

static struct ioengine_ops ioengine = {
	.name		= "syslet-rw",
	.version	= FIO_IOOPS_VERSION,
	.init		= fio_syslet_init,
};

#endif /* FIO_HAVE_SYSLET */

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

static void fio_exit fio_syslet_unregister(void)
{
	unregister_ioengine(&ioengine);
}
Commit	Line	Data
	1	/*
	2	* syslet engine
	3	*
	4	* IO engine that does regular pread(2)/pwrite(2) to transfer data, but
	5	* with syslets to make the execution async.
	6	*
	7	*/
	8	#include <stdio.h>
	9	#include <stdlib.h>
	10	#include <unistd.h>
	11	#include <errno.h>
	12	#include <assert.h>
	13	#include <asm/unistd.h>
	14
	15	#include "../fio.h"
	16	#include "../os.h"
	17
	18	#ifdef FIO_HAVE_SYSLET
	19
	20	struct syslet_data {
	21	struct io_u **events;
	22	unsigned int nr_events;
	23
	24	struct async_head_user ahu;
	25	struct syslet_uatom **ring;
	26
	27	struct syslet_uatom head, tail;
	28	};
	29
	30	static void fio_syslet_complete_atom(struct thread_data *td,
	31	struct syslet_uatom *atom)
	32	{
	33	struct syslet_data *sd = td->io_ops->data;
	34	struct syslet_uatom *last;
	35	struct io_u *io_u;
	36
	37	/*
	38	* complete from the beginning of the sequence up to (and
	39	* including) this atom
	40	*/
	41	last = atom;
	42	io_u = atom->private;
	43	atom = io_u->req.head;
	44
	45	/*
	46	* now complete in right order
	47	*/
	48	do {
	49	long ret;
	50
	51	io_u = atom->private;
	52	ret = *atom->ret_ptr;
	53	if (ret >= 0)
	54	io_u->resid = io_u->xfer_buflen - ret;
	55	else if (ret < 0)
	56	io_u->error = ret;
	57
	58	assert(sd->nr_events < td->iodepth);
	59	sd->events[sd->nr_events++] = io_u;
	60
	61	if (atom == last)
	62	break;
	63
	64	atom = atom->next;
	65	} while (1);
	66
	67	assert(!last->next);
	68	}
	69
	70	/*
	71	* Inspect the ring to see if we have completed events
	72	*/
	73	static void fio_syslet_complete(struct thread_data *td)
	74	{
	75	struct syslet_data *sd = td->io_ops->data;
	76
	77	do {
	78	struct syslet_uatom *atom;
	79
	80	atom = sd->ring[sd->ahu.user_ring_idx];
	81	if (!atom)
	82	break;
	83
	84	sd->ring[sd->ahu.user_ring_idx] = NULL;
	85	if (++sd->ahu.user_ring_idx == td->iodepth)
	86	sd->ahu.user_ring_idx = 0;
	87
	88	fio_syslet_complete_atom(td, atom);
	89	} while (1);
	90	}
	91
	92	static int fio_syslet_getevents(struct thread_data *td, int min,
	93	int fio_unused max,
	94	struct timespec fio_unused *t)
	95	{
	96	struct syslet_data *sd = td->io_ops->data;
	97	long ret;
	98
	99	do {
	100	fio_syslet_complete(td);
	101
	102	/*
	103	* do we have enough immediate completions?
	104	*/
	105	if (sd->nr_events >= (unsigned int) min)
	106	break;
	107
	108	/*
	109	* OK, we need to wait for some events...
	110	*/
	111	ret = async_wait(1, sd->ahu.user_ring_idx, &sd->ahu);
	112	if (ret < 0)
	113	return -errno;
	114	} while (1);
	115
	116	ret = sd->nr_events;
	117	sd->nr_events = 0;
	118	return ret;
	119	}
	120
	121	static struct io_u fio_syslet_event(struct thread_data td, int event)
	122	{
	123	struct syslet_data *sd = td->io_ops->data;
	124
	125	return sd->events[event];
	126	}
	127
	128	static void init_atom(struct syslet_uatom atom, int nr, void arg0,
	129	void arg1, void arg2, void arg3, void ret_ptr,
	130	unsigned long flags, void *priv)
	131	{
	132	atom->flags = flags;
	133	atom->nr = nr;
	134	atom->ret_ptr = ret_ptr;
	135	atom->next = NULL;
	136	atom->arg_ptr[0] = arg0;
	137	atom->arg_ptr[1] = arg1;
	138	atom->arg_ptr[2] = arg2;
	139	atom->arg_ptr[3] = arg3;
	140	atom->arg_ptr[4] = atom->arg_ptr[5] = NULL;
	141	atom->private = priv;
	142	}
	143
	144	/*
	145	* Use seek atom for sync
	146	*/
	147	static void fio_syslet_prep_sync(struct io_u io_u, struct fio_file f)
	148	{
	149	init_atom(&io_u->req.atom, __NR_fsync, &f->fd, NULL, NULL, NULL,
	150	&io_u->req.ret, 0, io_u);
	151	}
	152
	153	static void fio_syslet_prep_rw(struct io_u io_u, struct fio_file f)
	154	{
	155	int nr;
	156
	157	/*
	158	* prepare rw
	159	*/
	160	if (io_u->ddir == DDIR_READ)
	161	nr = __NR_pread64;
	162	else
	163	nr = __NR_pwrite64;
	164
	165	init_atom(&io_u->req.atom, nr, &f->fd, &io_u->xfer_buf,
	166	&io_u->xfer_buflen, &io_u->offset, &io_u->req.ret, 0, io_u);
	167	}
	168
	169	static int fio_syslet_prep(struct thread_data fio_unused td, struct io_u io_u)
	170	{
	171	struct fio_file *f = io_u->file;
	172
	173	if (io_u->ddir == DDIR_SYNC)
	174	fio_syslet_prep_sync(io_u, f);
	175	else
	176	fio_syslet_prep_rw(io_u, f);
	177
	178	return 0;
	179	}
	180
	181	static void cachemiss_thread_start(void)
	182	{
	183	while (1)
	184	async_thread(NULL, NULL);
	185	}
	186
	187	#define THREAD_STACK_SIZE (16384)
	188
	189	static unsigned long thread_stack_alloc()
	190	{
	191	return (unsigned long) malloc(THREAD_STACK_SIZE) + THREAD_STACK_SIZE;
	192	}
	193
	194	static void fio_syslet_queued(struct thread_data td, struct syslet_data sd)
	195	{
	196	struct syslet_uatom *atom;
	197	struct timeval now;
	198
	199	fio_gettime(&now, NULL);
	200
	201	atom = sd->head;
	202	while (atom) {
	203	struct io_u *io_u = atom->private;
	204
	205	memcpy(&io_u->issue_time, &now, sizeof(now));
	206	io_u_queued(td, io_u);
	207	atom = atom->next;
	208	}
	209	}
	210
	211	static int fio_syslet_commit(struct thread_data *td)
	212	{
	213	struct syslet_data *sd = td->io_ops->data;
	214	struct syslet_uatom *done;
	215
	216	if (!sd->head)
	217	return 0;
	218
	219	assert(!sd->tail->next);
	220
	221	if (!sd->ahu.new_thread_stack)
	222	sd->ahu.new_thread_stack = thread_stack_alloc();
	223
	224	fio_syslet_queued(td, sd);
	225
	226	/*
	227	* On sync completion, the atom is returned. So on NULL return
	228	* it's queued asynchronously.
	229	*/
	230	done = async_exec(sd->head, &sd->ahu);
	231
	232	sd->head = sd->tail = NULL;
	233
	234	if (done)
	235	fio_syslet_complete_atom(td, done);
	236
	237	return 0;
	238	}
	239
	240	static int fio_syslet_queue(struct thread_data td, struct io_u io_u)
	241	{
	242	struct syslet_data *sd = td->io_ops->data;
	243
	244	if (sd->tail) {
	245	sd->tail->next = &io_u->req.atom;
	246	sd->tail = &io_u->req.atom;
	247	} else
	248	sd->head = sd->tail = &io_u->req.atom;
	249
	250	io_u->req.head = sd->head;
	251	return FIO_Q_QUEUED;
	252	}
	253
	254	static int async_head_init(struct syslet_data *sd, unsigned int depth)
	255	{
	256	unsigned long ring_size;
	257
	258	memset(&sd->ahu, 0, sizeof(struct async_head_user));
	259
	260	ring_size = sizeof(struct syslet_uatom ) depth;
	261	sd->ring = malloc(ring_size);
	262	memset(sd->ring, 0, ring_size);
	263
	264	sd->ahu.user_ring_idx = 0;
	265	sd->ahu.completion_ring = sd->ring;
	266	sd->ahu.ring_size_bytes = ring_size;
	267	sd->ahu.head_stack = thread_stack_alloc();
	268	sd->ahu.head_eip = (unsigned long) cachemiss_thread_start;
	269	sd->ahu.new_thread_eip = (unsigned long) cachemiss_thread_start;
	270
	271	return 0;
	272	}
	273
	274	static void async_head_exit(struct syslet_data *sd)
	275	{
	276	free(sd->ring);
	277	}
	278
	279	static void fio_syslet_cleanup(struct thread_data *td)
	280	{
	281	struct syslet_data *sd = td->io_ops->data;
	282
	283	if (sd) {
	284	async_head_exit(sd);
	285	free(sd->events);
	286	free(sd);
	287	td->io_ops->data = NULL;
	288	}
	289	}
	290
	291	static int fio_syslet_init(struct thread_data *td)
	292	{
	293	struct syslet_data *sd;
	294
	295
	296	sd = malloc(sizeof(*sd));
	297	memset(sd, 0, sizeof(*sd));
	298	sd->events = malloc(sizeof(struct io_u ) td->iodepth);
	299	memset(sd->events, 0, sizeof(struct io_u ) td->iodepth);
	300
	301	/*
	302	* This will handily fail for kernels where syslet isn't available
	303	*/
	304	if (async_head_init(sd, td->iodepth)) {
	305	free(sd->events);
	306	free(sd);
	307	return 1;
	308	}
	309
	310	td->io_ops->data = sd;
	311	return 0;
	312	}
	313
	314	static struct ioengine_ops ioengine = {
	315	.name = "syslet-rw",
	316	.version = FIO_IOOPS_VERSION,
	317	.init = fio_syslet_init,
	318	.prep = fio_syslet_prep,
	319	.queue = fio_syslet_queue,
	320	.commit = fio_syslet_commit,
	321	.getevents = fio_syslet_getevents,
	322	.event = fio_syslet_event,
	323	.cleanup = fio_syslet_cleanup,
	324	.open_file = generic_open_file,
	325	.close_file = generic_close_file,
	326	};
	327
	328	#else /* FIO_HAVE_SYSLET */
	329
	330	/*
	331	* When we have a proper configure system in place, we simply wont build
	332	* and install this io engine. For now install a crippled version that
	333	* just complains and fails to load.
	334	*/
	335	static int fio_syslet_init(struct thread_data fio_unused *td)
	336	{
	337	fprintf(stderr, "fio: syslet not available\n");
	338	return 1;
	339	}
	340
	341	static struct ioengine_ops ioengine = {
	342	.name = "syslet-rw",
	343	.version = FIO_IOOPS_VERSION,
	344	.init = fio_syslet_init,
	345	};
	346
	347	#endif /* FIO_HAVE_SYSLET */
	348
	349	static void fio_init fio_syslet_register(void)
	350	{
	351	register_ioengine(&ioengine);
	352	}
	353
	354	static void fio_exit fio_syslet_unregister(void)
	355	{
	356	unregister_ioengine(&ioengine);
	357	}