Support vitastor_c_create_epoll() in fio driver

src/cluster_client.cpp

@@ -193,6 +193,10 @@ void cluster_client_t::inc_wait(uint64_t opcode, uint64_t flags, cluster_op_t *n
                     bh_op_max *= 2;
                     cluster_op_t **n = (cluster_op_t**)malloc_or_die(sizeof(cluster_op_t*) * bh_op_max);
                     memcpy(n, bh_ops, sizeof(cluster_op_t*) * bh_op_count);
+                    if (bh_ops != bh_ops_local)
+                    {
+                        free(bh_ops);
+                    }
                     bh_ops = n;
                 }
                 bh_ops[bh_op_count++] = next;

src/epoll_manager.cpp

@@ -23,19 +23,24 @@ epoll_manager_t::epoll_manager_t(ring_loop_t *ringloop)
 
     tfd = new timerfd_manager_t([this](int fd, bool wr, std::function<void(int, int)> handler) { set_fd_handler(fd, wr, handler); });
 
-    consumer.loop = [this]()
+    if (ringloop)
     {
-        if (pending)
-            handle_epoll_events();
-    };
-    ringloop->register_consumer(&consumer);
-
-    handle_epoll_events();
+        consumer.loop = [this]()
+        {
+            if (pending)
+                handle_uring_event();
+        };
+        ringloop->register_consumer(&consumer);
+        handle_uring_event();
+    }
 }
 
 epoll_manager_t::~epoll_manager_t()
 {
-    ringloop->unregister_consumer(&consumer);
+    if (ringloop)
+    {
+        ringloop->unregister_consumer(&consumer);
+    }
     if (tfd)
     {
         delete tfd;
@@ -44,6 +49,11 @@ epoll_manager_t::~epoll_manager_t()
     close(epoll_fd);
 }
 
+int epoll_manager_t::get_fd()
+{
+    return epoll_fd;
+}
+
 void epoll_manager_t::set_fd_handler(int fd, bool wr, std::function<void(int, int)> handler)
 {
     if (handler != NULL)
@@ -75,7 +85,7 @@ void epoll_manager_t::set_fd_handler(int fd, bool wr, std::function<void(int, in
     }
 }
 
-void epoll_manager_t::handle_epoll_events()
+void epoll_manager_t::handle_uring_event()
 {
     io_uring_sqe *sqe = ringloop->get_sqe();
     if (!sqe)
@@ -95,14 +105,20 @@ void epoll_manager_t::handle_epoll_events()
         {
             throw std::runtime_error(std::string("epoll failed: ") + strerror(-data->res));
         }
-        handle_epoll_events();
+        handle_uring_event();
     };
     ringloop->submit();
+    handle_events(0);
+}
+
+void epoll_manager_t::handle_events(int timeout)
+{
     int nfds;
     epoll_event events[MAX_EPOLL_EVENTS];
     do
     {
-        nfds = epoll_wait(epoll_fd, events, MAX_EPOLL_EVENTS, 0);
+        nfds = epoll_wait(epoll_fd, events, MAX_EPOLL_EVENTS, timeout);
+        timeout = 0;
         for (int i = 0; i < nfds; i++)
         {
             auto cb_it = epoll_handlers.find(events[i].data.fd);

src/epoll_manager.h

@@ -15,11 +15,14 @@ class epoll_manager_t
     ring_consumer_t consumer;
     ring_loop_t *ringloop;
     std::map<int, std::function<void(int, int)>> epoll_handlers;
+
+    void handle_uring_event();
 public:
     epoll_manager_t(ring_loop_t *ringloop);
     ~epoll_manager_t();
+    int get_fd();
     void set_fd_handler(int fd, bool wr, std::function<void(int, int)> handler);
-    void handle_epoll_events();
+    void handle_events(int timeout);
 
     timerfd_manager_t *tfd;
 };

src/fio_cluster.cpp

@@ -32,6 +32,7 @@
 struct sec_data
 {
     vitastor_c *cli = NULL;
+    bool epoll_based = false;
     void *watch = NULL;
     bool last_sync = false;
     /* The list of completed io_u structs. */
@@ -58,6 +59,7 @@ struct sec_options
     int rdma_port_num = 0;
     int rdma_gid_index = 0;
     int rdma_mtu = 0;
+    int no_io_uring = 0;
 };
 
 static struct fio_option options[] = {
@@ -193,6 +195,16 @@ static struct fio_option options[] = {
         .category = FIO_OPT_C_ENGINE,
         .group  = FIO_OPT_G_FILENAME,
     },
+    {
+        .name   = "no_io_uring",
+        .lname  = "Disable io_uring",
+        .type   = FIO_OPT_BOOL,
+        .off1   = offsetof(struct sec_options, no_io_uring),
+        .help   = "Use epoll and plain sendmsg/recvmsg instead of io_uring (slower)",
+        .def    = "0",
+        .category = FIO_OPT_C_ENGINE,
+        .group  = FIO_OPT_G_FILENAME,
+    },
     {
         .name = NULL,
     },
@@ -281,7 +293,17 @@ static int sec_setup(struct thread_data *td)
         opt_push(options, "log_level", std::to_string(o->cluster_log).c_str());
     // allow writeback caching if -direct is not set
     opt_push(options, "client_writeback_allowed", td->o.odirect ? "0" : "1");
-    bsd->cli = vitastor_c_create_uring_json((const char**)options.data(), options.size());
+    bsd->cli = o->no_io_uring ? NULL : vitastor_c_create_uring_json((const char**)options.data(), options.size());
+    bsd->epoll_based = false;
+    if (!bsd->cli)
+    {
+        if (o->no_io_uring)
+            fprintf(stderr, "vitastor: io_uring disabled - I/O will be slower\n");
+        else
+            fprintf(stderr, "vitastor: failed to create io_uring: %s - I/O will be slower\n", strerror(errno));
+        bsd->cli = vitastor_c_create_epoll_json((const char**)options.data(), options.size());
+        bsd->epoll_based = true;
+    }
     for (auto opt: options)
         free(opt);
     options.clear();
@@ -289,12 +311,24 @@ static int sec_setup(struct thread_data *td)
     {
         bsd->watch = NULL;
         vitastor_c_watch_inode(bsd->cli, o->image, watch_callback, bsd);
-        while (true)
+        if (!bsd->epoll_based)
         {
-            vitastor_c_uring_handle_events(bsd->cli);
-            if (bsd->watch)
-                break;
-            vitastor_c_uring_wait_events(bsd->cli);
+            while (true)
+            {
+                vitastor_c_uring_handle_events(bsd->cli);
+                if (bsd->watch)
+                    break;
+                vitastor_c_uring_wait_events(bsd->cli);
+            }
+        }
+        else
+        {
+            while (true)
+            {
+                if (bsd->watch)
+                    break;
+                vitastor_c_epoll_handle_events(bsd->cli, 1000);
+            }
         }
         td->files[0]->real_file_size = vitastor_c_inode_get_size(bsd->watch);
         if (!vitastor_c_inode_get_num(bsd->watch) ||
@@ -437,12 +471,24 @@ static enum fio_q_status sec_queue(struct thread_data *td, struct io_u *io)
 static int sec_getevents(struct thread_data *td, unsigned int min, unsigned int max, const struct timespec *t)
 {
     sec_data *bsd = (sec_data*)td->io_ops_data;
-    while (true)
+    if (!bsd->epoll_based)
     {
-        vitastor_c_uring_handle_events(bsd->cli);
-        if (bsd->completed.size() >= min)
-            break;
-        vitastor_c_uring_wait_events(bsd->cli);
+        while (true)
+        {
+            vitastor_c_uring_handle_events(bsd->cli);
+            if (bsd->completed.size() >= min)
+                break;
+            vitastor_c_uring_wait_events(bsd->cli);
+        }
+    }
+    else
+    {
+        while (true)
+        {
+            if (bsd->completed.size() >= min)
+                break;
+            vitastor_c_epoll_handle_events(bsd->cli, 1000);
+        }
     }
     return bsd->completed.size();
 }

src/vitastor_c.cpp

@@ -186,12 +186,26 @@ vitastor_c *vitastor_c_create_uring_json(const char **options, int options_len)
     return self;
 }
 
+vitastor_c *vitastor_c_create_epoll_json(const char **options, int options_len)
+{
+    json11::Json::object cfg;
+    for (int i = 0; i < options_len-1; i += 2)
+    {
+        cfg[options[i]] = std::string(options[i+1]);
+    }
+    json11::Json cfg_json(cfg);
+    vitastor_c *self = new vitastor_c;
+    self->epmgr = new epoll_manager_t(NULL);
+    self->cli = new cluster_client_t(NULL, self->epmgr->tfd, cfg_json);
+    return self;
+}
+
 void vitastor_c_destroy(vitastor_c *client)
 {
     delete client->cli;
     if (client->epmgr)
         delete client->epmgr;
-    else
+    else if (client->tfd)
         delete client->tfd;
     if (client->ringloop)
         delete client->ringloop;
@@ -229,6 +243,16 @@ int vitastor_c_uring_has_work(vitastor_c *client)
     return client->ringloop->has_work();
 }
 
+int vitastor_c_epoll_get_fd(vitastor_c *client)
+{
+    return !client->ringloop && client->epmgr ? client->epmgr->get_fd() : -1;
+}
+
+void vitastor_c_epoll_handle_events(vitastor_c *client, int timeout)
+{
+    return client->epmgr->handle_events(timeout);
+}
+
 void vitastor_c_read(vitastor_c *client, uint64_t inode, uint64_t offset, uint64_t len,
     struct iovec *iov, int iovcnt, VitastorReadHandler cb, void *opaque)
 {

src/vitastor_c.h

@@ -7,7 +7,7 @@
 #define VITASTOR_QEMU_PROXY_H
 
 // C API wrapper version
-#define VITASTOR_C_API_VERSION 2
+#define VITASTOR_C_API_VERSION 3
 
 #ifndef POOL_ID_BITS
 #define POOL_ID_BITS 16
@@ -40,6 +40,7 @@ vitastor_c *vitastor_c_create_qemu_uring(QEMUSetFDHandler *aio_set_fd_handler, v
 vitastor_c *vitastor_c_create_uring(const char *config_path, const char *etcd_host, const char *etcd_prefix,
     int use_rdma, const char *rdma_device, int rdma_port_num, int rdma_gid_index, int rdma_mtu, int log_level);
 vitastor_c *vitastor_c_create_uring_json(const char **options, int options_len);
+vitastor_c *vitastor_c_create_epoll_json(const char **options, int options_len);
 void vitastor_c_destroy(vitastor_c *client);
 int vitastor_c_is_ready(vitastor_c *client);
 int vitastor_c_uring_register_eventfd(vitastor_c *client);
@@ -47,6 +48,8 @@ void vitastor_c_uring_wait_ready(vitastor_c *client);
 void vitastor_c_uring_handle_events(vitastor_c *client);
 void vitastor_c_uring_wait_events(vitastor_c *client);
 int vitastor_c_uring_has_work(vitastor_c *client);
+int vitastor_c_epoll_get_fd(vitastor_c *client);
+void vitastor_c_epoll_handle_events(vitastor_c *client, int timeout);
 void vitastor_c_read(vitastor_c *client, uint64_t inode, uint64_t offset, uint64_t len,
     struct iovec *iov, int iovcnt, VitastorReadHandler cb, void *opaque);
 void vitastor_c_write(vitastor_c *client, uint64_t inode, uint64_t offset, uint64_t len, uint64_t check_version,