From 407d717c5ef2ffbe4b92285cfce1f74b63f53631 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Simon=20D=C3=A9saulniers?= Date: Thu, 10 Mar 2016 15:19:49 -0500 Subject: [PATCH] handling request status editing in NetworkEngine --- src/dht.cpp | 121 ++++++++++++++++++----------------------- src/network_engine.cpp | 10 +++- 2 files changed, 61 insertions(+), 70 deletions(-) diff --git a/src/dht.cpp b/src/dht.cpp index f12e80a7..5b78b99c 100644 --- a/src/dht.cpp +++ b/src/dht.cpp @@ -736,8 +736,6 @@ Dht::Search::insertNode(std::shared_ptr node, time_point now, const Blob& expired = false; } if (not token.empty()) { - n->getStatus->reply_time = now; - n->getStatus->last_try = TIME_INVALID; n->candidate = false; if (token.size() <= 64) n->token = token; @@ -771,7 +769,7 @@ Dht::expireSearches() Dht::SearchNode* Dht::searchSendGetValues(std::shared_ptr sr, SearchNode* pn, bool update) { - auto now = scheduler.time(); + const auto& now = scheduler.time(); const time_point up = update ? sr->getLastGetTime() : time_point::min(); SearchNode* n = nullptr; if (pn) { @@ -808,11 +806,12 @@ Dht::searchSendGetValues(std::shared_ptr sr, SearchNode* pn, bool update searchStep(sr); } }; - if (sr.callbacks.empty() and sr.listeners.empty()) - network_engine.sendFindNode(n->node, sr.id, -1, onDone, onExpired); + std::shared_ptr rstatus; + if (sr->callbacks.empty() and sr->listeners.empty()) + rstatus = network_engine.sendFindNode(n->node, sr->id, -1, onDone, onExpired); else - network_engine.sendGetValues(n->node, sr.id, -1, onDone, onExpired); - n->getStatus->last_try = now; + rstatus = network_engine.sendGetValues(n->node, sr->id, -1, onDone, onExpired); + n->getStatus = rstatus; pinged(*n->node); if (n->node->pinged > 1 and not n->candidate) { n->candidate = true; @@ -829,7 +828,7 @@ Dht::searchStep(std::shared_ptr sr) if (not sr) return; - auto now = scheduler.time(); + const auto& now = scheduler.time(); DHT_LOG.DEBUG("[search %s IPv%c] step", sr->id.toString().c_str(), sr->af == AF_INET ? '4' : '6'); sr->step_time = now; @@ -880,8 +879,7 @@ Dht::searchStep(std::shared_ptr sr) //std::cout << "Sending listen to " << n.node->id << " " << print_addr(n.node->ss, n.node->sslen) << std::endl; { - auto srp = &sr; - network_engine.sendListen(n.node, sr.id, n.token, + n.listenStatus = network_engine.sendListen(n.node, sr->id, n.token, [=](std::shared_ptr status, NetworkEngine::RequestAnswer&& answer) mutable { /* on done */ @@ -931,27 +929,19 @@ Dht::searchStep(std::shared_ptr sr) //std::cout << "Sending announce_value to " << n.node->id << " " << print_addr(n.node->ss, n.node->sslen) << std::endl; { - auto srp = &sr; - network_engine.sendAnnounceValue(n.node, sr.id, *a.value, a.created, n.token, - [=](std::shared_ptr, NetworkEngine::RequestAnswer&&) mutable - { /* on done */ - if (srp) { - searchStep(*srp); - } - }, - [=](std::shared_ptr, NetworkEngine::RequestAnswer&&) mutable - { /* on expired */ - if (srp) { - searchStep(*srp); - } + n.acked[vid] = network_engine.sendAnnounceValue(n.node, sr->id, *a.value, a.created, n.token, + [=](std::shared_ptr status, NetworkEngine::RequestAnswer&& answer) mutable + { /* on done */ + if (sr) { + onAnnounceDone(status, answer, sr); + searchStep(sr); } - ); - } - if (a_status == n.acked.end()) { - n.acked[vid] = std::make_shared(now); - } else { - a_status->second->last_try = now; - } + }, + [=](std::shared_ptr, NetworkEngine::RequestAnswer&&) mutable + { /* on expired */ + if (sr) { searchStep(sr); } + } + ); // use the "pending" flag so we update the "pinged" // fields after sending the announce requests for every value to announce n.pending = true; @@ -1026,7 +1016,7 @@ std::shared_ptr Dht::newSearch(InfoHash id, sa_family_t af) { auto& srs = af == AF_INET ? searches4 : searches6; - auto o = std::min_element(srs.begin(), srs.end(), + const auto& o = std::min_element(srs.begin(), srs.end(), [](std::pair>& lsr, std::pair>& rsr) { return lsr.second->done && rsr.second->step_time > lsr.second->step_time; }); @@ -1253,7 +1243,7 @@ Dht::Search::getNextStepTime(const std::map& types, ti void Dht::bootstrapSearch(Dht::Search& sr) { - auto now = scheduler.time(); + const auto& now = scheduler.time(); auto& list = (sr.af == AF_INET) ? buckets : buckets6; if (list.empty() || (list.size() == 1 && list.front().nodes.empty())) return; @@ -1346,7 +1336,7 @@ Dht::search(const InfoHash& id, sa_family_t af, GetCallback callback, DoneCallba void Dht::announce(const InfoHash& id, sa_family_t af, std::shared_ptr value, DoneCallback callback, time_point created) { - auto now = scheduler.time(); + const auto& now = scheduler.time(); if (!value) { if (callback) callback(false, {}); @@ -1402,7 +1392,7 @@ Dht::announce(const InfoHash& id, sa_family_t af, std::shared_ptr value, size_t Dht::listenTo(const InfoHash& id, sa_family_t af, GetCallback cb, Value::Filter f) { - auto now = scheduler.time(); + const auto& now = scheduler.time(); if (!isRunning(af)) return 0; // DHT_LOG.ERROR("[search %s IPv%c] search_time is now in %lfs", sr->id.toString().c_str(), (sr->af == AF_INET) ? '4' : '6', print_dt(tm-clock::now())); @@ -1699,7 +1689,7 @@ Dht::cancelPut(const InfoHash& id, const Value::Id& vid) void Dht::storageChanged(Storage& st, ValueStorage& v) { - auto now = scheduler.time(); + const auto& now = scheduler.time(); { std::vector>>> cbs; for (const auto& l : st.local_listeners) { @@ -1729,7 +1719,7 @@ Dht::storageChanged(Storage& st, ValueStorage& v) bool Dht::storageStore(const InfoHash& id, const std::shared_ptr& value, time_point created) { - auto now = scheduler.time(); + const auto& now = scheduler.time(); created = std::min(created, now); auto st = findStorage(id); if (st == store.end()) { @@ -1787,7 +1777,7 @@ Dht::Storage::clear() void Dht::storageAddListener(const InfoHash& id, const InfoHash& node, const sockaddr *from, socklen_t fromlen, size_t rid) { - auto now = scheduler.time(); + const auto& now = scheduler.time(); auto st = findStorage(id); if (st == store.end()) { if (store.size() >= MAX_HASHES) @@ -1816,7 +1806,7 @@ Dht::storageAddListener(const InfoHash& id, const InfoHash& node, const sockaddr void Dht::expireStorage() { - auto now = scheduler.time(); + const auto& now = scheduler.time(); auto i = store.begin(); while (i != store.end()) { // put elements to remove at the end with std::partition, @@ -1871,7 +1861,7 @@ Dht::Storage::expire(const std::map& types, time_point void Dht::connectivityChanged() { - auto now = scheduler.time(); + const auto& now = scheduler.time(); scheduler.add(now, std::bind(&Dht::confirmNodes, this)); mybucket_grow_time = now; mybucket6_grow_time = now; @@ -1889,7 +1879,7 @@ Dht::connectivityChanged() void Dht::rotateSecrets() { - auto now = scheduler.time(); + const auto& now = scheduler.time(); uniform_duration_distribution<> time_dist(std::chrono::minutes(15), std::chrono::minutes(45)); auto rotate_secrets_time = now + time_dist(rd); @@ -1954,7 +1944,7 @@ Dht::tokenMatch(const Blob& token, const sockaddr *sa) const int Dht::getNodesStats(sa_family_t af, unsigned *good_return, unsigned *dubious_return, unsigned *cached_return, unsigned *incoming_return) const { - auto now = scheduler.time(); + const auto& now = scheduler.time(); unsigned good = 0, dubious = 0, cached = 0, incoming = 0; auto& list = (af == AF_INET) ? buckets : buckets6; @@ -1985,7 +1975,7 @@ Dht::getNodesStats(sa_family_t af, unsigned *good_return, unsigned *dubious_retu void Dht::dumpBucket(const Bucket& b, std::ostream& out) const { - auto now = scheduler.time(); + const auto& now = scheduler.time(); using namespace std::chrono; out << b.first << " count " << b.nodes.size() << " age " << duration_cast(now - b.time).count() << " sec"; if (b.cached.ss_family) @@ -2010,7 +2000,7 @@ Dht::dumpBucket(const Bucket& b, std::ostream& out) const void Dht::dumpSearch(const Search& sr, std::ostream& out) const { - auto now = scheduler.time(); + const auto& now = scheduler.time(); using namespace std::chrono; out << std::endl << "Search IPv" << (sr.af == AF_INET6 ? '6' : '4') << ' ' << sr.id << " G" << sr.callbacks.size(); out << " age " << duration_cast(now - sr.step_time).count() << " s tid " << sr.tid; @@ -2119,7 +2109,7 @@ Dht::dumpTables() const std::string Dht::getStorageLog() const { - auto now = scheduler.time(); + const auto& now = scheduler.time(); using namespace std::chrono; std::stringstream out; for (const auto& st : store) { @@ -2198,10 +2188,10 @@ Dht::Dht(int s, int s6, Config config) std::generate_n(secret.begin(), secret.size(), std::bind(rand_byte, std::ref(rdev))); } rotateSecrets(); + expireBuckets(buckets); expireBuckets(buckets6); - DHT_LOG.DEBUG("DHT initialised with node ID %s", myid.toString().c_str()); } @@ -2322,7 +2312,7 @@ Dht::bucketMaintenance(RoutingTable& list) void Dht::dataPersistence() { - auto now = scheduler.time(); + const auto& now = scheduler.time(); auto storage_maintenance_time = time_point::max(); for (auto &str : store) { if (now > str.maintenance_time) { @@ -2337,7 +2327,7 @@ Dht::dataPersistence() { size_t Dht::maintainStorage(InfoHash id, bool force, DoneCallback donecb) { - auto now = scheduler.time(); + const auto& now = scheduler.time(); size_t announce_per_af = 0; auto local_storage = findStorage(id); if (local_storage == store.end()) { return 0; } @@ -2401,8 +2391,9 @@ Dht::processMessage(const uint8_t *buf, size_t buflen, const sockaddr *from, soc try { network_engine.processMessage(buf, buflen, from, fromlen); } catch (DhtProtocolException& e) { + DHT_LOG.ERROR("Can't parse message from %s: %s", e.getNodeId().toString().c_str(), e.what()); auto code = e.getCode(); - if (code == DhtProtocolException::INVALID_REPLY_TID + if (code == DhtProtocolException::INVALID_TID_SIZE or code == DhtProtocolException::WRONG_NODE_INFO_BUF_LEN) { /* This is really annoying, as it means that we will time-out all our searches that go through this node. @@ -2416,11 +2407,14 @@ Dht::processMessage(const uint8_t *buf, size_t buflen, const sockaddr *from, soc time_point Dht::periodic(const uint8_t *buf, size_t buflen, const sockaddr *from, socklen_t fromlen) { + scheduler.syncTime(); processMessage(buf, buflen, from, fromlen); return scheduler.run(); } -void Dht::expire() { +void +Dht::expire() +{ uniform_duration_distribution<> time_dis(std::chrono::minutes(2), std::chrono::minutes(6)); auto expire_stuff_time = scheduler.time() + duration(time_dis(rd)); @@ -2436,7 +2430,7 @@ Dht::confirmNodes() { using namespace std::chrono; bool soon = false; - auto now = scheduler.time(); + const auto& now = scheduler.time(); if (searches4.empty() and searches6.empty() and getStatus() != Status::Disconnected) { get(myid, GetCallbackSimple{}); @@ -2530,7 +2524,7 @@ Dht::importValues(const std::vector& import) std::vector Dht::exportNodes() { - auto now = scheduler.time(); + const auto& now = scheduler.time(); std::vector nodes; const auto b4 = buckets.findBucket(myid); if (b4 != buckets.end()) { @@ -2572,6 +2566,7 @@ Dht::insertNode(const InfoHash& id, const sockaddr *sa, socklen_t salen) int Dht::pingNode(const sockaddr *sa, socklen_t salen) { + scheduler.syncTime(); DHT_LOG.DEBUG("Sending ping to %s", print_addr(sa, salen).c_str()); network_engine.sendPing(sa, salen, nullptr, nullptr); return -1; @@ -2589,8 +2584,6 @@ Dht::onError(std::shared_ptr status, DhtProtocolEx for (auto& n : sr->nodes) { if (n.node != status->node) continue; cleared++; - n.getStatus->last_try = TIME_INVALID; - n.getStatus->reply_time = TIME_INVALID; if (searchSendGetValues(sr)) sr->get_step_time = scheduler.time(); break; @@ -2619,7 +2612,7 @@ Dht::onPing(std::shared_ptr) NetworkEngine::RequestAnswer Dht::onFindNode(std::shared_ptr node, InfoHash& hash, want_t want) { - auto now = scheduler.time(); + const auto& now = scheduler.time(); Blob ntoken = makeToken((sockaddr*)&node->ss, false); std::vector> nodes, nodes6; if ((want & WANT4)) { @@ -2637,7 +2630,7 @@ Dht::onFindNode(std::shared_ptr node, InfoHash& hash, want_t want) NetworkEngine::RequestAnswer Dht::onGetValues(std::shared_ptr node, InfoHash& hash, want_t) { - auto now = scheduler.time(); + const auto& now = scheduler.time(); NetworkEngine::RequestAnswer* answer; DHT_LOG.DEBUG("[node %s %s] got 'get' request for %s.", node->id.toString().c_str(), print_addr(node->ss, node->sslen).c_str(), hash.toString().c_str()); @@ -2676,9 +2669,7 @@ Dht::onGetValuesDone(std::shared_ptr status, Netwo if (not sr) return; - auto now = scheduler.time(); - if (sr == nullptr) - return; + const auto& now = scheduler.time(); sr->insertNode(status->node, now, a.ntoken); if (!a.values.empty()) { DHT_LOG.DEBUG("[search %s IPv%c] found %u values", @@ -2737,7 +2728,7 @@ Dht::onListen(std::shared_ptr node, InfoHash& hash, Blob& token, size_t ri void Dht::onListenDone(std::shared_ptr status, NetworkEngine::RequestAnswer&, std::shared_ptr sr) { - auto now = scheduler.time(); + const auto& now = scheduler.time(); DHT_LOG.DEBUG("Got reply to listen."); if (sr) { for (auto& sn : sr->nodes) @@ -2824,20 +2815,16 @@ Dht::onAnnounce(std::shared_ptr node, InfoHash& hash, Blob& token, std::ve } void -Dht::onAnnounceDone(std::shared_ptr status, NetworkEngine::RequestAnswer& answer, std::shared_ptr sr) +Dht::onAnnounceDone(std::shared_ptr, + NetworkEngine::RequestAnswer& answer, + std::shared_ptr sr) { - auto now = scheduler.time(); + const auto& now = scheduler.time(); DHT_LOG.DEBUG("[search %s IPv%c] got reply to put!", sr->id.toString().c_str(), sr->af == AF_INET ? '4' : '6', answer.values.size()); auto& v = answer.values.front(); - for (auto& sn : sr->nodes) - if (sn.node == status->node) { - auto it = sn.acked[v->id] = status; - break; - } - /* See comment for gp above. */ if (searchSendGetValues(sr)) sr->get_step_time = now; diff --git a/src/network_engine.cpp b/src/network_engine.cpp index 111417aa..6422e127 100644 --- a/src/network_engine.cpp +++ b/src/network_engine.cpp @@ -128,8 +128,6 @@ NetworkEngine::processMessage(const uint8_t *buf, size_t buflen, const sockaddr uint16_t ttid = 0; if (msg.type == MessageType::Error or msg.type == MessageType::Reply) { - auto node = onNewNode(msg.id, from, fromlen, 2); - onReportedAddr(msg.id, (sockaddr*)&msg.addr.first, msg.addr.second); Request* req = nullptr; const auto& reqp = requests.find(msg.tid[2]); if (reqp != requests.end()) @@ -138,12 +136,16 @@ NetworkEngine::processMessage(const uint8_t *buf, size_t buflen, const sockaddr if (not req) throw DhtProtocolException {DhtProtocolException::UNKNOWN_TID, "Can't find transaction", msg.id}; + auto node = onNewNode(msg.id, from, fromlen, 2); + onReportedAddr(msg.id, (sockaddr*)&msg.addr.first, msg.addr.second); switch (msg.type) { case MessageType::Error: { if (msg.error_code == DhtProtocolException::UNAUTHORIZED && msg.id != zeroes && (msg.tid.matches(TransPrefix::ANNOUNCE_VALUES, &ttid) || msg.tid.matches(TransPrefix::LISTEN, &ttid))) { + req->status->last_try = TIME_INVALID; + req->status->reply_time = TIME_INVALID; onError(req->status, DhtProtocolException {DhtProtocolException::UNAUTHORIZED}); } else { DHT_LOG.WARN("[node %s %s] received unknown error message %u", @@ -153,7 +155,9 @@ NetworkEngine::processMessage(const uint8_t *buf, size_t buflen, const sockaddr break; } case MessageType::Reply: - req->on_done(req->status, msg.tid[2], std::move(msg)); + req->on_done(req->status, std::move(msg)); + req->status->reply_time = scheduler.time(); + requests.erase(reqp); break; default: break;