[fio.git] / engines / nfs.c

#include <stdlib.h>
#include <poll.h>
#include <nfsc/libnfs.h>
#include <nfsc/libnfs-raw.h>
#include <nfsc/libnfs-raw-mount.h>

#include "../fio.h"
#include "../optgroup.h"

enum nfs_op_type {
	NFS_READ_WRITE = 0,
	NFS_STAT_MKDIR_RMDIR,
	NFS_STAT_TOUCH_RM,
};

struct fio_libnfs_options {
	struct nfs_context *context;
	char *nfs_url;
	/* nfs_callback needs this info, but doesn't have fio td structure to
	 * pull it from
	 */
	unsigned int queue_depth;

	/* the following implement a circular queue of outstanding IOs */

	/* IOs issued to libnfs, that have not returned yet */
	int outstanding_events;
	/* event last returned via fio_libnfs_event */
	int prev_requested_event_index;
	int next_buffered_event; /* round robin-pointer within events[] */
	int buffered_event_count; /* IOs completed by libnfs, waiting for FIO */
	int free_event_buffer_index; /* next free buffer */
	struct io_u**events;
};

struct nfs_data {
	struct nfsfh *nfsfh;
	struct fio_libnfs_options *options;
};

static struct fio_option options[] = {
	{
		.name	= "nfs_url",
		.lname	= "nfs_url",
		.type	= FIO_OPT_STR_STORE,
		.help	= "URL in libnfs format, eg nfs://<server|ipv4|"
			  "ipv6>/path[?arg=val[&arg=val]*]",
		.off1	= offsetof(struct fio_libnfs_options, nfs_url),
		.category = FIO_OPT_C_ENGINE,
		.group	= __FIO_OPT_G_NFS,
	},
	{
		.name     = NULL,
	},
};

static struct io_u *fio_libnfs_event(struct thread_data *td, int event)
{
	struct fio_libnfs_options *o = td->eo;
	struct io_u *io_u = o->events[o->next_buffered_event];

	assert(o->events[o->next_buffered_event]);
	o->events[o->next_buffered_event] = NULL;
	o->next_buffered_event = (o->next_buffered_event + 1) % td->o.iodepth;

	/* validate our state machine */
	assert(o->buffered_event_count);
	o->buffered_event_count--;
	assert(io_u);

	/* assert that fio_libnfs_event is being called in sequential fashion */
	assert(event == 0 || o->prev_requested_event_index + 1 == event);
	if (o->buffered_event_count == 0)
		o->prev_requested_event_index = -1;
	else
		o->prev_requested_event_index = event;
	return io_u;
}

/*
 * fio core logic seems to stop calling this event-loop if we ever return with
 * 0 events
 */
#define SHOULD_WAIT(td, o, flush)			\
 	((o)->outstanding_events == (td)->o.iodepth ||	\
		(flush && (o)->outstanding_events))

static int nfs_event_loop(struct thread_data *td, bool flush)
{
	struct fio_libnfs_options *o = td->eo;
	struct pollfd pfds[1]; /* nfs:0 */

	/* we already have stuff queued for fio, no need to waste cpu on poll() */
	if (o->buffered_event_count)
		return o->buffered_event_count;

	do {
		int timeout = SHOULD_WAIT(td, o, flush) ? -1 : 0;
		int ret = 0;

		pfds[0].fd = nfs_get_fd(o->context);
		pfds[0].events = nfs_which_events(o->context);
		ret = poll(&pfds[0], 1, timeout);
		if (ret < 0) {
			if (errno == EINTR || errno == EAGAIN)
				continue;
			log_err("nfs: failed to poll events: %s\n", strerror(errno));
			break;
		}

		ret = nfs_service(o->context, pfds[0].revents);
		if (ret < 0) {
			log_err("nfs: socket is in an unrecoverable error state.\n");
			break;
		}
	} while (SHOULD_WAIT(td, o, flush));

	return o->buffered_event_count;
}

static int fio_libnfs_getevents(struct thread_data *td, unsigned int min,
				unsigned int max, const struct timespec *t)
{
	return nfs_event_loop(td, false);
}

