Extract msgr_iothread to separate .{cpp,h}

src/client/CMakeLists.txt

@@ -12,7 +12,7 @@ if (RDMACM_LIBRARIES)
 	set(MSGR_RDMACM "msgr_rdmacm.cpp")
 endif (RDMACM_LIBRARIES)
 add_library(vitastor_common STATIC
-	../util/epoll_manager.cpp etcd_state_client.cpp messenger.cpp ../util/addr_util.cpp
+	../util/epoll_manager.cpp etcd_state_client.cpp messenger.cpp msgr_iothread.cpp ../util/addr_util.cpp
 	msgr_stop.cpp msgr_op.cpp msgr_send.cpp msgr_receive.cpp ../util/ringloop.cpp ../../json11/json11.cpp
 	http_client.cpp osd_ops.cpp pg_states.cpp ../util/timerfd_manager.cpp ../util/str_util.cpp ../util/json_util.cpp ${MSGR_RDMA} ${MSGR_RDMACM}
 )

src/client/messenger.cpp

@@ -15,106 +15,6 @@
 #include "msgr_rdma.h"
 #endif
 
-#include <sys/poll.h>
-
-msgr_iothread_t::msgr_iothread_t():
-    ring(RINGLOOP_DEFAULT_SIZE, true),
-    thread(&msgr_iothread_t::run, this)
-{
-    eventfd = ring.register_eventfd();
-    if (eventfd < 0)
-    {
-        throw std::runtime_error(std::string("failed to register eventfd: ") + strerror(-eventfd));
-    }
-}
-
-msgr_iothread_t::~msgr_iothread_t()
-{
-    stop();
-}
-
-void msgr_iothread_t::add_sqe(io_uring_sqe & sqe)
-{
-    mu.lock();
-    queue.push_back((iothread_sqe_t){ .sqe = sqe, .data = std::move(*(ring_data_t*)sqe.user_data) });
-    if (queue.size() == 1)
-    {
-        cond.notify_all();
-    }
-    mu.unlock();
-}
-
-void msgr_iothread_t::stop()
-{
-    mu.lock();
-    if (stopped)
-    {
-        mu.unlock();
-        return;
-    }
-    stopped = true;
-    if (outer_loop_data)
-    {
-        outer_loop_data->callback = [](ring_data_t*){};
-    }
-    cond.notify_all();
-    close(eventfd);
-    mu.unlock();
-    thread.join();
-}
-
-void msgr_iothread_t::add_to_ringloop(ring_loop_t *outer_loop)
-{
-    assert(!this->outer_loop || this->outer_loop == outer_loop);
-    io_uring_sqe *sqe = outer_loop->get_sqe();
-    assert(sqe != NULL);
-    this->outer_loop = outer_loop;
-    this->outer_loop_data = ((ring_data_t*)sqe->user_data);
-    io_uring_prep_poll_add(sqe, eventfd, POLLIN);
-    outer_loop_data->callback = [this](ring_data_t *data)
-    {
-        if (data->res < 0)
-        {
-            throw std::runtime_error(std::string("eventfd poll failed: ") + strerror(-data->res));
-        }
-        outer_loop_data = NULL;
-        if (stopped)
-        {
-            return;
-        }
-        add_to_ringloop(this->outer_loop);
-        ring.loop();
-    };
-}
-
-void msgr_iothread_t::run()
-{
-    while (true)
-    {
-        {
-            std::unique_lock<std::mutex> lk(mu);
-            while (!stopped && !queue.size())
-                cond.wait(lk);
-            if (stopped)
-                return;
-            int i = 0;
-            for (; i < queue.size(); i++)
-            {
-                io_uring_sqe *sqe = ring.get_sqe();
-                if (!sqe)
-                    break;
-                ring_data_t *data = ((ring_data_t*)sqe->user_data);
-                *data = std::move(queue[i].data);
-                *sqe = queue[i].sqe;
-                sqe->user_data = (uint64_t)data;
-            }
-            queue.erase(queue.begin(), queue.begin()+i);
-        }
-        // We only want to offload sendmsg/recvmsg. Callbacks will be called in main thread
-        ring.submit();
-    }
-}
-
 void osd_messenger_t::init()
 {
 #ifdef WITH_RDMACM
@@ -173,12 +73,7 @@ void osd_messenger_t::init()
     }
     if (ringloop && iothread_count > 0)
     {
-        for (int i = 0; i < iothread_count; i++)
-        {
-            auto iot = new msgr_iothread_t();
-            iothreads.push_back(iot);
-            iot->add_to_ringloop(ringloop);
-        }
+        init_iothreads();
     }
     keepalive_timer_id = tfd->set_timer(1000, true, [this](int)
     {
@@ -272,14 +167,7 @@ osd_messenger_t::~osd_messenger_t()
     {
         stop_client(clients.begin()->first, true);
     }
-    if (iothreads.size())
-    {
-        for (auto iot: iothreads)
-        {
-            delete iot;
-        }
-        iothreads.clear();
-    }
+    destroy_iothreads();
 #ifdef WITH_RDMA
     for (auto rdma_context: rdma_contexts)
     {

src/client/messenger.h

@@ -129,43 +129,7 @@ struct osd_op_stats_t
     uint64_t subop_stat_count[OSD_OP_MAX+1] = { 0 };
 };
 
-#include <mutex>
-#include <condition_variable>
-#include <thread>
-
-#ifdef __MOCK__
 class msgr_iothread_t;
-#else
-struct iothread_sqe_t
-{
-    io_uring_sqe sqe;
-    ring_data_t data;
-};
-
-class msgr_iothread_t
-{
-protected:
-    ring_loop_t ring;
-    ring_loop_t *outer_loop = NULL;
-    ring_data_t *outer_loop_data = NULL;
-    int eventfd = -1;
-    bool stopped = false;
-    std::mutex mu;
-    std::condition_variable cond;
-    std::vector<iothread_sqe_t> queue;
-    std::thread thread;
-
-    void run();
-public:
-
-    msgr_iothread_t();
-    ~msgr_iothread_t();
-
-    void add_sqe(io_uring_sqe & sqe);
-    void stop();
-    void add_to_ringloop(ring_loop_t *outer_loop);
-};
-#endif
 
 #ifdef WITH_RDMA
 struct rdma_event_channel;
@@ -234,6 +198,7 @@ public:
     osd_op_stats_t stats, recovery_stats;
 
     void init();
+    void init_iothreads();
     void parse_config(const json11::Json & config);
     void connect_peer(uint64_t osd_num, json11::Json peer_state);
     void stop_client(uint64_t client_id, bool force_delete = false);
@@ -246,6 +211,7 @@ public:
     void read_requests();
     void send_replies();
     void accept_connections(int listen_fd);
+    void destroy_iothreads();
     ~osd_messenger_t();
 
     static json11::Json::object read_config(const json11::Json & config);

src/client/msgr_iothread.cpp

@@ -0,0 +1,128 @@
+// Copyright (c) Vitaliy Filippov, 2019+
+// License: VNPL-1.1 or GNU GPL-2.0+ (see README.md for details)
+
+#include <stdexcept>
+#include <sys/poll.h>
+
+#include "messenger.h"
+#include "msgr_iothread.h"
+
+msgr_iothread_t::msgr_iothread_t():
+    ring(RINGLOOP_DEFAULT_SIZE, true),
+    thread(&msgr_iothread_t::run, this)
+{
+    eventfd = ring.register_eventfd();
+    if (eventfd < 0)
+    {
+        throw std::runtime_error(std::string("failed to register eventfd: ") + strerror(-eventfd));
+    }
+}
+
+msgr_iothread_t::~msgr_iothread_t()
+{
+    stop();
+}
+
+void msgr_iothread_t::add_sqe(io_uring_sqe & sqe)
+{
+    mu.lock();
+    queue.push_back((iothread_sqe_t){ .sqe = sqe, .data = std::move(*(ring_data_t*)sqe.user_data) });
+    if (queue.size() == 1)
+    {
+        cond.notify_all();
+    }
+    mu.unlock();
+}
+
+void msgr_iothread_t::stop()
+{
+    mu.lock();
+    if (stopped)
+    {
+        mu.unlock();
+        return;
+    }
+    stopped = true;
+    if (outer_loop_data)
+    {
+        outer_loop_data->callback = [](ring_data_t*){};
+    }
+    cond.notify_all();
+    close(eventfd);
+    mu.unlock();
+    thread.join();
+}
+
+void msgr_iothread_t::add_to_ringloop(ring_loop_i *outer_loop)
+{
+    assert(!this->outer_loop || this->outer_loop == outer_loop);
+    io_uring_sqe *sqe = outer_loop->get_sqe();
+    assert(sqe != NULL);
+    this->outer_loop = outer_loop;
+    this->outer_loop_data = ((ring_data_t*)sqe->user_data);
+    io_uring_prep_poll_add(sqe, eventfd, POLLIN);
+    outer_loop_data->callback = [this](ring_data_t *data)
+    {
+        if (data->res < 0)
+        {
+            throw std::runtime_error(std::string("eventfd poll failed: ") + strerror(-data->res));
+        }
+        outer_loop_data = NULL;
+        if (stopped)
+        {
+            return;
+        }
+        add_to_ringloop(this->outer_loop);
+        ring.loop();
+    };
+}
+
+void msgr_iothread_t::run()
+{
+    while (true)
+    {
+        {
+            std::unique_lock<std::mutex> lk(mu);
+            while (!stopped && !queue.size())
+                cond.wait(lk);
+            if (stopped)
+                return;
+            int i = 0;
+            for (; i < queue.size(); i++)
+            {
+                io_uring_sqe *sqe = ring.get_sqe();
+                if (!sqe)
+                    break;
+                ring_data_t *data = ((ring_data_t*)sqe->user_data);
+                *data = std::move(queue[i].data);
+                *sqe = queue[i].sqe;
+                sqe->user_data = (uint64_t)data;
+            }
+            queue.erase(queue.begin(), queue.begin()+i);
+        }
+        // We only want to offload sendmsg/recvmsg. Callbacks will be called in main thread
+        ring.submit();
+    }
+}
+
+void osd_messenger_t::init_iothreads()
+{
+    for (int i = 0; i < iothread_count; i++)
+    {
+        auto iot = new msgr_iothread_t();
+        iothreads.push_back(iot);
+        iot->add_to_ringloop(ringloop);
+    }
+}
+
+void osd_messenger_t::destroy_iothreads()
+{
+    if (iothreads.size())
+    {
+        for (auto iot: iothreads)
+        {
+            delete iot;
+        }
+        iothreads.clear();
+    }
+}

src/client/msgr_iothread.h

@@ -0,0 +1,38 @@
+// Copyright (c) Vitaliy Filippov, 2019+
+// License: VNPL-1.1 or GNU GPL-2.0+ (see README.md for details)
+
+#include <mutex>
+#include <condition_variable>
+#include <thread>
+
+#include "ringloop.h"
+
+struct iothread_sqe_t
+{
+    io_uring_sqe sqe;
+    ring_data_t data;
+};
+
+class msgr_iothread_t
+{
+protected:
+    ring_loop_t ring;
+    ring_loop_i *outer_loop = NULL;
+    ring_data_t *outer_loop_data = NULL;
+    int eventfd = -1;
+    bool stopped = false;
+    std::mutex mu;
+    std::condition_variable cond;
+    std::vector<iothread_sqe_t> queue;
+    std::thread thread;
+
+    void run();
+public:
+
+    msgr_iothread_t();
+    ~msgr_iothread_t();
+
+    void add_sqe(io_uring_sqe & sqe);
+    void stop();
+    void add_to_ringloop(ring_loop_i *outer_loop);
+};

src/client/msgr_receive.cpp

@@ -2,6 +2,7 @@
 // License: VNPL-1.1 or GNU GPL-2.0+ (see README.md for details)
 
 #include "messenger.h"
+#include "msgr_iothread.h"
 
 void osd_messenger_t::read_requests()
 {

src/client/msgr_send.cpp

@@ -6,6 +6,7 @@
 #include <sys/epoll.h>
 
 #include "messenger.h"
+#include "msgr_iothread.h"
 
 void osd_messenger_t::outbox_push(osd_op_t *cur_op)
 {