diff --git a/include/opendht/network_engine.h b/include/opendht/network_engine.h index 0dde4a6b..84dac9db 100644 --- a/include/opendht/network_engine.h +++ b/include/opendht/network_engine.h @@ -239,12 +239,6 @@ public: bool isRunning(sa_family_t af) const; inline want_t want () const { return dht_socket >= 0 && dht_socket6 >= 0 ? (WANT4 | WANT6) : -1; } - /** - * Cancel a request. Setting req->cancelled = true is not enough in the case - * a request is "persistent". - */ - void cancelRequest(Sp& req); - void connectivityChanged(sa_family_t); /************** @@ -458,7 +452,7 @@ private: * Sends a request to a node. Request::MAX_ATTEMPT_COUNT attempts will * be made before the request expires. */ - void sendRequest(Sp& request); + void sendRequest(const Sp& request); /** * Generates a new request id, skipping the invalid id. diff --git a/include/opendht/node.h b/include/opendht/node.h index a36b4c51..6afdcd61 100644 --- a/include/opendht/node.h +++ b/include/opendht/node.h @@ -85,8 +85,10 @@ struct Node { void update(const SockAddr&); - void requested(std::shared_ptr& req); - void received(time_point now, std::shared_ptr req); + void requested(const Sp& req); + void received(time_point now, const Sp& req); + Sp getRequest(const net::TransId& tid); + void cancelRequest(const Sp& req); void setExpired(); @@ -131,17 +133,11 @@ private: /* Number of times we accept authentication errors from this node. */ static const constexpr unsigned MAX_AUTH_ERRORS {3}; - std::list> requests_ {}; + std::map> requests_ {}; std::map> sockets_ {}; unsigned auth_errors {0}; bool expired_ {false}; - - void clearPendingQueue() { - requests_.remove_if([](std::weak_ptr& w) { - return w.expired(); - }); - } }; } diff --git a/src/dht.cpp b/src/dht.cpp index 878625e3..35a8ff19 100644 --- a/src/dht.cpp +++ b/src/dht.cpp @@ -866,35 +866,13 @@ Dht::cancelListen(const InfoHash& id, size_t token) if (st != store.end() && tokenlocal) st->second.local_listeners.erase(tokenlocal); - auto searches_cancel_listen = [&](std::map> srs) { - for (auto& sp : srs) { - auto& s = sp.second; - if (s->id != id) continue; - auto af_token = s->af == AF_INET ? std::get<1>(it->second) : std::get<2>(it->second); - if (af_token == 0) - continue; - Sp query; - const auto& ll = s->listeners.find(af_token); - if (ll != s->listeners.cend()) - query = ll->second.query; - for (auto& sn : s->nodes) { - if (s->listeners.empty()) { /* also erase requests for all searchnodes. */ - for (auto& ls : sn.listenStatus) - network_engine.cancelRequest(ls.second); - sn.listenStatus.clear(); - } else if (query) { - auto it = sn.listenStatus.find(query); - if (it != sn.listenStatus.end()) { - network_engine.cancelRequest(it->second); - sn.listenStatus.erase(it); - } - } - } - s->listeners.erase(af_token); - } + auto searches_cancel_listen = [&id](std::map>& srs, size_t token) { + auto srp = srs.find(id); + if (srp != srs.end() and token) + srp->second->cancelListen(token); }; - searches_cancel_listen(searches4); - searches_cancel_listen(searches6); + searches_cancel_listen(searches4, std::get<1>(it->second)); + searches_cancel_listen(searches6, std::get<2>(it->second)); listeners.erase(it); return true; } @@ -1331,7 +1309,7 @@ Dht::connectivityChanged(sa_family_t af) for (auto& sp : searches(af)) for (auto& sn : sp.second->nodes) { for (auto& ls : sn.listenStatus) - network_engine.cancelRequest(ls.second); + sn.node->cancelRequest(ls.second); sn.listenStatus.clear(); } reported_addr.erase(std::remove_if(reported_addr.begin(), reported_addr.end(), [&](const ReportedAddr& addr){ @@ -2090,14 +2068,15 @@ Dht::pingNode(const sockaddr* sa, socklen_t salen, DoneCallbackSimple&& cb) void Dht::onError(Sp req, net::DhtProtocolException e) { + const auto& node = req->node; if (e.getCode() == net::DhtProtocolException::UNAUTHORIZED) { - DHT_LOG.e(req->node->id, "[node %s] token flush", req->node->toString().c_str()); - req->node->authError(); - network_engine.cancelRequest(req); - for (auto& srp : searches(req->node->getFamily())) { + DHT_LOG.e(node->id, "[node %s] token flush", node->toString().c_str()); + node->authError(); + node->cancelRequest(req); + for (auto& srp : searches(node->getFamily())) { auto& sr = srp.second; for (auto& n : sr->nodes) { - if (n.node != req->node) continue; + if (n.node != node) continue; n.token.clear(); n.last_get_reply = time_point::min(); searchSendGetValues(sr); @@ -2105,8 +2084,8 @@ Dht::onError(Sp req, net::DhtProtocolException e) { } } } else if (e.getCode() == net::DhtProtocolException::NOT_FOUND) { - DHT_LOG.e(req->node->id, "[node %s] returned error 404: storage not found", req->node->toString().c_str()); - network_engine.cancelRequest(req); + DHT_LOG.e(node->id, "[node %s] returned error 404: storage not found", node->toString().c_str()); + node->cancelRequest(req); } } diff --git a/src/network_engine.cpp b/src/network_engine.cpp index 73991902..080fa737 100644 --- a/src/network_engine.cpp +++ b/src/network_engine.cpp @@ -232,23 +232,9 @@ NetworkEngine::isRunning(sa_family_t af) const } } -void -NetworkEngine::cancelRequest(Sp& req) -{ - if (req) { - req->cancel(); - req->node->closeSocket(req->socket); - requests.erase(req->tid); - } -} - void NetworkEngine::clear() -{ - for (auto& req : requests) - req.second->setExpired(); - requests.clear(); -} +{} void NetworkEngine::connectivityChanged(sa_family_t af) @@ -260,18 +246,16 @@ void NetworkEngine::requestStep(Sp sreq) { auto& req = *sreq; - if (not req.pending()) { - if (req.cancelled()) - requests.erase(req.tid); + if (not req.pending()) return; - } auto now = scheduler.time(); auto& node = *req.node; if (req.isExpired(now)) { DHT_LOG.e(node.id, "[node %s] expired !", node.toString().c_str()); node.setExpired(); - requests.erase(req.tid); + if (node.id == zeroes) + requests.erase(req.tid); return; } else if (req.attempt_count == 1) { req.on_expired(req, false); @@ -285,7 +269,8 @@ NetworkEngine::requestStep(Sp sreq) err == EAFNOSUPPORT) { node.setExpired(); - requests.erase(req.tid); + if (node.id == zeroes) + requests.erase(req.tid); } else { if (err != EAGAIN) { ++req.attempt_count; @@ -304,18 +289,13 @@ NetworkEngine::requestStep(Sp sreq) * be made before the request expires. */ void -NetworkEngine::sendRequest(Sp& request) +NetworkEngine::sendRequest(const Sp& request) { + auto& node = request->node; + if (node->id == zeroes) + requests.emplace(request->tid, request); request->start = scheduler.time(); - auto e = requests.emplace(request->tid, request); - if (not e.second) { - // Should not happen ! - // Try to handle this scenario as well as we can - e.first->second->setExpired(); - e.first->second = request; - DHT_LOG.e(request->node->id, "Request already existed (tid: %d)!", request->tid.getTid()); - } - request->node->requested(request); + node->requested(request); requestStep(request); } @@ -375,14 +355,6 @@ void NetworkEngine::blacklistNode(const Sp& n) { n->setExpired(); - for (auto rit = requests.begin(); rit != requests.end();) { - if (rit->second->node == n) { - //rit->second->cancel(); - requests.erase(rit++); - } else { - ++rit; - } - } blacklist.emplace(n->addr); } @@ -488,20 +460,20 @@ NetworkEngine::process(std::unique_ptr&& msg, const SockAddr& fro if (msg->type == MessageType::Error or msg->type == MessageType::Reply) { auto node = cache.getNode(msg->id, from, now, true); auto rsocket = node->getSocket(msg->tid); - //auto request = node->getRequest(msg->tid); + auto req = node->getRequest(msg->tid); /* either response for a request or data for an opened socket */ - auto req_it = requests.find(msg->tid); - if (req_it == requests.end() and not rsocket) { - throw DhtProtocolException {DhtProtocolException::UNKNOWN_TID, "Can't find transaction", msg->id}; - } - - auto req = req_it != requests.end() ? req_it->second : nullptr; - if (req and req->node != node and req->node->id != zeroes) { - node->received(now, req); - onNewNode(node, 2); - DHT_LOG.w(node->id, "[node %s] message received from unexpected node", node->toString().c_str()); - return; + if (not req and not rsocket) { + auto req_it = requests.find(msg->tid); + if (req_it != requests.end() and req_it->second->node->id == zeroes) { + req = req_it->second; + req->node = node; + requests.erase(req_it); + } else { + node->received(now, req); + onNewNode(node, 1); + throw DhtProtocolException {DhtProtocolException::UNKNOWN_TID, "Can't find transaction", msg->id}; + } } node->received(now, req); @@ -511,7 +483,6 @@ NetworkEngine::process(std::unique_ptr&& msg, const SockAddr& fro if (req and (req->cancelled() or req->expired() or req->completed())) { DHT_LOG.w(node->id, "[node %s] response to expired, cancelled or completed request", node->toString().c_str()); - requests.erase(req_it); return; } @@ -532,21 +503,19 @@ NetworkEngine::process(std::unique_ptr&& msg, const SockAddr& fro break; } case MessageType::Reply: - if (req) { /* request reply */ - if (msg->type == MessageType::AnnounceValue or msg->type == MessageType::Listen) - req->node->authSuccess(); + if (req) { /* request reply */ + if (msg->type == MessageType::AnnounceValue or msg->type == MessageType::Listen) + req->node->authSuccess(); + req->reply_time = scheduler.time(); - // erase before calling callback to make sure iterator is still valid - requests.erase(req_it); - req->reply_time = scheduler.time(); - - deserializeNodes(*msg, from); - req->setDone(std::move(*msg)); - break; - } else { /* request socket data */ - deserializeNodes(*msg, from); - rsocket->on_receive(node, std::move(*msg)); - } + deserializeNodes(*msg, from); + req->setDone(std::move(*msg)); + break; + } else { /* request socket data */ + deserializeNodes(*msg, from); + rsocket->on_receive(node, std::move(*msg)); + } + break; default: break; } diff --git a/src/node.cpp b/src/node.cpp index f4c3d0c8..a6bcc77e 100644 --- a/src/node.cpp +++ b/src/node.cpp @@ -41,11 +41,9 @@ Node::isGood(time_point now) const bool Node::isPendingMessage() const { - for (auto w : requests_) { - if (auto r = w.lock()) { - if (r->pending()) - return true; - } + for (auto r : requests_) { + if (r.second->pending()) + return true; } return false; } @@ -54,11 +52,9 @@ size_t Node::getPendingMessageCount() const { size_t count {0}; - for (auto w : requests_) { - if (auto r = w.lock()) { - if (r->pending()) - count++; - } + for (auto r : requests_) { + if (r.second->pending()) + count++; } return count; } @@ -71,27 +67,44 @@ Node::update(const SockAddr& new_addr) /** To be called when a message was sent to the node */ void -Node::requested(std::shared_ptr& req) +Node::requested(const Sp& req) { - requests_.emplace_back(req); + auto e = requests_.emplace(req->getTid(), req); + if (not e.second) { + // Should not happen ! + // Try to handle this scenario as well as we can + e.first->second->setExpired(); + e.first->second = req; + } } /** To be called when a message was received from the node. Req should be true if the message was an aswer to a request we made*/ void -Node::received(time_point now, std::shared_ptr req) +Node::received(time_point now, const Sp& req) { time = now; expired_ = false; if (req) { reply_time = now; - for (auto it = requests_.begin(); it != requests_.end();) { - auto r = it->lock(); - if (not r or r == req) - it = requests_.erase(it); - else - ++it; - } + requests_.erase(req->getTid()); + } +} + +Sp +Node::getRequest(const net::TransId& tid) +{ + auto it = requests_.find(tid); + return it != requests_.end() ? it->second : nullptr; +} + +void +Node::cancelRequest(const Sp& req) +{ + if (req) { + req->cancel(); + closeSocket(req->getSocket()); + requests_.erase(req->getTid()); } } @@ -99,9 +112,8 @@ void Node::setExpired() { expired_ = true; - for (auto w : requests_) { - if (auto r = w.lock()) - r->setExpired(); + for (auto r : requests_) { + r.second->setExpired(); } requests_.clear(); sockets_.clear(); diff --git a/src/search.h b/src/search.h index 81a3fe94..9d7ae861 100644 --- a/src/search.h +++ b/src/search.h @@ -77,6 +77,13 @@ struct Dht::SearchNode { SearchNode() : node() {} SearchNode(const Sp& node) : node(node) {} + ~SearchNode() { + if (node) { + cancelGet(); + cancelListen(); + cancelAnnounce(); + } + } /** * Can we use this node to listen/announce now ? @@ -173,6 +180,14 @@ struct Dht::SearchNode { } } + void cancelGet() { + for (const auto& status : getStatus) { + if (status.second->pending()) { + node->cancelRequest(status.second); + } + } + } + /** * Tells if a request in the status map is expired. * @@ -180,12 +195,12 @@ struct Dht::SearchNode { * * @return true if there exists an expired request, else false. */ - bool expired(const SyncStatus& status) const { + /*static bool expired(const SyncStatus& status) const { return std::find_if(status.begin(), status.end(), [](const SyncStatus::value_type& r){ return r.second and r.second->expired(); }) != status.end(); - } + }*/ /** * Tells if a request in the status map is pending. @@ -194,19 +209,29 @@ struct Dht::SearchNode { * * @return true if there exists an expired request, else false. */ - bool pending(const SyncStatus& status) const { + static bool pending(const SyncStatus& status) { return std::find_if(status.begin(), status.end(), [](const SyncStatus::value_type& r){ return r.second and r.second->pending(); }) != status.end(); } + bool pendingGet() const { return pending(getStatus); } + bool isAnnounced(Value::Id vid) const { auto ack = acked.find(vid); if (ack == acked.end() or not ack->second.first) return false; return ack->second.first->completed(); } + void cancelAnnounce() { + for (const auto& status : acked) { + const auto& req = status.second.first; + if (req->pending()) { + node->cancelRequest(req); + } + } + } bool isListening(time_point now) const { auto ls = listenStatus.begin(); @@ -229,6 +254,18 @@ struct Dht::SearchNode { return false; return listen_status->second->reply_time + LISTEN_EXPIRE_TIME > now; } + void cancelListen() { + for (const auto& status : listenStatus) + node->cancelRequest(status.second); + listenStatus.clear(); + } + void cancelListen(const Sp& query) { + auto it = listenStatus.find(query); + if (it != listenStatus.end()) { + node->cancelRequest(it->second); + listenStatus.erase(it); + } + } /** * Assumng the node is synced, should a "put" request be sent to this node now ? @@ -319,7 +356,7 @@ struct Dht::Search { unsigned currentlySolicitedNodeCount() const { unsigned count = 0; for (const auto& n : nodes) - if (not n.isBad() and n.pending(n.getStatus)) + if (not n.isBad() and n.pendingGet()) count++; return count; } @@ -382,6 +419,21 @@ struct Dht::Search { bool isAnnounced(Value::Id id) const; bool isListening(time_point now) const; + void cancelListen(size_t token) { + Sp query; + const auto& ll = listeners.find(token); + if (ll != listeners.cend()) { + query = ll->second.query; + listeners.erase(ll); + } + for (auto& sn : nodes) { + if (listeners.empty()) + sn.cancelListen(); + else if (query) + sn.cancelListen(query); + } + } + /** * @return The number of non-good search nodes. */ @@ -672,7 +724,7 @@ Dht::Search::getUpdateTime(time_point now) const for (const auto& sn : nodes) { if (sn.node->isExpired() or (sn.candidate and t >= TARGET_NODES)) continue; - auto pending = sn.pending(sn.getStatus); + auto pending = sn.pendingGet(); if (sn.last_get_reply < std::max(now - Node::NODE_EXPIRE_TIME, last_get) or pending) { // not isSynced if (not pending and solicited_nodes < MAX_REQUESTED_SEARCH_NODES)