--- /dev/null
+// https://github.com/axboe/fio/pull/762 sample pull req for new engine
+#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;
+ // the following implements 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 faiting for FIO
+ int free_event_buffer_index; // next empty buffer
+ unsigned int queue_depth; // nfs_callback needs this info, but doesn't have fio td structure to pull it from
+ 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,
+ },
+};
+
+/*
+ * The ->event() hook is called to match an event number with an io_u.
+ * After the core has called ->getevents() and it has returned eg 3,
+ * the ->event() hook must return the 3 events that have completed for
+ * subsequent calls to ->event() with [0-2]. Required.
+ */
+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
+
+/*
+ * The ->getevents() hook is used to reap completion events from an async
+ * io engine. It returns the number of completed events since the last call,
+ * which may then be retrieved by calling the ->event() hook with the event
+ * numbers. Required.
+ */
+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);
+}
+
+/*
+ * The ->queue() hook is responsible for initiating io on the io_u
+ * being passed in. If the io engine is a synchronous one, io may complete
+ * before ->queue() returns. Required.
+ *
+ * The io engine must transfer in the direction noted by io_u->ddir
+ * to the buffer pointed to by io_u->xfer_buf for as many bytes as
+ * io_u->xfer_buflen. Residual data count may be set in io_u->resid
+ * for a short read/write.
+ */
+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;
+}
+
+/*
+ * The init function is called once per thread/process, and should set up
+ * any structures that this io engine requires to keep track of io. Not
+ * required.
+ */
+static int fio_libnfs_setup(struct thread_data *td)
+{
+ // flipping this makes using gdb easier, but tends to hang fio on exit
+ td->o.use_thread = 0;
+ return 0;
+}
+
+/*
+ * This is paired with the ->init() function and is called when a thread is
+ * done doing io. Should tear down anything setup by the ->init() function.
+ * Not required.
+ */
+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);
+}
+
projects / fio.git / commitdiff