static void nfs_callback(int res, struct nfs_context *nfs, void *data,
			 void *private_data)
{
	struct io_u *io_u = private_data;
	struct nfs_data *nfs_data = io_u->file->engine_data;
	struct fio_libnfs_options *o = nfs_data->options;
	if (res < 0) {
		log_err("Failed NFS operation(code:%d): %s\n", res,
						nfs_get_error(o->context));
		io_u->error = -res;
		/* res is used for read math below, don't want to pass negative
		 * there
		 */
		res = 0;
	} else if (io_u->ddir == DDIR_READ) {
		memcpy(io_u->buf, data, res);
		if (res == 0)
			log_err("Got NFS EOF, this is probably not expected\n");
	}
	/* fio uses resid to track remaining data */
	io_u->resid = io_u->xfer_buflen - res;

	assert(!o->events[o->free_event_buffer_index]);
	o->events[o->free_event_buffer_index] = io_u;
	o->free_event_buffer_index = (o->free_event_buffer_index + 1) % o->queue_depth;
	o->outstanding_events--;
	o->buffered_event_count++;
}

static int queue_write(struct fio_libnfs_options *o, struct io_u *io_u)
{
	struct nfs_data *nfs_data = io_u->engine_data;

	return nfs_pwrite_async(o->context, nfs_data->nfsfh, io_u->offset,
				io_u->buflen, io_u->buf, nfs_callback, io_u);
}

static int queue_read(struct fio_libnfs_options *o, struct io_u *io_u)
{
	struct nfs_data *nfs_data = io_u->engine_data;

	return nfs_pread_async(o->context, nfs_data->nfsfh, io_u->offset,
				io_u->buflen, nfs_callback, io_u);
}

static enum fio_q_status fio_libnfs_queue(struct thread_data *td,
					  struct io_u *io_u)
{
	struct nfs_data *nfs_data = io_u->file->engine_data;
	struct fio_libnfs_options *o = nfs_data->options;
	struct nfs_context *nfs = o->context;
	enum fio_q_status ret = FIO_Q_QUEUED;
	int err;

	io_u->engine_data = nfs_data;
	switch (io_u->ddir) {
	case DDIR_WRITE:
		err = queue_write(o, io_u);
		break;
	case DDIR_READ:
		err = queue_read(o, io_u);
		break;
	case DDIR_TRIM:
		log_err("nfs: trim is not supported");
		err = -1;
		break;
	default:
		log_err("nfs: unhandled io %d\n", io_u->ddir);
		err = -1;
	}
	if (err) {
		log_err("nfs: Failed to queue nfs op: %s\n", nfs_get_error(nfs));
		td->error = 1;
		return FIO_Q_COMPLETED;
	}
	o->outstanding_events++;
	return ret;
}

/*
 * Do a mount if one has not been done before 
 */
static int do_mount(struct thread_data *td, const char *url)
{
	size_t event_size = sizeof(struct io_u **) * td->o.iodepth;
	struct fio_libnfs_options *options = td->eo;
	struct nfs_url *nfs_url = NULL;
	int ret = 0;
	int path_len = 0;
	char *mnt_dir = NULL;

	if (options->context)
		return 0;

	options->context = nfs_init_context();
	if (!options->context) {
		log_err("nfs: failed to init nfs context\n");
		return -1;
	}

	options->events = malloc(event_size);
	memset(options->events, 0, event_size);

	options->prev_requested_event_index = -1;
	options->queue_depth = td->o.iodepth;

	nfs_url = nfs_parse_url_full(options->context, url);
	path_len = strlen(nfs_url->path);
	mnt_dir = malloc(path_len + strlen(nfs_url->file) + 1);
	strcpy(mnt_dir, nfs_url->path);
	strcpy(mnt_dir + strlen(nfs_url->path), nfs_url->file);
	ret = nfs_mount(options->context, nfs_url->server, mnt_dir);
	free(mnt_dir);
	nfs_destroy_url(nfs_url);
	return ret;
}

static int fio_libnfs_setup(struct thread_data *td)
{
	/* Using threads with libnfs causes fio to hang on exit, lower
	 * performance
	 */
	td->o.use_thread = 0;
	return 0;
}

static void fio_libnfs_cleanup(struct thread_data *td)
{
	struct fio_libnfs_options *o = td->eo;

	nfs_umount(o->context);
	nfs_destroy_context(o->context);
	free(o->events);
}

