use io context pool

include/rest_rpc/io_context_pool.hpp

@@ -38,11 +38,13 @@ public:
 
   size_t size() const { return io_contexts_.size(); }
 
-  std::shared_ptr<asio::io_context> &get_io_context() {
+  std::shared_ptr<asio::io_context> &get_io_context_ptr() {
     size_t i = next_.fetch_add(1, std::memory_order::relaxed);
     return io_contexts_[i % io_contexts_.size()];
   }
 
+  asio::io_context &get_io_context() { return *get_io_context_ptr(); }
+
 private:
   std::vector<std::shared_ptr<asio::io_context>> io_contexts_;
   std::vector<asio::io_context::work> works_;

include/rest_rpc/rpc_server.h

@@ -2,7 +2,7 @@
 #define REST_RPC_RPC_SERVER_H_
 
 #include "connection.h"
-#include "io_service_pool.h"
+#include "io_context_pool.hpp"
 #include "router.h"
 #include <condition_variable>
 #include <mutex>
@@ -17,20 +17,13 @@ class rpc_server : private asio::noncopyable {
 public:
   rpc_server(unsigned short port, size_t size, size_t timeout_seconds = 15,
              size_t check_seconds = 10)
-      : io_service_pool_(size), acceptor_(io_service_pool_.get_io_service(),
+      : io_context_pool_(size), acceptor_(io_context_pool_.get_io_context(),
                                           tcp::endpoint(tcp::v4(), port)),
-        timeout_seconds_(timeout_seconds), check_seconds_(check_seconds),
-        signals_(io_service_pool_.get_io_service()) {
+        timeout_seconds_(timeout_seconds), check_seconds_(check_seconds) {
     do_accept();
     check_thread_ = std::make_shared<std::thread>([this] { clean(); });
     pub_sub_thread_ =
         std::make_shared<std::thread>([this] { clean_sub_pub(); });
-    signals_.add(SIGINT);
-    signals_.add(SIGTERM);
-#if defined(SIGQUIT)
-    signals_.add(SIGQUIT);
-#endif // defined(SIGQUIT)
-    do_await_stop();
   }
 
   rpc_server(unsigned short port, size_t size, ssl_configure ssl_conf,
@@ -47,10 +40,10 @@ public:
   ~rpc_server() { stop(); }
 
   void async_run() {
-    thd_ = std::make_shared<std::thread>([this] { io_service_pool_.run(); });
+    thd_ = std::make_shared<std::thread>([this] { io_context_pool_.run(); });
   }
 
-  void run() { io_service_pool_.run(); }
+  void run() { io_context_pool_.run(); }
 
   template <bool is_pub = false, typename Function>
   void register_handler(std::string const &name, const Function &f) {
@@ -94,7 +87,7 @@ public:
 
 private:
   void do_accept() {
-    conn_.reset(new connection(io_service_pool_.get_io_service(),
+    conn_.reset(new connection(io_context_pool_.get_io_context(),
                                timeout_seconds_, router_));
     conn_->set_callback([this](std::string key, std::string token,
                                std::weak_ptr<connection> conn) {
@@ -110,9 +103,11 @@ private:
         return;
       }
 
-      if (ec) {
+      if (ec == asio::error::operation_aborted ||
+          ec == asio::error::bad_descriptor) {
         // LOG(INFO) << "acceptor error: " <<
         // ec.message();
+        return;
       } else {
 #ifdef CINATRA_ENABLE_SSL
         if (!ssl_conf_.cert_file.empty()) {
@@ -153,7 +148,7 @@ private:
   void clean_sub_pub() {
     while (!stop_check_pub_sub_) {
       std::unique_lock<std::mutex> lock(sub_mtx_);
-      sub_cv_.wait_for(lock, std::chrono::seconds(10));
+      sub_cv_.wait_for(lock, std::chrono::seconds(check_seconds_));
 
       for (auto it = sub_map_.cbegin(); it != sub_map_.cend();) {
         auto conn = it->second.lock();
@@ -223,11 +218,6 @@ private:
     return std::make_shared<std::string>(buf.data(), buf.size());
   }
 
-  void do_await_stop() {
-    signals_.async_wait(
-        [this](std::error_code /*ec*/, int /*signo*/) { stop(); });
-  }
-
   void stop() {
     if (has_stoped_) {
       return;
@@ -247,14 +237,20 @@ private:
     }
     pub_sub_thread_->join();
 
-    io_service_pool_.stop();
+    asio::dispatch(acceptor_.get_executor(), [this]() {
+      asio::error_code ec;
+      (void)acceptor_.cancel(ec);
+      (void)acceptor_.close(ec);
+    });
+
+    io_context_pool_.stop();
     if (thd_) {
       thd_->join();
     }
     has_stoped_ = true;
   }
 
-  io_service_pool io_service_pool_;
+  io_context_pool io_context_pool_;
   tcp::acceptor acceptor_;
   std::shared_ptr<connection> conn_;
   std::shared_ptr<std::thread> thd_;
@@ -268,8 +264,6 @@ private:
   bool stop_check_ = false;
   std::condition_variable cv_;
 
-  asio::signal_set signals_;
-
   std::function<void(asio::error_code, string_view)> err_cb_;
   std::function<void(int64_t)> conn_timeout_callback_;
   std::function<void(std::shared_ptr<connection>, std::string)>

tests/test_io_context_pool.cpp

@@ -55,7 +55,7 @@ TEST_CASE("test context pool automatic stop") {
     p.set_value();
     pool->run();
     block = false;
-    CHECK(block);
+    CHECK(!block);
   });
   p.get_future().wait();
   std::this_thread::sleep_for(std::chrono::milliseconds(200));
@@ -66,10 +66,10 @@ TEST_CASE("test context pool automatic stop") {
 TEST_CASE("test get io context") {
   io_context_pool pool(2);
   CHECK(pool.size() == 2);
-  auto &ctx1 = pool.get_io_context();
-  auto &ctx2 = pool.get_io_context();
-  auto &ctx3 = pool.get_io_context();
-  auto ctx4 = pool.get_io_context();
+  auto &ctx1 = pool.get_io_context_ptr();
+  auto &ctx2 = pool.get_io_context_ptr();
+  auto &ctx3 = pool.get_io_context_ptr();
+  auto ctx4 = pool.get_io_context_ptr();
   CHECK(ctx1 != ctx2);
   CHECK(ctx1 == ctx3);
   CHECK(ctx2 == ctx4);

tests/test_rest_rpc.cpp

@@ -72,7 +72,7 @@ TEST_CASE("test_client_reconnect") {
   });
   client.connect("127.0.0.1", 9000);
 
-  rpc_server server(9000, std::thread::hardware_concurrency());
+  rpc_server server(9000, std::thread::hardware_concurrency(), 2, 2);
   dummy d;
   server.register_handler("add", &dummy::add, &d);
   server.async_run();
@@ -95,7 +95,7 @@ TEST_CASE("test_client_reconnect") {
 }
 
 TEST_CASE("test_client_default_constructor") {
-  rpc_server server(9000, std::thread::hardware_concurrency());
+  rpc_server server(9000, std::thread::hardware_concurrency(), 1, 1);
   dummy d;
   server.register_handler("add", &dummy::add, &d);
   server.async_run();
@@ -110,7 +110,7 @@ TEST_CASE("test_client_default_constructor") {
 }
 
 TEST_CASE("test_constructor_with_language") {
-  rpc_server server(9000, std::thread::hardware_concurrency());
+  rpc_server server(9000, std::thread::hardware_concurrency(), 1, 1);
   dummy d;
   server.register_handler("add", &dummy::add, &d);
   server.async_run();
@@ -124,7 +124,7 @@ TEST_CASE("test_constructor_with_language") {
 }
 
 TEST_CASE("test_client_async_connect") {
-  rpc_server server(9000, std::thread::hardware_concurrency());
+  rpc_server server(9000, std::thread::hardware_concurrency(), 1, 1);
   dummy d;
   server.register_handler("add", &dummy::add, &d);
   server.async_run();
@@ -142,7 +142,7 @@ TEST_CASE("test_client_async_connect") {
 }
 
 TEST_CASE("test_client_sync_call") {
-  rpc_server server(9000, std::thread::hardware_concurrency());
+  rpc_server server(9000, std::thread::hardware_concurrency(), 1, 1);
   dummy d;
   server.register_handler("add", &dummy::add, &d);
   server.async_run();
@@ -160,7 +160,7 @@ TEST_CASE("test_client_sync_call") {
 }
 
 TEST_CASE("test_client_sync_call_return_void") {
-  rpc_server server(9000, std::thread::hardware_concurrency());
+  rpc_server server(9000, std::thread::hardware_concurrency(), 1, 1);
   server.register_handler("echo", echo);
   server.async_run();
   std::this_thread::sleep_for(std::chrono::milliseconds(200));
@@ -172,7 +172,7 @@ TEST_CASE("test_client_sync_call_return_void") {
 }
 
 TEST_CASE("test_client_async_call_empty_obj") {
-  rpc_server server(9000, std::thread::hardware_concurrency());
+  rpc_server server(9000, std::thread::hardware_concurrency(), 1, 1);
   server.register_handler("get_empty_obj", get_empty_obj);
   server.async_run();
   std::this_thread::sleep_for(std::chrono::milliseconds(200));
@@ -194,7 +194,7 @@ TEST_CASE("test_client_async_call_empty_obj") {
 }
 
 TEST_CASE("test_client_async_call") {
-  rpc_server server(9000, std::thread::hardware_concurrency());
+  rpc_server server(9000, std::thread::hardware_concurrency(), 1, 1);
   server.register_handler("get_person", get_person);
   server.register_handler("hello", hello);
   server.async_run();
@@ -233,7 +233,7 @@ TEST_CASE("test_client_async_call_not_connect") {
 }
 
 TEST_CASE("test_client_async_call_with_timeout") {
-  rpc_server server(9000, std::thread::hardware_concurrency());
+  rpc_server server(9000, std::thread::hardware_concurrency(), 1, 1);
   server.register_handler("echo", echo);
   server.register_handler("get_person", get_person);
   server.async_run();
@@ -285,7 +285,7 @@ TEST_CASE("test_client_async_call_with_timeout") {
 }
 
 TEST_CASE("test_client_subscribe") {
-  rpc_server server(9000, std::thread::hardware_concurrency());
+  rpc_server server(9000, std::thread::hardware_concurrency(), 1, 1);
   server.register_handler<true>(
       "publish", [&server](rpc_conn conn, std::string key, std::string token,
                            std::string val) {
@@ -315,7 +315,7 @@ TEST_CASE("test_client_subscribe") {
 }
 
 TEST_CASE("test_client_subscribe_not_exist_key") {
-  rpc_server server(9000, std::thread::hardware_concurrency());
+  rpc_server server(9000, std::thread::hardware_concurrency(), 1, 1);
   server.register_handler("publish",
                           [&server](rpc_conn conn, std::string key,
                                     std::string token, std::string val) {
@@ -354,7 +354,7 @@ TEST_CASE("test_client_subscribe_not_exist_key") {
 }
 
 TEST_CASE("test_server_publish_encode_msg") {
-  rpc_server server(9000, std::thread::hardware_concurrency());
+  rpc_server server(9000, std::thread::hardware_concurrency(), 1, 1);
   server.register_handler("publish",
                           [&server](rpc_conn conn, std::string key,
                                     std::string token, std::string val) {
@@ -389,7 +389,7 @@ TEST_CASE("test_server_publish_encode_msg") {
 }
 
 TEST_CASE("test_client_subscribe_by_token") {
-  rpc_server server(9000, std::thread::hardware_concurrency());
+  rpc_server server(9000, std::thread::hardware_concurrency(), 1, 1);
   bool stop = false;
   std::thread thd([&server, &stop] {
     while (!stop) {
@@ -416,7 +416,7 @@ TEST_CASE("test_client_subscribe_by_token") {
 }
 
 TEST_CASE("test_client_publish_and_subscribe_by_token") {
-  rpc_server server(9000, std::thread::hardware_concurrency());
+  rpc_server server(9000, std::thread::hardware_concurrency(), 1, 1);
   server.register_handler("publish_by_token", [&server](rpc_conn conn,
                                                         std::string key,
                                                         std::string token,
@@ -450,7 +450,7 @@ TEST_CASE("test_client_publish_and_subscribe_by_token") {
 }
 
 TEST_CASE("test_server_callback") {
-  rpc_server server(9000, std::thread::hardware_concurrency());
+  rpc_server server(9000, std::thread::hardware_concurrency(), 1, 1);
   dummy d;
   server.register_handler("add", &dummy::add, &d);
   server.set_network_err_callback(
@@ -471,7 +471,7 @@ TEST_CASE("test_server_callback") {
   CHECK_EQ(result, 3);
 }
 TEST_CASE("test_server_user_data") {
-  rpc_server server(9000, std::thread::hardware_concurrency());
+  rpc_server server(9000, std::thread::hardware_concurrency(), 1, 1);
   server.register_handler("server_user_data", server_user_data);
   server.async_run();
   std::this_thread::sleep_for(std::chrono::milliseconds(200));
@@ -482,7 +482,7 @@ TEST_CASE("test_server_user_data") {
   client.call<>("server_user_data");
 }
 TEST_CASE("test_server_delay_response") {
-  rpc_server server(9000, std::thread::hardware_concurrency());
+  rpc_server server(9000, std::thread::hardware_concurrency(), 2, 2);
   server.register_handler("delay_echo", delay_echo);
   server.async_run();
   std::this_thread::sleep_for(std::chrono::milliseconds(200));
@@ -494,7 +494,7 @@ TEST_CASE("test_server_delay_response") {
   CHECK_EQ(result, "test_delay_echo");
 }
 TEST_CASE("test_server_duplicate_registration_key") {
-  rpc_server server(9000, std::thread::hardware_concurrency());
+  rpc_server server(9000, std::thread::hardware_concurrency(), 2, 2);
   server.register_handler("delay_echo", delay_echo);
   try {
     server.register_handler("delay_echo", delay_echo);