diff --git a/include/opendht/network_engine.h b/include/opendht/network_engine.h index 91fb8aa2..99b394a8 100644 --- a/include/opendht/network_engine.h +++ b/include/opendht/network_engine.h @@ -280,6 +280,8 @@ public: * * @return the request with information concerning its success. */ + bool needsSession(const Node& node) const; + Sp sendPing(const Sp& n, RequestCb&& on_done, RequestExpiredCb&& on_expired); /** @@ -600,6 +602,9 @@ private: bool isNodeBlacklisted(const SockAddr& addr) const; void requestStep(Sp req); + void transmitRequest(const Sp& req); + void notifyRequestTimeout(Request& req); + void completeSession(Node& node); /* Re-establishes the session with a ping in an Initial envelope, unless one is pending */ void recoverSession(const Sp& node); /* Sends the requests withheld during the recovery, under the new session */ diff --git a/include/opendht/node_cache.h b/include/opendht/node_cache.h index 1a716534..efdf3287 100644 --- a/include/opendht/node_cache.h +++ b/include/opendht/node_cache.h @@ -24,6 +24,7 @@ struct NodeCache * To use in case of connectivity change etc. */ void clearBadNodes(sa_family_t family = 0); + void clear(); NodeCache(std::mt19937_64& r) : rd(r) {}; diff --git a/src/network_engine.cpp b/src/network_engine.cpp index 4b2494b0..94195c19 100644 --- a/src/network_engine.cpp +++ b/src/network_engine.cpp @@ -362,6 +362,7 @@ NetworkEngine::clear() request.second->node->setExpired(); } requests.clear(); + cache.clear(); partial_messages.clear(); } @@ -371,6 +372,22 @@ NetworkEngine::connectivityChanged(sa_family_t af) cache.clearBadNodes(af); } +void +NetworkEngine::notifyRequestTimeout(Request& req) +{ + if (not std::exchange(req.timeoutNotified, true) and req.on_expired) + req.on_expired(req, false); +} + +bool +NetworkEngine::needsSession(const Node& node) const +{ + if (not transport_) + return false; + const auto& peer = node.getTransport(); + return peer ? not peer->ephemeral or peer->recovering : node.getReplyTime() == time_point::min(); +} + void NetworkEngine::requestStep(Sp sreq) { @@ -388,23 +405,36 @@ NetworkEngine::requestStep(Sp sreq) requests.erase(req.tid); return; } else if (req.attempt_count == 1) { - if (req.on_expired) - req.on_expired(req, false); + notifyRequestTimeout(req); + if (not req.pending()) + return; // The peer may have lost our session or credential: start a new one if (const auto& peer = node.getTransport(); peer and peer->ephemeral) { peer->resetEphemeral(); peer->recovering = true; } } + req.withheld = req.type != MessageType::Ping and req.type != MessageType::FindNode and needsSession(node); + if (req.withheld) { + scheduler.cancel(req.retryJob); + recoverSession(req.node); + return; + } + transmitRequest(sreq); +} + +void +NetworkEngine::transmitRequest(const Sp& sreq) +{ + auto& req = *sreq; + if (not req.pending()) + return; + scheduler.cancel(req.retryJob); + auto now = scheduler.time(); + auto& node = *req.node; const auto& peer = node.getTransport(); req.initial = peer and not peer->ephemeral; - // After the first contact, payloads stay out of Initial envelopes: a later compromise of the receiver key reveals them - req.withheld = peer and peer->recovering and req.type != MessageType::Ping; - if (req.withheld) - recoverSession(req.node); - - auto err = req.withheld ? 0 - : send(node, + auto err = send(node, node.getAddr(), (char*) req.msg.data(), req.msg.size(), @@ -419,14 +449,16 @@ NetworkEngine::requestStep(Sp sreq) ++req.attempt_count; req.attempt_duration += req.attempt_duration + uniform_duration_distribution<>(0ms, ((duration) Node::MAX_RESPONSE_TIME) / 4)(rd); - if (not req.withheld and not req.parts.empty()) { + if (not req.parts.empty()) { sendValueParts(req.tid, req.parts, node, node.getAddr()); } } std::weak_ptr wreq = sreq; - scheduler.add(req.last_try + req.attempt_duration, [this, wreq] { - if (auto req = wreq.lock()) + req.retryJob = scheduler.add(req.last_try + req.attempt_duration, [this, wreq] { + if (auto req = wreq.lock()) { + req->retryJob.reset(); requestStep(req); + } }); } } @@ -435,21 +467,60 @@ void NetworkEngine::recoverSession(const Sp& node) { for (const auto& [tid, req] : node->getRequests()) - if (req->type == MessageType::Ping and req->pending()) + if (req->sessionProbe and req->pending()) return; - sendPing(node, {}, {}); + auto probe = sendPing(node, {}, [this](const Request& req, bool over) { + if (over) + return; + std::vector> waiting; + for (const auto& [tid, pending] : req.node->getRequests()) + if (pending->pending() and pending->withheld) + waiting.emplace_back(pending); + for (const auto& pending : waiting) + if (pending->pending()) + notifyRequestTimeout(*pending); + }); + probe->sessionProbe = true; +} + +void +NetworkEngine::completeSession(Node& node) +{ + if (const auto& peer = node.getTransport()) { + if (not peer->ephemeral) { + for (const auto& [tid, req] : node.getRequests()) { + if (req->pending() and req->withheld) { + recoverSession(req->node); + break; + } + } + return; + } + peer->recovering = false; + } + if (not needsSession(node)) + resumeRequests(node); } void NetworkEngine::resumeRequests(Node& node) { + std::vector> waiting; + std::vector> probes; for (const auto& [tid, req] : node.getRequests()) { - if (not req->withheld or not req->pending()) + if (req->pending() and req->sessionProbe) + probes.emplace_back(req); + else if (req->pending() and req->withheld) + waiting.emplace_back(req); + } + for (const auto& probe : probes) + node.cancelRequest(probe); + for (const auto& req : waiting) { + if (not req->pending() or needsSession(node)) continue; req->withheld = false; - req->initial = false; - if (send(node, node.getAddr(), (char*) req->msg.data(), req->msg.size()) == 0 and not req->parts.empty()) - sendValueParts(req->tid, req->parts, node, node.getAddr()); + req->attempt_duration = ((duration) Node::MAX_RESPONSE_TIME) / 2; + transmitRequest(req); } } @@ -748,6 +819,17 @@ NetworkEngine::process(std::unique_ptr&& msg, const SockAddr& fro trusted = true; } + if (msg->type == MessageType::Reply or msg->type == MessageType::Error) { + auto req = node and node->getAddr() == from ? node->getRequest(msg->tid) : nullptr; + if (not req) { + auto pending = requests.find(msg->tid); + if (pending != requests.end() and pending->second->node->getAddr() == from) + req = pending->second; + } + if (req ? not req->pending() : not (node and node->getSocket(msg->tid))) + return; + } + if (trusted) { node = cache.getNode(msg->id, from, now, true, msg->is_client); if (msg->peer) @@ -857,6 +939,7 @@ NetworkEngine::process(std::unique_ptr&& msg, const SockAddr& fro try { deserializeNodes(*msg, from); r.setDone(std::move(*msg)); + completeSession(*node); } catch (const DhtProtocolException& e) { if (logger_) logger_->warn("Unable to deserialize nodes {}", e.what()); @@ -1221,8 +1304,6 @@ NetworkEngine::attachPeer(Node& node, const Sp& peer) current->authorizedUntil = std::max(current->authorizedUntil, peer->authorizedUntil); if (peer->ephemeral) { current->setEphemeral(*peer->ephemeral, peer->validatedAddress); - if (std::exchange(current->recovering, false)) - resumeRequests(node); } } } diff --git a/src/node_cache.cpp b/src/node_cache.cpp index 73b720de..44d4dfc6 100644 --- a/src/node_cache.cpp +++ b/src/node_cache.cpp @@ -9,6 +9,12 @@ constexpr size_t CLEANUP_MAX_NODES {4 * 1024}; constexpr size_t CLEANUP_FREQ {4 * 1024}; NodeCache::~NodeCache() +{ + clear(); +} + +void +NodeCache::clear() { cache_4.setExpired(); cache_6.setExpired(); diff --git a/src/request.h b/src/request.h index e4154f12..d9ccfd36 100644 --- a/src/request.h +++ b/src/request.h @@ -3,6 +3,7 @@ #pragma once #include "net.h" +#include "scheduler.h" #include "value.h" namespace dht { @@ -133,6 +134,9 @@ private: void clear() { + if (retryJob) + retryJob->cancel(); + retryJob.reset(); on_done = {}; on_error = {}; on_expired = {}; @@ -159,6 +163,9 @@ private: bool initial {false}; /* the last attempt waited for the session recovery: never sent in an Initial envelope */ bool withheld {false}; + bool sessionProbe {false}; + bool timeoutNotified {false}; + Sp retryJob; }; } /* namespace net */