From b9604807ea072a9000fb050ff4a7dceb2564364c Mon Sep 17 00:00:00 2001 From: Seva Date: Wed, 12 Jun 2019 16:36:18 -0400 Subject: [PATCH] dhtproxy: replace threads by io_context replace scheduler and its jobs by asio timers --- include/opendht/dht_proxy_server.h | 39 ++-- src/dht_proxy_server.cpp | 314 +++++++++++++---------------- 2 files changed, 163 insertions(+), 190 deletions(-) diff --git a/include/opendht/dht_proxy_server.h b/include/opendht/dht_proxy_server.h index 892cc109..1db5eb68 100644 --- a/include/opendht/dht_proxy_server.h +++ b/include/opendht/dht_proxy_server.h @@ -73,7 +73,6 @@ namespace Json { namespace dht { class DhtRunner; -class ThreadPool; /** * Describes the REST API @@ -313,6 +312,8 @@ private: #endif //OPENDHT_PUSH_NOTIFICATIONS + void asyncPrintStats(); + using clock = std::chrono::steady_clock; using time_point = clock::time_point; @@ -322,17 +323,11 @@ private: std::unique_ptr httpServerThreadPool_; std::unique_ptr> httpServer_; - std::thread httpServerThread_ {}; - std::mutex schedulerLock_; - std::condition_variable schedulerCv_; - Scheduler scheduler_; - std::thread schedulerThread_; - std::unique_ptr threadPool_; - - Sp printStatsJob_; mutable std::mutex statsMutex_; + mutable ServerStats stats_; mutable NodeInfo nodeInfo_ {}; + std::unique_ptr printStatsTimer_; // Thread-safe access to listeners map. std::shared_ptr lockListener_; @@ -342,8 +337,17 @@ private: // Connection Listener observing conn state changes. std::shared_ptr connListener_; - struct PermanentPut; - struct SearchPuts; + struct PermanentPut { + time_point expiration; + std::string pushToken; + std::string clientId; + std::unique_ptr expireTimer; + std::unique_ptr expireNotifyTimer; + }; + struct SearchPuts { + std::map puts; + }; + std::mutex lockSearchPuts_; std::map puts_; mutable std::atomic requestNum_ {0}; @@ -351,11 +355,16 @@ private: const std::string pushServer_; - mutable ServerStats stats_; - #ifdef OPENDHT_PUSH_NOTIFICATIONS - struct Listener; - struct PushListener; + struct Listener { + std::string clientId; + std::future internalToken; + std::unique_ptr expireTimer; + std::unique_ptr expireNotifyTimer; + }; + struct PushListener { + std::map> listeners; + }; std::mutex lockPushListeners_; std::map pushListeners_; proxy::ListenToken tokenPushNotif_ {0}; diff --git a/src/dht_proxy_server.cpp b/src/dht_proxy_server.cpp index ba672c19..bf909273 100644 --- a/src/dht_proxy_server.cpp +++ b/src/dht_proxy_server.cpp @@ -36,17 +36,6 @@ using namespace std::placeholders; namespace dht { -struct DhtProxyServer::PermanentPut { - time_point expiration; - std::string pushToken; - std::string clientId; - Sp expireJob; - Sp expireNotifyJob; -}; -struct DhtProxyServer::SearchPuts { - std::map puts; -}; - constexpr const std::chrono::minutes PRINT_STATS_PERIOD {2}; constexpr const size_t IO_THREADS_MAX {64}; @@ -55,8 +44,7 @@ DhtProxyServer::DhtProxyServer(std::shared_ptr dht, in_port_t port, std::shared_ptr logger, const unsigned int ioThreads ): - dht_(dht), logger_(logger), threadPool_(new ThreadPool(IO_THREADS_MAX)), - lockListener_(std::make_shared()), + dht_(dht), logger_(logger), lockListener_(std::make_shared()), listeners_(std::make_shared>()), connListener_(std::make_shared( dht, listeners_, lockListener_, logger)), @@ -86,46 +74,20 @@ DhtProxyServer::DhtProxyServer(std::shared_ptr dht, in_port_t port, std::forward(settings) )); // run http server - httpServerThread_ = std::thread([this](){ - try { - 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; - } - }); - schedulerThread_ = std::thread([this](){ - while (not httpServerThread_.joinable()){ - std::this_thread::sleep_for(std::chrono::seconds(1)); - } - while (httpServerThread_.joinable()){ - std::unique_lock lock(schedulerLock_); - auto next = scheduler_.run(); - if (next == time_point::max()) - schedulerCv_.wait(lock); - else - schedulerCv_.wait_until(lock, next); - } - }); + try { + httpServer_->open_async([]{/*Ok.*/}, [](std::exception_ptr ex){ + std::rethrow_exception(ex); + }); + httpServerThreadPool_->start(); + } + catch(const std::exception &ex){ + std::cerr << "Error starting RESTinio: " << ex.what() << std::endl; + } dht->forwardAllMessages(true); - printStatsJob_ = scheduler_.add(scheduler_.time() + PRINT_STATS_PERIOD, [this] { - - if (httpServerThread_.joinable()) - updateStats(); - - // Refresh stats cache - auto newInfo = dht_->getNodeInfo(); - { - std::lock_guard lck(statsMutex_); - nodeInfo_ = std::move(newInfo); - } - scheduler_.edit(printStatsJob_, scheduler_.time() + PRINT_STATS_PERIOD); - }); + printStatsTimer_ = std::make_unique( + httpServer_->io_context(), PRINT_STATS_PERIOD); + printStatsTimer_->async_wait(std::bind(&DhtProxyServer::asyncPrintStats, this)); } DhtProxyServer::~DhtProxyServer() @@ -170,13 +132,6 @@ DhtProxyServer::makeHttpServerSettings(const in_port_t port, const unsigned int void DhtProxyServer::stop() { - if (printStatsJob_) - printStatsJob_->cancel(); - - std::lock_guard lock(*lockListener_); - for (auto it = listeners_->begin(); it != listeners_->end(); ++it) - it->second.response->done(); - logger_->d("[restinio] closing http server async operations"); httpServer_->close_async([this]{ logger_->d("[restinio] stopping http server io_context thread pool"); @@ -190,13 +145,6 @@ DhtProxyServer::stop() logger_->e("[restinio] error closing http server async: %s", ex.what()); } }); - if (httpServerThread_.joinable()) - httpServerThread_.join(); - - if (schedulerThread_.joinable()) - schedulerThread_.join(); - - threadPool_->stop(); } void @@ -215,6 +163,26 @@ DhtProxyServer::updateStats() const stats_.nodeInfo = nodeInfo_; } +void +DhtProxyServer::asyncPrintStats() +{ + if (httpServer_->io_context().stopped()) + return; + + if (dht_){ + updateStats(); + // Refresh stats cache + auto newInfo = dht_->getNodeInfo(); + std::lock_guard lck(statsMutex_); + nodeInfo_ = std::move(newInfo); + auto json = nodeInfo_.toJson(); + auto str = Json::writeString(jsonBuilder_, json); + logger_->d("[stats] %s", str.c_str()); + } + printStatsTimer_->expires_at(printStatsTimer_->expiry() + PRINT_STATS_PERIOD); + printStatsTimer_->async_wait(std::bind(&DhtProxyServer::asyncPrintStats, this)); +} + template HttpResponse DhtProxyServer::initHttpResponse(HttpResponse response) const { @@ -406,18 +374,6 @@ DhtProxyServer::listen(restinio::request_handle_t request, #ifdef OPENDHT_PUSH_NOTIFICATIONS -struct DhtProxyServer::Listener { - std::string clientId; - std::future internalToken; - Sp expireJob; - Sp expireNotifyJob; -}; - -struct DhtProxyServer::PushListener { - std::map> listeners; - bool isAndroid; -}; - RequestStatus DhtProxyServer::subscribe(restinio::request_handle_t request, restinio::router::route_params_t params) @@ -434,7 +390,6 @@ DhtProxyServer::subscribe(restinio::request_handle_t request, response.set_body(this->RESP_MSG_SERVICE_UNAVAILABLE); return response.done(); } - try { std::string err; Json::Value root; @@ -458,94 +413,96 @@ DhtProxyServer::subscribe(restinio::request_handle_t request, auto isAndroid = platform == "android"; auto clientId = root.isMember("client_id") ? root["client_id"].asString() : std::string(); - std::cout << "Subscribe " << infoHash << " client:" << clientId << std::endl; - { - std::lock(schedulerLock_, *lockListener_); - std::lock_guard lk1(*lockListener_, std::adopt_lock); - std::lock_guard lk2(schedulerLock_, std::adopt_lock); - scheduler_.syncTime(); - auto timeout = scheduler_.time() + proxy::OP_TIMEOUT; - // Check if listener is already present and refresh timeout if launched - // One push listener per pushToken.infoHash.clientId - auto pushListener = pushListeners_.emplace(pushToken, PushListener{}).first; - auto listeners = pushListener->second.listeners.emplace(infoHash, std::vector{}).first; + logger_->w("[subscribe] %s client: %s", infoHash.toString().c_str(), clientId.c_str()); + // ================ Search for existing listener =================== + // start the timer + auto timeout = std::chrono::steady_clock::now() + proxy::OP_TIMEOUT; + std::lock_guard lock(lockPushListeners_); - // 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(); + // Insert new or return existing push listeners of a token + auto pushListener = pushListeners_.emplace(pushToken, PushListener{}).first; + auto pushListeners = pushListener->second.listeners.emplace(infoHash, std::vector{}).first; - // 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"; - response->append_chunk(output); - response->flush(); - return true; - }, - [response] (bool /*ok*/){ - response->done(); - }); - } else { - response->append_chunk("{}\n"); + for (auto &listener: pushListeners->second){ + logger_->w("[subscribe] found client_id: %s", listener.clientId.c_str()); + // Found -> Resubscribe + if (listener.clientId == clientId){ + // Reset timers + listener.expireTimer->expires_at(timeout); + listener.expireNotifyTimer->expires_at(timeout - proxy::OP_MARGIN); + // Send response header + 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"; + response->append_chunk(output); + response->flush(); + return true; + }, + [response] (bool){ response->done(); - } - schedulerCv_.notify_one(); - return restinio::request_handling_status_t::accepted; - } - } - // New listener - listeners->second.emplace_back(Listener{}); - auto& listener = listeners->second.back(); - listener.clientId = clientId; - 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){ - threadPool_->run([this, infoHash, pushToken, isAndroid, clientId, values, expired](){ - // Build message content - Json::Value json; - json["key"] = infoHash.toString(); - json["to"] = clientId; - if (expired and values.size() < 3){ - std::stringstream ss; - for(size_t i = 0; i < values.size(); ++i){ - if(i != 0) ss << ","; - ss << values[i]->id; - } - json["exp"] = ss.str(); - } - sendPushNotification(pushToken, std::move(json), isAndroid); }); - return true; + // Refresh + } else { + response->append_chunk("{}\n"); + response->done(); } - ); - listener.expireJob = scheduler_.add(timeout, - [this, clientId, infoHash, pushToken] { - cancelPushListen(pushToken, infoHash, clientId); - } - ); - listener.expireNotifyJob = scheduler_.add(timeout - proxy::OP_MARGIN, - [this, infoHash, pushToken, isAndroid, clientId] { - std::cout << "Listener: sending refresh " << infoHash << std::endl; - Json::Value json; - json["timeout"] = infoHash.toString(); - json["to"] = clientId; - sendPushNotification(pushToken, std::move(json), isAndroid); - } - ); + return restinio::request_handling_status_t::accepted; + } } - schedulerCv_.notify_one(); + // =========== No existing listener for an infoHash ============ + // Add new listener to list of listeners + pushListeners->second.emplace_back(Listener{}); + auto &listener = pushListeners->second.back(); + listener.clientId = clientId; + + // Add listen on dht + listener.internalToken = dht_->listen(infoHash, + [this, infoHash, pushToken, isAndroid, clientId] + (const std::vector>& values, bool expired){ + // Build message content + Json::Value json; + json["key"] = infoHash.toString(); + json["to"] = clientId; + if (expired and values.size() < 2){ + std::stringstream ss; + for(size_t i = 0; i < values.size(); ++i){ + if(i != 0) ss << ","; + ss << values[i]->id; + } + json["exp"] = ss.str(); + } + sendPushNotification(pushToken, std::move(json), isAndroid); + return true; + } + ); + // Init & set timers + auto &ctx = httpServer_->io_context(); + listener.expireTimer = std::make_unique(ctx, timeout); + listener.expireNotifyTimer = std::make_unique(ctx, + timeout - proxy::OP_MARGIN); + // Launch timers + listener.expireTimer->async_wait(std::bind( + &DhtProxyServer::cancelPushListen, this, pushToken, infoHash, clientId)); + + listener.expireNotifyTimer->async_wait( + [this, infoHash, pushToken, isAndroid, clientId](const asio::error_code &ec){ + logger_->d("[subscribe] sending refresh %s", infoHash.toString().c_str()); + if (ec) + logger_->d("[subscribe] error sending refresh: %s", ec.message().c_str()); + Json::Value json; + json["timeout"] = infoHash.toString(); + json["to"] = clientId; + sendPushNotification(pushToken, std::move(json), isAndroid); + }); auto response = this->initHttpResponse(request->create_response()); response.set_body("{}\n"); return response.done(); - } catch (...){ + } + catch (...) { auto response = this->initHttpResponse( request->create_response(restinio::status_internal_server_error())); response.set_body(this->RESP_MSG_INTERNAL_SERVER_ERRROR); @@ -592,7 +549,8 @@ DhtProxyServer::unsubscribe(restinio::request_handle_t request, cancelPushListen(pushToken, infoHash, clientId); auto response = this->initHttpResponse(request->create_response()); return response.done(); - } catch (...){ + } + catch (...) { auto response = this->initHttpResponse( request->create_response(restinio::status_internal_server_error())); response.set_body(this->RESP_MSG_INTERNAL_SERVER_ERRROR); @@ -603,14 +561,16 @@ DhtProxyServer::unsubscribe(restinio::request_handle_t request, void DhtProxyServer::cancelPushListen(const std::string& pushToken, const dht::InfoHash& key, const std::string& clientId) { - std::cout << "cancelPushListen: " << key << " clientId:" << clientId << std::endl; + logger_->d("[cancelpushlisten] %s %s", key.toString().c_str(), clientId.c_str()); std::lock_guard lock(*lockListener_); + auto pushListener = pushListeners_.find(pushToken); if (pushListener == pushListeners_.end()) return; auto listeners = pushListener->second.listeners.find(key); if (listeners == pushListener->second.listeners.end()) return; + for (auto listener = listeners->second.begin(); listener != listeners->second.end();){ if (listener->clientId == clientId){ if (dht_) @@ -620,12 +580,10 @@ DhtProxyServer::cancelPushListen(const std::string& pushToken, const dht::InfoHa ++listener; } } - if (listeners->second.empty()){ + if (listeners->second.empty()) pushListener->second.listeners.erase(listeners); - } - if (pushListener->second.listeners.empty()){ + if (pushListener->second.listeners.empty()) pushListeners_.erase(pushListener); - } } void @@ -703,8 +661,8 @@ DhtProxyServer::cancelPut(const InfoHash& key, Value::Id vid) return; if (dht_) dht_->cancelPut(key, vid); - if (put->second.expireNotifyJob) - put->second.expireNotifyJob->cancel(); + if (put->second.expireNotifyTimer) + put->second.expireNotifyTimer->cancel(); sPutsMap.erase(put); if (sPutsMap.empty()) puts_.erase(sPuts); @@ -752,25 +710,32 @@ DhtProxyServer::put(restinio::request_handle_t request, clientId = pVal["client_id"].asString(); platform = pVal["platform"].asString(); } - std::unique_lock lock(schedulerLock_); - scheduler_.syncTime(); - auto timeout = scheduler_.time() + proxy::OP_TIMEOUT; + std::unique_lock lock(lockSearchPuts_); + auto timeout = std::chrono::steady_clock::now() + proxy::OP_TIMEOUT; auto vid = value->id; auto sPuts = puts_.emplace(infoHash, SearchPuts{}).first; auto r = sPuts->second.puts.emplace(vid, PermanentPut{}); auto& pput = r.first->second; if (r.second){ - pput.expireJob = scheduler_.add(timeout, [this, infoHash, vid]{ + auto &ctx = httpServer_->io_context(); + pput.expireTimer = std::make_unique(ctx, timeout); + pput.expireTimer->async_wait([this, infoHash, vid](const asio::error_code &ec){ std::cout << "Permanent put expired: " << infoHash << " " << vid << std::endl; + if (ec) + std::cout << "Permanent put error: " << ec.message() << std::endl; cancelPut(infoHash, vid); }); #ifdef OPENDHT_PUSH_NOTIFICATIONS if (not pushToken.empty()){ bool isAndroid = platform == "android"; - pput.expireNotifyJob = scheduler_.add(timeout - proxy::OP_MARGIN, - [this, infoHash, vid, pushToken, clientId, isAndroid] + pput.expireNotifyTimer = std::make_unique(ctx, + timeout - proxy::OP_MARGIN); + pput.expireNotifyTimer->async_wait( + [this, infoHash, vid, pushToken, clientId, isAndroid](const asio::error_code &ec) { std::cout << "Permanent put refresh: " << infoHash << " " << vid << std::endl; + if (ec) + std::cout << "Permanent put refresh error: " << ec.message() << std::endl; Json::Value json; json["timeout"] = infoHash.toString(); json["to"] = clientId; @@ -780,12 +745,11 @@ DhtProxyServer::put(restinio::request_handle_t request, } #endif } else { - scheduler_.edit(pput.expireJob, timeout); - if (pput.expireNotifyJob) - scheduler_.edit(pput.expireNotifyJob, timeout - proxy::OP_MARGIN); + pput.expireTimer->expires_at(timeout); + if (pput.expireNotifyTimer) + pput.expireNotifyTimer->expires_at(timeout - proxy::OP_MARGIN); } lock.unlock(); - schedulerCv_.notify_one(); } dht_->put(infoHash, value, [this, request, value](bool ok){ if (ok){