static int fio_libnfs_open(struct thread_data *td, struct fio_file *f)
{
	struct fio_libnfs_options *options = td->eo;
	struct nfs_data *nfs_data = NULL;
	int flags = 0;
	int ret;

	if (!options->nfs_url) {
		log_err("nfs: nfs_url is a required parameter\n");
		return -1;
	}

	ret = do_mount(td, options->nfs_url);

	if (ret) {
		log_err("nfs: Failed to mount %s with code %d: %s\n",
			options->nfs_url, ret, nfs_get_error(options->context));
		return ret;
	}
	nfs_data = malloc(sizeof(struct nfs_data));
	memset(nfs_data, 0, sizeof(struct nfs_data));
	nfs_data->options = options;

	if (td->o.td_ddir == TD_DDIR_WRITE)
		flags |= O_CREAT | O_RDWR;
	else
		flags |= O_RDWR;

	ret = nfs_open(options->context, f->file_name, flags, &nfs_data->nfsfh);

	if (ret)
		log_err("Failed to open %s: %s\n", f->file_name,
					nfs_get_error(options->context));
	f->engine_data = nfs_data;
	return ret;
}

static int fio_libnfs_close(struct thread_data *td, struct fio_file *f)
{
	struct nfs_data *nfs_data = f->engine_data;
	struct fio_libnfs_options *o = nfs_data->options;
	int ret = 0;

	if (nfs_data->nfsfh)
		ret = nfs_close(o->context, nfs_data->nfsfh);

	free(nfs_data);
	f->engine_data = NULL;
	return ret;
}

struct ioengine_ops ioengine = {
	.name		= "nfs",
	.version	= FIO_IOOPS_VERSION,
	.setup		= fio_libnfs_setup,
	.queue		= fio_libnfs_queue,
	.getevents	= fio_libnfs_getevents,
	.event		= fio_libnfs_event,
	.cleanup	= fio_libnfs_cleanup,
	.open_file	= fio_libnfs_open,
	.close_file	= fio_libnfs_close,
	.flags		= FIO_DISKLESSIO | FIO_NOEXTEND | FIO_NODISKUTIL,
	.options	= options,
	.option_struct_size	= sizeof(struct fio_libnfs_options),
};

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

static void fio_exit fio_nfs_unregister(void)
{
	unregister_ioengine(&ioengine);
}
Commit	Line	Data
9326926b TG	1	#include <stdlib.h>
	2	#include <poll.h>
	3	#include <nfsc/libnfs.h>
	4	#include <nfsc/libnfs-raw.h>
	5	#include <nfsc/libnfs-raw-mount.h>
	6
	7	#include "../fio.h"
	8	#include "../optgroup.h"
	9
	10	enum nfs_op_type {
	11	NFS_READ_WRITE = 0,
	12	NFS_STAT_MKDIR_RMDIR,
	13	NFS_STAT_TOUCH_RM,
	14	};
	15
	16	struct fio_libnfs_options {
	17	struct nfs_context *context;
	18	char *nfs_url;
db64b7b6 JA	19	/* nfs_callback needs this info, but doesn't have fio td structure to
	20	* pull it from
	21	*/
	22	unsigned int queue_depth;
	23
1fb2bc2f	24	/* the following implement a circular queue of outstanding IOs */
db64b7b6 JA	25
	26	/* IOs issued to libnfs, that have not returned yet */
	27	int outstanding_events;
	28	/* event last returned via fio_libnfs_event */
	29	int prev_requested_event_index;
1fb2bc2f TG	30	int next_buffered_event; /* round robin-pointer within events[] */
	31	int buffered_event_count; /* IOs completed by libnfs, waiting for FIO */
	32	int free_event_buffer_index; /* next free buffer */
9326926b TG	33	struct io_u**events;
	34	};
	35
	36	struct nfs_data {
	37	struct nfsfh *nfsfh;
	38	struct fio_libnfs_options *options;
	39	};
	40
	41	static struct fio_option options[] = {
	42	{
db64b7b6 JA	43	.name = "nfs_url",
	44	.lname = "nfs_url",
	45	.type = FIO_OPT_STR_STORE,
	46	.help = "URL in libnfs format, eg nfs://<server\|ipv4\|"
	47	"ipv6>/path[?arg=val[&arg=val]*]",
	48	.off1 = offsetof(struct fio_libnfs_options, nfs_url),
9326926b TG	49	.category = FIO_OPT_C_ENGINE,
	50	.group = __FIO_OPT_G_NFS,
	51	},
	52	{
	53	.name = NULL,
	54	},
	55	};
	56
