From 368ee9e22e24bbdf3e9e82424e44ba79c5dd2098 Mon Sep 17 00:00:00 2001 From: Seva Date: Tue, 11 Jun 2019 11:31:43 -0400 Subject: [PATCH] dhtproxy: ensure io_context thread pool & isolate settings --- include/opendht/dht_proxy_server.h | 5 ++ src/dht_proxy_server.cpp | 109 +++++++++++++++++++---------- src/http.cpp | 4 +- 3 files changed, 79 insertions(+), 39 deletions(-) diff --git a/include/opendht/dht_proxy_server.h b/include/opendht/dht_proxy_server.h index c40690e6..ea9f49c7 100644 --- a/include/opendht/dht_proxy_server.h +++ b/include/opendht/dht_proxy_server.h @@ -91,6 +91,8 @@ struct RestRouterTraits : public restinio::default_traits_t using connection_state_listener_t = http::ConnectionListener; }; using ServerSettings = restinio::run_on_thread_pool_settings_t; +using IOContextThreadPool = restinio::impl::ioctx_on_thread_pool_t< + restinio::impl::own_io_context_for_thread_pool_t>; using RequestStatus = restinio::request_handling_status_t; using ResponseByParts = restinio::chunked_output_t; using ResponseByPartsBuilder = restinio::response_builder_t; @@ -187,6 +189,8 @@ private: template HttpResponse initHttpResponse(HttpResponse response) const; + ServerSettings makeHttpServerSettings(const in_port_t port); + std::unique_ptr createRestRouter(); /** @@ -351,6 +355,7 @@ private: std::shared_ptr logger_; Json::StreamWriterBuilder jsonBuilder_; + std::unique_ptr httpServerThreadPool_; std::unique_ptr> httpServer_; std::thread httpServerThread_ {}; diff --git a/src/dht_proxy_server.cpp b/src/dht_proxy_server.cpp index 194bd244..6becd181 100644 --- a/src/dht_proxy_server.cpp +++ b/src/dht_proxy_server.cpp @@ -55,7 +55,7 @@ DhtProxyServer::DhtProxyServer(std::shared_ptr dht, in_port_t port, std::shared_ptr logger): dht_(dht), logger_(logger), threadPool_(new ThreadPool(IO_THREADS_MAX)), connListener_(std::make_shared( - ¤tListeners_, &lockListener_, logger)), + ¤tListeners_, &lockListener_, logger)), pushServer_(pushServer) { if (not dht_) @@ -73,34 +73,33 @@ DhtProxyServer::DhtProxyServer(std::shared_ptr dht, in_port_t port, jsonBuilder_["commentStyle"] = "None"; jsonBuilder_["indentation"] = ""; - httpServerThread_ = std::thread([this, port, logger](){ - using namespace std::chrono; - auto maxThreads = std::thread::hardware_concurrency() - 1; - auto restThreads = maxThreads > 1 ? maxThreads : 1; - auto settings = ServerSettings(restThreads); - settings.logger(logger); - settings.port(port); - settings.protocol(restinio::asio_ns::ip::tcp::v6()); - settings.request_handler(this->createRestRouter()); - // time limits // ~ 0.8 month - std::chrono::milliseconds timeout_request(std::numeric_limits::max()); - settings.read_next_http_message_timelimit(timeout_request); - settings.write_http_response_timelimit(60s); - settings.handle_request_timeout(timeout_request); - // socket options - settings.socket_options_setter([](auto & options){ - options.set_option(asio::ip::tcp::no_delay{true}); - }); - settings.connection_state_listener(connListener_); - httpServer_.reset(new restinio::http_server_t( - restinio::own_io_context(), // requirement: each thread has its own - std::forward(settings) - )); - restinio::asio_ns::post(httpServer_->io_context(), [&]{ - httpServer_->open_sync(); - }); + // build http server + auto settings = makeHttpServerSettings(port); + httpServerThreadPool_.reset(new IOContextThreadPool(settings.pool_size())); + + httpServer_.reset(new restinio::http_server_t( + restinio::external_io_context(httpServerThreadPool_->io_context()), + std::forward(settings) + )); + // run http server + httpServerThread_ = std::thread([this](){ try { - httpServer_->io_context().run(); + asio::signal_set break_signals{httpServer_->io_context(), SIGINT}; + break_signals.async_wait([&](const asio::error_code &ec, int){ + if (!ec){ + httpServer_->close_async([&]{ + httpServerThreadPool_->stop(); + }, + []( std::exception_ptr ex ){ + std::rethrow_exception( ex ); + }); + } + }); + httpServer_->open_async([]{/*Ok.*/}, [](std::exception_ptr ex){ + std::rethrow_exception(ex); + }); + httpServerThreadPool_->start(); + httpServerThreadPool_->wait(); } catch(const std::exception &ex){ std::cerr << "Error starting RESTinio: " << ex.what() << std::endl; @@ -151,6 +150,30 @@ DhtProxyServer::~DhtProxyServer() stop(); } +ServerSettings +DhtProxyServer::makeHttpServerSettings(const in_port_t port) +{ + using namespace std::chrono; + auto maxThreads = std::thread::hardware_concurrency() - 1; + auto restThreads = maxThreads > 1 ? maxThreads : 1; + auto settings = ServerSettings(restThreads); + settings.logger(logger_); + settings.port(port); + settings.protocol(restinio::asio_ns::ip::tcp::v6()); + settings.request_handler(this->createRestRouter()); + // time limits // ~ 0.8 month + std::chrono::milliseconds timeout_request(std::numeric_limits::max()); + settings.read_next_http_message_timelimit(timeout_request); + settings.write_http_response_timelimit(60s); + settings.handle_request_timeout(timeout_request); + // socket options + settings.socket_options_setter([](auto & options){ + options.set_option(asio::ip::tcp::no_delay{true}); + }); + settings.connection_state_listener(connListener_); + return settings; +} + void DhtProxyServer::stop() { @@ -165,8 +188,21 @@ DhtProxyServer::stop() stopListeners = true; schedulerCv_.notify_all(); - // stop its own io_context loop - httpServer_->close_sync(); + logger_->d("[restinio] closing http server async operations"); + httpServer_->close_async([this]{ + logger_->d("[restinio] stopping http server io_context thread pool"); + httpServerThreadPool_->stop(); + }, + [this](std::exception_ptr eptr){ + try { + std::rethrow_exception(eptr); + } + catch (const std::exception &ex){ + logger_->e("[restinio] error closing http server async: %s", ex.what()); + } + }); + if (httpServerThread_.joinable()) + httpServerThread_.join(); // listenThreads_ will stop because there is no more sessions if (listenThread_.joinable()) @@ -174,9 +210,6 @@ DhtProxyServer::stop() if (schedulerThread_.joinable()) schedulerThread_.join(); - if (httpServerThread_.joinable()) - httpServerThread_.join(); - threadPool_->stop(); } @@ -448,15 +481,17 @@ DhtProxyServer::subscribe(restinio::request_handle_t request, // One push listener per pushToken.infoHash.clientId auto pushListener = pushListeners_.emplace(pushToken, PushListener{}).first; auto listeners = pushListener->second.listeners.emplace(infoHash, std::vector{}).first; + + // Existing Listener for (auto& listener: listeners->second){ if (listener.clientId == clientId){ scheduler_.edit(listener.expireJob, timeout); scheduler_.edit(listener.expireNotifyJob, timeout - proxy::OP_MARGIN); - auto response = std::make_shared( this->initHttpResponse(request->create_response())); response->flush(); + // No Refresh if (!root.isMember("refresh") or !root["refresh"].asBool()){ dht_->get(infoHash, [this, response](const dht::Sp& value){ auto output = Json::writeString(jsonBuilder_, value->toJson()) + "\n"; @@ -475,16 +510,16 @@ DhtProxyServer::subscribe(restinio::request_handle_t request, return restinio::request_handling_status_t::accepted; } } + // New listener listeners->second.emplace_back(Listener{}); auto& listener = listeners->second.back(); listener.clientId = clientId; - - // New listener pushListener->second.isAndroid = isAndroid; // The listener is not found, so add it. listener.internalToken = dht_->listen(infoHash, - [this, infoHash, pushToken, isAndroid, clientId](const std::vector>& values, bool expired){ + [this, infoHash, pushToken, isAndroid, clientId] + (const std::vector>& values, bool expired){ threadPool_->run([this, infoHash, pushToken, isAndroid, clientId, values, expired](){ // Build message content Json::Value json; diff --git a/src/http.cpp b/src/http.cpp index 72707f4c..927b3317 100644 --- a/src/http.cpp +++ b/src/http.cpp @@ -83,10 +83,10 @@ ConnectionListener::~ConnectionListener() void ConnectionListener::state_changed(const restinio::connection_state::notice_t ¬ice) noexcept { + std::lock_guard lock(*lock_); + printf("[restinio] [connection:%li] listener is %s\n", notice.connection_id(), to_str(notice.cause()).c_str()); - - std::lock_guard lock(*lock_); printf("[restinio] [connection:%li] searching listeners\n", notice.connection_id()); //auto listener = listeners_->begin();