[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;
	unsigned int queue_depth; /* nfs_callback needs this info, but doesn't have fio td structure to pull it from */
	/* the following implement a circular queue of outstanding IOs */
	int outstanding_events; /* IOs issued to libnfs, that have not returned yet */
	int prev_requested_event_index; /* event last returned via fio_libnfs_event */
	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;
}

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

	do {
		int timeout = SHOULD_WAIT() ? -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());
	return o->buffered_event_count;
#undef SHOULD_WAIT
}

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 wanna 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;
	int err;
	enum fio_q_status ret = FIO_Q_QUEUED;

	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 == NULL) {
		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)
{
	int ret;
	struct fio_libnfs_options *options = td->eo;
	struct nfs_data *nfs_data = NULL;
	int flags = 0;

	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 != 0) {
		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 != 0)
		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;
}

/*
 * Hook for writing out outstanding data.
 */
static int fio_libnfs_commit(struct thread_data *td) {
	nfs_event_loop(td, true);
	return 0;
}

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