9326926b TG	57	static struct io_u fio_libnfs_event(struct thread_data td, int event)
	58	{
	59	struct fio_libnfs_options *o = td->eo;
	60	struct io_u *io_u = o->events[o->next_buffered_event];
db64b7b6	61
9326926b TG	62	assert(o->events[o->next_buffered_event]);
	63	o->events[o->next_buffered_event] = NULL;
	64	o->next_buffered_event = (o->next_buffered_event + 1) % td->o.iodepth;
db64b7b6	65
1fb2bc2f	66	/* validate our state machine */
9326926b TG	67	assert(o->buffered_event_count);
	68	o->buffered_event_count--;
	69	assert(io_u);
db64b7b6	70
1fb2bc2f	71	/* assert that fio_libnfs_event is being called in sequential fashion */
9326926b	72	assert(event == 0 \|\| o->prev_requested_event_index + 1 == event);
db64b7b6	73	if (o->buffered_event_count == 0)
9326926b	74	o->prev_requested_event_index = -1;
db64b7b6	75	else
9326926b	76	o->prev_requested_event_index = event;
9326926b TG	77	return io_u;
	78	}
	79
db64b7b6 JA	80	/*
	81	* fio core logic seems to stop calling this event-loop if we ever return with
	82	* 0 events
	83	*/
	84	#define SHOULD_WAIT(td, o, flush) \
	85	((o)->outstanding_events == (td)->o.iodepth \|\| \
	86	(flush && (o)->outstanding_events))
	87
	88	static int nfs_event_loop(struct thread_data *td, bool flush)
	89	{
9326926b TG	90	struct fio_libnfs_options *o = td->eo;
9326926b TG	91	struct pollfd pfds[1]; /* nfs:0 */
db64b7b6	92
1fb2bc2f	93	/* we already have stuff queued for fio, no need to waste cpu on poll() */
7654a8d5	94	if (o->buffered_event_count)
9326926b	95	return o->buffered_event_count;
9326926b TG	96
9326926b TG	97	do {
db64b7b6	98	int timeout = SHOULD_WAIT(td, o, flush) ? -1 : 0;
9326926b	99	int ret = 0;
db64b7b6	100
9326926b TG	101	pfds[0].fd = nfs_get_fd(o->context);
	102	pfds[0].events = nfs_which_events(o->context);
	103	ret = poll(&pfds[0], 1, timeout);
	104	if (ret < 0) {
db64b7b6	105	if (errno == EINTR \|\| errno == EAGAIN)
9326926b	106	continue;
db64b7b6	107	log_err("nfs: failed to poll events: %s\n", strerror(errno));
9326926b TG	108	break;
	109	}
	110
	111	ret = nfs_service(o->context, pfds[0].revents);
	112	if (ret < 0) {
	113	log_err("nfs: socket is in an unrecoverable error state.\n");
	114	break;
	115	}
db64b7b6 JA	116	} while (SHOULD_WAIT(td, o, flush));
db64b7b6 JA	117
9326926b	118	return o->buffered_event_count;
7654a8d5	119	}
9326926b	120
9326926b	121	static int fio_libnfs_getevents(struct thread_data *td, unsigned int min,
db64b7b6	122	unsigned int max, const struct timespec *t)
9326926b TG	123	{
	124	return nfs_event_loop(td, false);
	125	}
	126
	127	static void nfs_callback(int res, struct nfs_context nfs, void data,
db64b7b6	128	void *private_data)
9326926b TG	129	{
	130	struct io_u *io_u = private_data;
	131	struct nfs_data *nfs_data = io_u->file->engine_data;
	132	struct fio_libnfs_options *o = nfs_data->options;
	133	if (res < 0) {
db64b7b6 JA	134	log_err("Failed NFS operation(code:%d): %s\n", res,
db64b7b6 JA	135	nfs_get_error(o->context));
9326926b	136	io_u->error = -res;
db64b7b6 JA	137	/* res is used for read math below, don't want to pass negative
	138	* there
	139	*/
9326926b TG	140	res = 0;
	141	} else if (io_u->ddir == DDIR_READ) {
	142	memcpy(io_u->buf, data, res);
7654a8d5	143	if (res == 0)
9326926b	144	log_err("Got NFS EOF, this is probably not expected\n");
9326926b	145	}
1fb2bc2f	146	/* fio uses resid to track remaining data */
9326926b TG	147	io_u->resid = io_u->xfer_buflen - res;
	148
	149	assert(!o->events[o->free_event_buffer_index]);
	150	o->events[o->free_event_buffer_index] = io_u;
	151	o->free_event_buffer_index = (o->free_event_buffer_index + 1) % o->queue_depth;
	152	o->outstanding_events--;
	153	o->buffered_event_count++;
	154	}
	155
