From 02aee639c5b3bf7becfd45e742cb59698a1ba45d Mon Sep 17 00:00:00 2001 From: qicosmos Date: Tue, 26 Aug 2025 16:11:28 +0800 Subject: [PATCH] use io context pool --- include/rest_rpc/io_context_pool.hpp | 4 ++- include/rest_rpc/rpc_server.h | 42 ++++++++++++---------------- tests/test_io_context_pool.cpp | 10 +++---- tests/test_rest_rpc.cpp | 36 ++++++++++++------------ 4 files changed, 44 insertions(+), 48 deletions(-) diff --git a/include/rest_rpc/io_context_pool.hpp b/include/rest_rpc/io_context_pool.hpp index 68df390..31d988e 100644 --- a/include/rest_rpc/io_context_pool.hpp +++ b/include/rest_rpc/io_context_pool.hpp @@ -38,11 +38,13 @@ public: size_t size() const { return io_contexts_.size(); } - std::shared_ptr &get_io_context() { + std::shared_ptr &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> io_contexts_; std::vector works_; diff --git a/include/rest_rpc/rpc_server.h b/include/rest_rpc/rpc_server.h index ee60da9..aa135df 100644 --- a/include/rest_rpc/rpc_server.h +++ b/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 #include @@ -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([this] { clean(); }); pub_sub_thread_ = std::make_shared([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([this] { io_service_pool_.run(); }); + thd_ = std::make_shared([this] { io_context_pool_.run(); }); } - void run() { io_service_pool_.run(); } + void run() { io_context_pool_.run(); } template 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 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 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(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 conn_; std::shared_ptr thd_; @@ -268,8 +264,6 @@ private: bool stop_check_ = false; std::condition_variable cv_; - asio::signal_set signals_; - std::function err_cb_; std::function conn_timeout_callback_; std::function, std::string)> diff --git a/tests/test_io_context_pool.cpp b/tests/test_io_context_pool.cpp index f1c0795..c5d1260 100644 --- a/tests/test_io_context_pool.cpp +++ b/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); diff --git a/tests/test_rest_rpc.cpp b/tests/test_rest_rpc.cpp index ee0f37d..2353748 100644 --- a/tests/test_rest_rpc.cpp +++ b/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( "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);