diff --git a/src/client/CMakeLists.txt b/src/client/CMakeLists.txt index d08d4bc0..a7677830 100644 --- a/src/client/CMakeLists.txt +++ b/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} ) diff --git a/src/client/messenger.cpp b/src/client/messenger.cpp index b4a0d84e..4f0df5c2 100644 --- a/src/client/messenger.cpp +++ b/src/client/messenger.cpp @@ -15,106 +15,6 @@ #include "msgr_rdma.h" #endif -#include - -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 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) { diff --git a/src/client/messenger.h b/src/client/messenger.h index 88b34894..c7c30bea 100644 --- a/src/client/messenger.h +++ b/src/client/messenger.h @@ -129,43 +129,7 @@ struct osd_op_stats_t uint64_t subop_stat_count[OSD_OP_MAX+1] = { 0 }; }; -#include -#include -#include - -#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 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); diff --git a/src/client/msgr_iothread.cpp b/src/client/msgr_iothread.cpp new file mode 100644 index 00000000..ace420a3 --- /dev/null +++ b/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 +#include + +#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 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(); + } +} diff --git a/src/client/msgr_iothread.h b/src/client/msgr_iothread.h new file mode 100644 index 00000000..e7ff2efc --- /dev/null +++ b/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 +#include +#include + +#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 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); +}; diff --git a/src/client/msgr_receive.cpp b/src/client/msgr_receive.cpp index c7ca2160..19e26896 100644 --- a/src/client/msgr_receive.cpp +++ b/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() { diff --git a/src/client/msgr_send.cpp b/src/client/msgr_send.cpp index 8e18c47f..c8f95b7f 100644 --- a/src/client/msgr_send.cpp +++ b/src/client/msgr_send.cpp @@ -6,6 +6,7 @@ #include #include "messenger.h" +#include "msgr_iothread.h" void osd_messenger_t::outbox_push(osd_op_t *cur_op) {