db64b7b6 JA	156	static int queue_write(struct fio_libnfs_options o, struct io_u io_u)
db64b7b6 JA	157	{
9326926b	158	struct nfs_data *nfs_data = io_u->engine_data;
db64b7b6 JA	159
	160	return nfs_pwrite_async(o->context, nfs_data->nfsfh, io_u->offset,
	161	io_u->buflen, io_u->buf, nfs_callback, io_u);
9326926b TG	162	}
9326926b TG	163
db64b7b6 JA	164	static int queue_read(struct fio_libnfs_options o, struct io_u io_u)
db64b7b6 JA	165	{
9326926b	166	struct nfs_data *nfs_data = io_u->engine_data;
db64b7b6 JA	167
	168	return nfs_pread_async(o->context, nfs_data->nfsfh, io_u->offset,
	169	io_u->buflen, nfs_callback, io_u);
9326926b TG	170	}
9326926b TG	171
9326926b	172	static enum fio_q_status fio_libnfs_queue(struct thread_data *td,
db64b7b6	173	struct io_u *io_u)
9326926b TG	174	{
	175	struct nfs_data *nfs_data = io_u->file->engine_data;
	176	struct fio_libnfs_options *o = nfs_data->options;
	177	struct nfs_context *nfs = o->context;
9326926b	178	enum fio_q_status ret = FIO_Q_QUEUED;
db64b7b6	179	int err;
9326926b TG	180
9326926b TG	181	io_u->engine_data = nfs_data;
db64b7b6 JA	182	switch (io_u->ddir) {
	183	case DDIR_WRITE:
	184	err = queue_write(o, io_u);
	185	break;
	186	case DDIR_READ:
	187	err = queue_read(o, io_u);
	188	break;
	189	case DDIR_TRIM:
	190	log_err("nfs: trim is not supported");
	191	err = -1;
	192	break;
	193	default:
	194	log_err("nfs: unhandled io %d\n", io_u->ddir);
	195	err = -1;
9326926b TG	196	}
	197	if (err) {
	198	log_err("nfs: Failed to queue nfs op: %s\n", nfs_get_error(nfs));
	199	td->error = 1;
	200	return FIO_Q_COMPLETED;
	201	}
	202	o->outstanding_events++;
	203	return ret;
	204	}
	205
388f1111 TG	206	/*
	207	* Do a mount if one has not been done before
	208	*/
9326926b TG	209	static int do_mount(struct thread_data td, const char url)
	210	{
	211	size_t event_size = sizeof(struct io_u *) td->o.iodepth;
	212	struct fio_libnfs_options *options = td->eo;
	213	struct nfs_url *nfs_url = NULL;
	214	int ret = 0;
	215	int path_len = 0;
	216	char *mnt_dir = NULL;
	217
7654a8d5	218	if (options->context)
9326926b	219	return 0;
9326926b TG	220
9326926b TG	221	options->context = nfs_init_context();
db64b7b6	222	if (!options->context) {
9326926b TG	223	log_err("nfs: failed to init nfs context\n");
	224	return -1;
	225	}
	226
	227	options->events = malloc(event_size);
	228	memset(options->events, 0, event_size);
	229
	230	options->prev_requested_event_index = -1;
	231	options->queue_depth = td->o.iodepth;
	232
	233	nfs_url = nfs_parse_url_full(options->context, url);
	234	path_len = strlen(nfs_url->path);
	235	mnt_dir = malloc(path_len + strlen(nfs_url->file) + 1);
	236	strcpy(mnt_dir, nfs_url->path);
	237	strcpy(mnt_dir + strlen(nfs_url->path), nfs_url->file);
	238	ret = nfs_mount(options->context, nfs_url->server, mnt_dir);
	239	free(mnt_dir);
	240	nfs_destroy_url(nfs_url);
	241	return ret;
	242	}
	243
