diff --git a/include/opendht/dht_proxy_client.h b/include/opendht/dht_proxy_client.h index 7c3cd131..ac22747e 100644 --- a/include/opendht/dht_proxy_client.h +++ b/include/opendht/dht_proxy_client.h @@ -330,10 +330,12 @@ private: std::string pushClientId_; /* - * ASIO I/O Context for sockets used in requests + * ASIO I/O Context for sockets in httpClient_ + * Note: Each context is used in one thread only. */ - asio::io_context ctx_; + asio::io_context httpContext_; std::unique_ptr httpClient_; + std::thread httpClientThread_; mutable std::mutex lockCurrentProxyInfos_; NodeStatus statusIpv4_ {NodeStatus::Disconnected}; @@ -368,6 +370,11 @@ private: Sp statusTimer_; mutable std::mutex statusLock_; + /* + * ASIO I/O Context for periodic scheduler + * Note: Each context is used in one thread only. + */ + asio::io_context periodicContext_; Scheduler scheduler; /** * Retrieve if we can connect to the proxy (update statusIpvX_) diff --git a/src/dht_proxy_client.cpp b/src/dht_proxy_client.cpp index b7c520e5..1741ac57 100644 --- a/src/dht_proxy_client.cpp +++ b/src/dht_proxy_client.cpp @@ -76,10 +76,22 @@ DhtProxyClient::DhtProxyClient(std::function signal, const std::string& serverHost_(serverHost), pushClientId_(pushClientId), loopSignal_(signal), logger_(logger) { + // build http client auto hostAndPort = splitPort(serverHost_); serverHostIp_ = hostAndPort.first; serverHostPort_ = std::atoi(hostAndPort.second.c_str()); - httpClient_ = std::make_unique(ctx_, serverHostIp_, serverHostPort_, logger); + httpClient_ = std::make_unique(httpContext_, serverHostIp_, serverHostPort_, logger); + // run http client + httpClientThread_ = std::thread([this](){ + try { + DHT_LOG.d("Starting the HTTP Client io_context"); + httpContext_.run(); + DHT_LOG.d("HTTP Client io_context stopped"); + } + catch(const std::exception &ex){ + DHT_LOG.e("Error starting the HTTP Client io_context"); + } + }); if (serverHost_.find("://") == std::string::npos) serverHost_ = proxy::HTTP_PROTO + serverHost_; @@ -90,7 +102,7 @@ DhtProxyClient::DhtProxyClient(std::function signal, const std::string& asio::io_context& DhtProxyClient::io_context(){ - return ctx_; + return httpContext_; } void @@ -110,10 +122,10 @@ DhtProxyClient::startProxy() DHT_LOG.w("Staring proxy client to %s", serverHost_.c_str()); nextProxyConfirmationTimer_ = std::make_shared( - ctx_, std::chrono::steady_clock::now()); + periodicContext_, std::chrono::steady_clock::now()); nextProxyConfirmationTimer_->async_wait(std::bind(&DhtProxyClient::confirmProxy, this)); - listenerRestartTimer_ = std::make_shared(ctx_); + listenerRestartTimer_ = std::make_shared(periodicContext_); listenerRestartTimer_->async_wait(std::bind(&DhtProxyClient::restartListeners, this)); loopSignal_(); @@ -127,6 +139,8 @@ DhtProxyClient::~DhtProxyClient() if (infoState_) infoState_->cancel = true; statusTimer_->cancel(); + if (httpClientThread_.joinable()) + httpClientThread_.join(); } std::vector> @@ -150,8 +164,8 @@ DhtProxyClient::getLocalById(const InfoHash& k, Value::Id id) const { void DhtProxyClient::cancelAllOperations() { - if (!ctx_.stopped()) - ctx_.stop(); + if (!httpContext_.stopped()) + httpContext_.stop(); } void @@ -220,7 +234,11 @@ DhtProxyClient::isRunning(sa_family_t af) const time_point DhtProxyClient::periodic(const uint8_t*, size_t, const SockAddr&) { + // Execute all HTTP opeations + DHT_LOG.d("[periodic] running the io_context operations"); + periodicContext_.run(); // Exec all currently stored callbacks + DHT_LOG.d("[periodic] executing the stored callbacks"); scheduler.syncTime(); decltype(callbacks_) callbacks; { @@ -360,7 +378,7 @@ DhtProxyClient::put(const InfoHash& key, Sp val, DoneCallback cb, std::lock_guard lock(searchLock_); auto id = val->id; auto &search = searches_[key]; - auto refreshTimer = std::make_shared(ctx_, + auto refreshTimer = std::make_shared(periodicContext_, std::chrono::steady_clock::now() + proxy::OP_TIMEOUT - proxy::OP_MARGIN); auto ok = std::make_shared(false); // define refresh timer handler @@ -537,7 +555,7 @@ DhtProxyClient::getProxyInfos() // Try to contact the proxy and set the status to connected when done. // will change the connectivity status if (!statusTimer_) - statusTimer_ = std::make_shared(ctx_); + statusTimer_ = std::make_shared(periodicContext_); statusTimer_->async_wait(std::bind(&DhtProxyClient::handleProxyStatus, this, std::placeholders::_1, infoState));