9326926b TG	244	static int fio_libnfs_setup(struct thread_data *td)
9326926b TG	245	{
db64b7b6 JA	246	/* Using threads with libnfs causes fio to hang on exit, lower
	247	* performance
	248	*/
9326926b TG	249	td->o.use_thread = 0;
	250	return 0;
	251	}
	252
9326926b TG	253	static void fio_libnfs_cleanup(struct thread_data *td)
	254	{
	255	struct fio_libnfs_options *o = td->eo;
db64b7b6	256
9326926b TG	257	nfs_umount(o->context);
	258	nfs_destroy_context(o->context);
	259	free(o->events);
	260	}
	261
	262	static int fio_libnfs_open(struct thread_data td, struct fio_file f)
	263	{
9326926b TG	264	struct fio_libnfs_options *options = td->eo;
	265	struct nfs_data *nfs_data = NULL;
	266	int flags = 0;
db64b7b6	267	int ret;
9326926b TG	268
	269	if (!options->nfs_url) {
	270	log_err("nfs: nfs_url is a required parameter\n");
	271	return -1;
	272	}
	273
	274	ret = do_mount(td, options->nfs_url);
	275
db64b7b6 JA	276	if (ret) {
	277	log_err("nfs: Failed to mount %s with code %d: %s\n",
	278	options->nfs_url, ret, nfs_get_error(options->context));
9326926b TG	279	return ret;
	280	}
	281	nfs_data = malloc(sizeof(struct nfs_data));
	282	memset(nfs_data, 0, sizeof(struct nfs_data));
	283	nfs_data->options = options;
	284
db64b7b6	285	if (td->o.td_ddir == TD_DDIR_WRITE)
9326926b	286	flags \|= O_CREAT \| O_RDWR;
db64b7b6	287	else
9326926b	288	flags \|= O_RDWR;
db64b7b6	289
9326926b TG	290	ret = nfs_open(options->context, f->file_name, flags, &nfs_data->nfsfh);
9326926b TG	291
db64b7b6 JA	292	if (ret)
	293	log_err("Failed to open %s: %s\n", f->file_name,
	294	nfs_get_error(options->context));
9326926b TG	295	f->engine_data = nfs_data;
	296	return ret;
	297	}
	298
	299	static int fio_libnfs_close(struct thread_data td, struct fio_file f)
	300	{
	301	struct nfs_data *nfs_data = f->engine_data;
	302	struct fio_libnfs_options *o = nfs_data->options;
	303	int ret = 0;
db64b7b6	304
7654a8d5	305	if (nfs_data->nfsfh)
9326926b	306	ret = nfs_close(o->context, nfs_data->nfsfh);
db64b7b6	307
9326926b TG	308	free(nfs_data);
	309	f->engine_data = NULL;
	310	return ret;
	311	}
	312
9326926b TG	313	struct ioengine_ops ioengine = {
	314	.name = "nfs",
	315	.version = FIO_IOOPS_VERSION,
	316	.setup = fio_libnfs_setup,
	317	.queue = fio_libnfs_queue,
	318	.getevents = fio_libnfs_getevents,
	319	.event = fio_libnfs_event,
	320	.cleanup = fio_libnfs_cleanup,
	321	.open_file = fio_libnfs_open,
	322	.close_file = fio_libnfs_close,
db64b7b6	323	.flags = FIO_DISKLESSIO \| FIO_NOEXTEND \| FIO_NODISKUTIL,
9326926b TG	324	.options = options,
	325	.option_struct_size = sizeof(struct fio_libnfs_options),
	326	};
	327
	328	static void fio_init fio_nfs_register(void)
	329	{
	330	register_ioengine(&ioengine);
	331	}
	332
	333	static void fio_exit fio_nfs_unregister(void)
	334	{
	335	unregister_ioengine(&ioengine);
	336	}