diff --git a/include/opendht/dht.h b/include/opendht/dht.h index 0b2001bc..d577933c 100644 --- a/include/opendht/dht.h +++ b/include/opendht/dht.h @@ -68,6 +68,7 @@ struct Node { } bool isExpired(time_point now) const; bool isGood(time_point now) const; + bool isMessagePending(time_point now) const; NodeExport exportNode() const { return NodeExport {id, ss, sslen}; } sa_family_t getFamily() const { return ss.ss_family; } @@ -400,8 +401,11 @@ private: time_point reply_time {}; /* the time of the last confirmation */ RequestStatus() {}; RequestStatus(time_point q, time_point a = {}) : request_time(q), reply_time(a) {}; - bool expired(time_point now) { - return (reply_time < request_time && now - request_time <= Node::MAX_RESPONSE_TIME); + bool expired(time_point now) const { + return reply_time < request_time && now > request_time + Node::MAX_RESPONSE_TIME; + } + bool pending(time_point now) const { + return reply_time < request_time && now - request_time <= Node::MAX_RESPONSE_TIME; } }; typedef std::map AnnounceStatusMap; @@ -446,8 +450,6 @@ private: if (listenStatus.reply_time > listenStatus.request_time) return listenStatus.reply_time + LISTEN_EXPIRE_TIME - REANNOUNCE_MARGIN; return listenStatus.request_time + Node::MAX_RESPONSE_TIME; - //time_point min_t = listenStatus.request_time + MAX_RESPONSE_TIME; - //return listenStatus.reply_time.time_since_epoch().count() ? std::max(listenStatus.reply_time + NODE_EXPIRE_TIME - REANNOUNCE_MARGIN, min_t) : min_t; } std::shared_ptr node {}; @@ -458,11 +460,20 @@ private: Blob token {}; + /** + * A search node is candidate if the search is/was synced and this node is a new candidate for inclusion + * + */ + bool candidate {false}; + // Generic temporary flag. // Must be reset to false after use by the algorithm. bool pending {false}; }; + /** + * A single "get" operation data + */ struct Get { time_point start; Value::Filter filter; @@ -470,11 +481,17 @@ private: DoneCallback done_cb; }; + /** + * A single "put" operation data + */ struct Announce { std::shared_ptr value; DoneCallback callback; }; + /** + * A single "listen" operation data + */ struct LocalListener { Value::Filter filter; GetCallback get_cb; @@ -674,7 +691,7 @@ private: time_point now; time_point mybucket_grow_time {}, mybucket6_grow_time {}; time_point expire_stuff_time {}; - time_point search_time {}; + time_point search_time {time_point::max()}; time_point confirm_nodes_time {}; time_point rotate_secrets_time {}; std::queue rate_limit_time {}; @@ -795,7 +812,7 @@ private: /** * If update is true, this method will also send message to synced but non-updated search nodes. */ - bool searchSendGetValues(Search& sr, SearchNode *n = nullptr, bool update = false); + bool searchSendGetValues(Search& sr, SearchNode *n = nullptr, bool update = true); void searchStep(Search& sr); void dumpSearch(const Search& sr, std::ostream& out) const; diff --git a/include/opendht/value.h b/include/opendht/value.h index cfba20a7..822d4be9 100644 --- a/include/opendht/value.h +++ b/include/opendht/value.h @@ -65,7 +65,7 @@ using time_point = clock::time_point; using duration = clock::duration; static /*constexpr*/const time_point TIME_INVALID = {}; -static /*constexpr*/const time_point TIME_MAX {duration{std::numeric_limits::max()}}; +static /*constexpr*/const time_point TIME_MAX {time_point::max()}; template class uniform_duration_distribution : public std::uniform_int_distribution { diff --git a/src/dht.cpp b/src/dht.cpp index 31bee8d3..6fe905a8 100644 --- a/src/dht.cpp +++ b/src/dht.cpp @@ -336,6 +336,12 @@ Node::isExpired(time_point now) const return pinged >= 3 && reply_time < pinged_time && pinged_time + MAX_RESPONSE_TIME < now; } +bool +Node::isMessagePending(time_point now) const +{ + return reply_time < pinged_time && pinged_time + MAX_RESPONSE_TIME > now; +} + void Node::update(const sockaddr* sa, socklen_t salen) { @@ -343,7 +349,7 @@ Node::update(const sockaddr* sa, socklen_t salen) sslen = salen; } -/** To be called when a message was sent to the noe */ +/** To be called when a message was sent to the node */ void Node::requested(time_point now) { @@ -523,9 +529,7 @@ Dht::trySearchInsert(const std::shared_ptr& node) if (s.af != family) continue; if (s.insertNode(node, now)) { inserted = true; - auto tm = s.getNextStepTime(types, now); - if (tm != TIME_INVALID && (search_time == TIME_INVALID || search_time > tm)) - search_time = tm; + search_time = std::min(search_time, s.getNextStepTime(types, now)); } } return inserted; @@ -553,7 +557,7 @@ Dht::newNode(const InfoHash& id, const sockaddr *sa, socklen_t salen, int confir if (n->id != id) continue; /* Known node. Update stuff. */ - if (confirm || n->time < now - Node::NODE_EXPIRE_TIME) { + if (confirm || n->time + Node::NODE_EXPIRE_TIME < now) { n->update(sa, salen); if (confirm) { n->received(now, confirm >= 2); @@ -674,6 +678,7 @@ Dht::Search::removeExpiredNode(time_point now) e = std::prev(e); const Node& n = *e->node; if (n.isExpired(now) and n.time + Node::NODE_EXPIRE_TIME < now) { + //std::cout << "Removing expired node " << n.id << " from IPv" << (af==AF_INET?'4':'6') << " search " << id << std::endl; nodes.erase(e); return true; } @@ -695,7 +700,7 @@ Dht::Search::insertNode(std::shared_ptr node, time_point now, const Blob& const auto& nid = node->id; // Fast track for the case where the node is not relevant for this search - if (nodes.size() == SEARCH_NODES && id.xorCmp(nid, nodes.back().node->id) > 0 && node->isExpired(now)) + if (nodes.size() >= SEARCH_NODES && id.xorCmp(nid, nodes.back().node->id) > 0 && node->isExpired(now)) return false; // Reset search timer if it was empty @@ -705,7 +710,10 @@ Dht::Search::insertNode(std::shared_ptr node, time_point now, const Blob& } bool found = false; + unsigned num_candidates = 0; auto n = std::find_if(nodes.begin(), nodes.end(), [&](const SearchNode& sn) { + if (sn.candidate) + num_candidates++; if (sn.node == node) { found = true; return true; @@ -713,7 +721,7 @@ Dht::Search::insertNode(std::shared_ptr node, time_point now, const Blob& return id.xorCmp(nid, sn.node->id) < 0; }); if (!found) { - if (nodes.size() == SEARCH_NODES) { + if (nodes.size()-num_candidates >= SEARCH_NODES) { if (node->isExpired(now)) return false; if (n == nodes.end()) { @@ -723,14 +731,27 @@ Dht::Search::insertNode(std::shared_ptr node, time_point now, const Blob& n = nodes.end(); } } + bool synced = isSynced(now); n = nodes.insert(n, SearchNode(node)); - if (nodes.size() > SEARCH_NODES and not removeExpiredNode(now)) - nodes.pop_back(); + node->time = now; + if (synced) { + n->candidate = true; + //std::cout << "Adding candidate node " << node->id << " to IPv" << (af==AF_INET?'4':'6') << " synced search " << id << std::endl; + }/* else { + std::cout << "Adding real node " << node->id << " to IPv" << (af==AF_INET?'4':'6') << " synced search " << id << std::endl; + }*/ + while (nodes.size()-num_candidates > SEARCH_NODES) + if (not removeExpiredNode(now)) + nodes.pop_back(); expired = false; } if (not token.empty()) { n->getStatus.reply_time = now; n->getStatus.request_time = TIME_INVALID; + if (n->candidate) { + n->candidate = false; + //std::cout << "Confirm candidate node " << node->id << " to synced search " << id << std::endl; + } if (token.size() <= 64) n->token = token; expired = false; @@ -784,7 +805,7 @@ Dht::searchSendGetValues(Search& sr, SearchNode* pn, bool update) } DHT_WARN("Sending get_values to %s (%s) for %s (p %d last get %lf)", - print_addr((sockaddr*)&n->node->ss, n->node->sslen).c_str(), + print_addr(n->node->ss, n->node->sslen).c_str(), n->node->id.toString().c_str(), sr.id.toString().c_str(), n->node->pinged, @@ -793,6 +814,9 @@ Dht::searchSendGetValues(Search& sr, SearchNode* pn, bool update) sendGetValues((sockaddr*)&n->node->ss, n->node->sslen, TransId {TransPrefix::GET_VALUES, sr.tid}, sr.id, -1, n->node->reply_time >= now - UDP_REPLY_TIME); n->getStatus.request_time = now; pinged(*n->node); + + //std::cout << "Sending GET to " << n->node->id << " for IPv" << (sr.af==AF_INET?'4':'6') << " search " << sr.id << " (now pinged " << n->node->pinged << ")" << std::endl; + return true; } @@ -802,6 +826,7 @@ void Dht::searchStep(Search& sr) { DHT_WARN("Search IPv%c %s searchStep.", sr.af == AF_INET ? '4' : '6', sr.id.toString().c_str()); + sr.step_time = now; /* Check if the first TARGET_NODES (8) live nodes have replied. */ if (sr.isSynced(now)) { @@ -820,128 +845,123 @@ Dht::searchStep(Search& sr) } if (sr.callbacks.empty() && sr.announce.empty() && sr.listeners.empty()) sr.done = true; - else if (not sr.callbacks.empty()) { - unsigned i = 0; - for (auto& sn : sr.nodes) { - i += searchSendGetValues(sr, &sn, true) ? 1 : 0; - if (i >= 3) - break; - } - } - } else { - // true if this node is part of the target nodes cluter. - bool in = sr.id.xorCmp(myid, sr.nodes.back().node->id) < 0; + } - DHT_DEBUG("searchStep (synced%s).", in ? ", in" : ""); + // true if this node is part of the target nodes cluter. + bool in = sr.id.xorCmp(myid, sr.nodes.back().node->id) < 0; - if (not sr.listeners.empty()) { - unsigned i = 0; - for (auto& n : sr.nodes) { - if (n.node->isExpired(now)) - continue; - if (n.getListenTime() <= now) { - DHT_DEBUG("Sending listen to %s (%s).", print_addr((sockaddr*)&n.node->ss, n.node->sslen).c_str(), n.node->id.toString().c_str()); - sendListen((sockaddr*)&n.node->ss, n.node->sslen, TransId {TransPrefix::LISTEN, sr.tid}, sr.id, n.token, n.node->reply_time >= now - UDP_REPLY_TIME); - n.pending = true; - n.listenStatus.request_time = now; - } - if (++i == LISTEN_NODES) - break; - } - } + DHT_DEBUG("searchStep (synced%s).", in ? ", in" : ""); + DHT_DEBUG("%s", getSearchesLog(sr.af).c_str()); - // Announce requests - for (auto& a : sr.announce) { - if (!a.value) { + if (not sr.listeners.empty()) { + unsigned i = 0, t = 0; + for (auto& n : sr.nodes) { + if (n.node->isExpired(now) or (n.candidate and t >= LISTEN_NODES)) continue; - DHT_ERROR("Trying to announce a null value !"); - } - unsigned i = 0; - auto vid = a.value->id; - const auto& type = getType(a.value->type); - if (in) { - DHT_WARN("Storing local value"); - storageStore(sr.id, a.value); - } - for (auto& n : sr.nodes) { - if (n.node->isExpired(now)) - continue; - auto a_status = n.acked.find(vid); - auto at = n.getAnnounceTime(a_status, type); - if ( at <= now ) { - DHT_DEBUG("Sending announce_value to %s (%s).", print_addr((sockaddr*)&n.node->ss, n.node->sslen).c_str(), n.node->id.toString().c_str()); - sendAnnounceValue((sockaddr*)&n.node->ss, n.node->sslen, - TransId {TransPrefix::ANNOUNCE_VALUES, sr.tid}, sr.id, *a.value, - n.token, n.node->reply_time >= now - UDP_REPLY_TIME); - if (a_status == n.acked.end()) { - n.acked[vid] = { now }; - } else { - a_status->second.request_time = now; - } - // use the "pending" flag so we update the "pinged" - // fields after sending the announce requests for every value to announce - n.pending = true; - } - if (++i == TARGET_NODES) - break; + if (n.getListenTime() <= now) { + DHT_DEBUG("Sending listen to %s (%s).", print_addr(n.node->ss, n.node->sslen).c_str(), n.node->id.toString().c_str()); + sendListen((sockaddr*)&n.node->ss, n.node->sslen, TransId {TransPrefix::LISTEN, sr.tid}, sr.id, n.token, n.node->reply_time >= now - UDP_REPLY_TIME); + n.pending = true; + n.listenStatus.request_time = now; } + t++; + if (not n.candidate and ++i == LISTEN_NODES) + break; + } + } + + // Announce requests + for (auto& a : sr.announce) { + if (!a.value) { + continue; + DHT_ERROR("Trying to announce a null value !"); + } + unsigned i = 0, t = 0; + auto vid = a.value->id; + const auto& type = getType(a.value->type); + if (in) { + DHT_WARN("Storing local value"); + (sr.id, a.value); } for (auto& n : sr.nodes) { - if (n.pending) { - n.pending = false; - pinged(*n.node); + if (n.node->isExpired(now) or (n.candidate and t >= TARGET_NODES)) + continue; + auto a_status = n.acked.find(vid); + auto at = n.getAnnounceTime(a_status, type); + if ( at <= now ) { + DHT_DEBUG("Sending announce_value to %s (%s).", print_addr(n.node->ss, n.node->sslen).c_str(), n.node->id.toString().c_str()); + sendAnnounceValue((sockaddr*)&n.node->ss, n.node->sslen, + TransId {TransPrefix::ANNOUNCE_VALUES, sr.tid}, sr.id, *a.value, + n.token, n.node->reply_time >= now - UDP_REPLY_TIME); + if (a_status == n.acked.end()) { + n.acked[vid] = { now }; + } else { + a_status->second.request_time = now; + } + // use the "pending" flag so we update the "pinged" + // fields after sending the announce requests for every value to announce + n.pending = true; } + t++; + if (not n.candidate and ++i == TARGET_NODES) + break; } - if (sr.announce.empty() && sr.listeners.empty()) - sr.done = true; - DHT_DEBUG("Searchstep done."); } + for (auto& n : sr.nodes) { + if (n.pending) { + n.pending = false; + pinged(*n.node); + } + } + if (sr.callbacks.empty() && sr.announce.empty() && sr.listeners.empty()) + sr.done = true; + DHT_DEBUG("Searchstep done."); - } else { - if (sr.get_step_time + SEARCH_GET_STEP >= now) { - DHT_WARN("searchStep: too early"); - return; - } - DHT_DEBUG("searchStep."); - - unsigned i = 0, e = 0; - for (auto& sn : sr.nodes) { - if (sn.node->isExpired(now)) { - e++; - continue; - } - i += searchSendGetValues(sr, &sn) ? 1 : 0; - if (i >= 3) - break; - } - if (i > 0) - sr.get_step_time = now; - else if (e == sr.nodes.size()) { - sr.expired = true; - if (sr.announce.empty() && sr.listeners.empty()) { - // Listening or announcing requires keeping the cluster up to date. - sr.done = true; - } - { - auto get_cbs = std::move(sr.callbacks); - for (const auto& g : get_cbs) { - if (g.done_cb) - g.done_cb(false, {}); - } - } - { - std::vector a_cbs; - a_cbs.reserve(sr.announce.size()); - for (const auto& a : sr.announce) - if (a.callback) - a_cbs.emplace_back(std::move(a.callback)); - for (const auto& a : a_cbs) - a(false, {}); - } - return; - } } - sr.step_time = now; + + if (sr.get_step_time + SEARCH_GET_STEP >= now) { + DHT_WARN("searchStep: too early"); + return; + } + + unsigned i = 0; + bool sent; + do { + if (sent = searchSendGetValues(sr)) + i++; + } + while (sent and i < 3); + DHT_DEBUG("searchStep, sent %u.", i); + + if (i > 0) + sr.get_step_time = now; + else if (std::count_if(sr.nodes.begin(), sr.nodes.end(), [&](const SearchNode& sn) { + return sn.node->isExpired(now); + }) == sr.nodes.size()) + { + sr.expired = true; + if (sr.announce.empty() && sr.listeners.empty()) { + // Listening or announcing requires keeping the cluster up to date. + sr.done = true; + } + { + auto get_cbs = std::move(sr.callbacks); + for (const auto& g : get_cbs) { + if (g.done_cb) + g.done_cb(false, {}); + } + } + { + std::vector a_cbs; + a_cbs.reserve(sr.announce.size()); + for (const auto& a : sr.announce) + if (a.callback) + a_cbs.emplace_back(std::move(a.callback)); + for (const auto& a : a_cbs) + a(false, {}); + } + return; + } } @@ -985,9 +1005,9 @@ Dht::Search::isSynced(time_point now) const { unsigned i = 0; for (const auto& n : nodes) { - if (n.node->isExpired(now)) + if (n.node->isExpired(now) or n.candidate) continue; - if (!n.isSynced(now)) + if (not n.isSynced(now)) return false; if (++i == TARGET_NODES) break; @@ -1008,9 +1028,9 @@ bool Dht::Search::isDone(const Get& get, time_point now) const { unsigned i = 0; - auto limit = std::max(get.start, now - Node::NODE_EXPIRE_TIME); + const auto limit = std::max(get.start, now - Node::NODE_EXPIRE_TIME); for (const auto& sn : nodes) { - if (sn.node->isExpired(now)) + if (sn.node->isExpired(now) or sn.candidate) continue; if (sn.getStatus.reply_time < limit) return false; @@ -1023,31 +1043,33 @@ Dht::Search::isDone(const Get& get, time_point now) const time_point Dht::Search::getUpdateTime(time_point now) const { - time_point ut = TIME_MAX; + time_point ut = time_point::max(); const auto last_get = getLastGetTime(); - unsigned i = 0, d = 0; + unsigned i = 0, t = 0, d = 0; for (const auto& sn : nodes) { - if (sn.node->isExpired(now)) + if (sn.node->isExpired(now) or (sn.candidate and t >= TARGET_NODES)) continue; - if (sn.getStatus.reply_time < now - Node::NODE_EXPIRE_TIME) { // not isSynced - return std::min(ut, get_step_time + SEARCH_GET_STEP); - } - else if (sn.getStatus.reply_time < last_get) { - ut = std::min(ut, sn.getStatus.request_time + Node::MAX_RESPONSE_TIME); + if (sn.getStatus.reply_time < std::max(now - Node::NODE_EXPIRE_TIME, last_get)) { + // not isSynced + ut = std::min(ut, std::max( + sn.getStatus.request_time + Node::MAX_RESPONSE_TIME, + get_step_time + SEARCH_GET_STEP)); + if (not sn.candidate) + d++; } else { - d++; ut = std::min(ut, std::max( sn.getStatus.request_time + Node::MAX_RESPONSE_TIME, sn.getStatus.reply_time + Node::NODE_EXPIRE_TIME)); } - if (++i == TARGET_NODES) + t++; + if (not sn.candidate and ++i == TARGET_NODES) break; } - if (not callbacks.empty() and i == d) { + if (not callbacks.empty() and d == 0) { // If all synced/updated but some callbacks remain, step now to clear them return now; } - return ut == TIME_MAX ? TIME_INVALID : ut; + return ut; } bool @@ -1057,9 +1079,9 @@ Dht::Search::isAnnounced(Value::Id id, const ValueType& type, time_point now) co return false; unsigned i = 0; for (const auto& n : nodes) { - if (n.node->isExpired(now)) + if (n.candidate or n.node->isExpired(now)) continue; - if (!n.isAnnounced(id, type, now)) + if (not n.isAnnounced(id, type, now)) return false; if (++i == TARGET_NODES) break; @@ -1074,7 +1096,7 @@ Dht::Search::isListening(time_point now) const return false; unsigned i = 0; for (const auto& n : nodes) { - if (n.node->isExpired(now)) + if (n.candidate or n.node->isExpired(now)) continue; if (!n.isListening(now)) return false; @@ -1088,20 +1110,19 @@ time_point Dht::Search::getAnnounceTime(const std::map& types, time_point now) const { if (nodes.empty()) - return {}; - time_point ret {}; + return time_point::max(); + time_point ret {time_point::max()}; for (const auto& a : announce) { if (!a.value) continue; auto type_it = types.find(a.value->type); const ValueType& type = (type_it == types.end()) ? ValueType::USER_DATA : type_it->second; - unsigned i = 0; + unsigned i = 0, t = 0; for (const auto& n : nodes) { - if (n.node->isExpired(now)) + if (n.node->isExpired(now) or (n.candidate and t >= TARGET_NODES)) continue; - auto at = n.getAnnounceTime(a.value->id, type); - if (at != TIME_INVALID && (ret == TIME_INVALID || ret > at)) - ret = at; - if (++i == TARGET_NODES) + ret = std::min(ret, n.getAnnounceTime(a.value->id, type)); + t++; + if (not n.candidate and ++i == TARGET_NODES) break; } } @@ -1112,15 +1133,16 @@ time_point Dht::Search::getListenTime(time_point now) const { if (listeners.empty()) - return {}; - time_point listen_time {}; - unsigned i = 0; + return time_point::max(); + time_point listen_time {time_point::max()}; + unsigned i = 0, t = 0; for (const auto& sn : nodes) { - if (sn.node->isExpired(now)) + if (sn.node->isExpired(now) or (sn.candidate and t >= LISTEN_NODES)) continue; auto lt = sn.getListenTime(); - listen_time = listen_time == TIME_INVALID ? lt : std::min(listen_time, lt); - if (++i == LISTEN_NODES) + listen_time = std::min(listen_time, lt); + t++; + if (not sn.candidate and ++i == LISTEN_NODES) break; } return listen_time; @@ -1129,33 +1151,33 @@ Dht::Search::getListenTime(time_point now) const time_point Dht::Search::getNextStepTime(const std::map& types, time_point now) const { + auto next_step = time_point::max(); if (expired or done) - return {}; - - if (!isSynced(now)) - return get_step_time + SEARCH_GET_STEP; - - auto next_step = TIME_MAX; + return next_step; auto ut = getUpdateTime(now); - if (ut != TIME_INVALID) { + if (ut != time_point::max()) { //std::cout << id.toString() << " IPv" << (af==AF_INET?"4":"6") << " update time in " << print_dt(ut - now) << " s" << std::endl; next_step = std::min(next_step, ut); } - auto at = getAnnounceTime(types, now); - if (at != TIME_INVALID) { - //std::cout << id.toString() << " IPv" << (af==AF_INET?"4":"6") << " announce time in " << print_dt(at - now) << " s" << std::endl; - next_step = std::min(next_step, at); + if (isSynced(now)) + { + auto at = getAnnounceTime(types, now); + if (at != time_point::max()) { + //std::cout << id.toString() << " IPv" << (af==AF_INET?"4":"6") << " announce time in " << print_dt(at - now) << " s" << std::endl; + next_step = std::min(next_step, at); + } + + auto lt = getListenTime(now); + if (lt != time_point::max()) { + //std::cout << id.toString() << " IPv" << (af==AF_INET?"4":"6") << " listen time in " << print_dt(lt - now) << " s" << std::endl; + next_step = std::min(next_step, lt); + } } - auto lt = getListenTime(now); - if (lt != TIME_INVALID) { - //std::cout << id.toString() << " IPv" << (af==AF_INET?"4":"6") << " listen time in " << print_dt(lt - now) << " s" << std::endl; - next_step = std::min(next_step, lt); - } - return next_step == TIME_MAX ? TIME_INVALID : next_step; + return next_step; } void @@ -1259,7 +1281,7 @@ Dht::announce(const InfoHash& id, sa_family_t af, const std::shared_ptr& a_sr->callback = callback; } auto tm = sr->getNextStepTime(types, now); - if (tm != TIME_INVALID && (search_time == TIME_INVALID || search_time > tm)) { + if (tm < search_time) { DHT_DEBUG("search_time is now in %lf", print_dt(tm-clock::now())); search_time = tm; } else { @@ -1285,9 +1307,7 @@ Dht::listenTo(const InfoHash& id, sa_family_t af, GetCallback cb, Value::Filter sr->done = false; auto token = ++sr->listener_token; sr->listeners.emplace(token, LocalListener{f, cb}); - auto tm = sr->getNextStepTime(types, now); - if (tm != TIME_INVALID && (search_time == TIME_INVALID || search_time > tm)) - search_time = tm; + search_time = std::min(search_time, sr->getNextStepTime(types, now)); return token; } @@ -1802,7 +1822,7 @@ void Dht::dumpSearch(const Search& sr, std::ostream& out) const { using namespace std::chrono; - out << std::endl << "Search (IPv" << (sr.af == AF_INET6 ? "6" : "4") << ") " << sr.id; + 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; if (sr.done) out << " [done]"; @@ -1816,28 +1836,65 @@ Dht::dumpSearch(const Search& sr, std::ostream& out) const for (const auto& n : sr.announce) { bool announced = sr.isAnnounced(n.value->id, getType(n.value->type), now); - out << " Announcement: " << *n.value << (announced ? " [announced]" : "") << std::endl; + out << "Announcement: " << *n.value << (announced ? " [announced]" : "") << std::endl; } + out << " Common bits InfoHash Conn. Get Put IP" << std::endl; unsigned i = 0; + auto last_get = sr.getLastGetTime(); for (const auto& n : sr.nodes) { - out << " Node " << i++ << " id " << n.node->id << " bits " << InfoHash::commonBits(sr.id, n.node->id); - if (n.node->pinged) { - out << " [c " << n.node->pinged << " " - << duration_cast(now - n.node->pinged_time).count() << "s " - << duration_cast(now - n.node->reply_time).count() << "s]"; + out << std::setfill (' ') << std::setw(3) << InfoHash::commonBits(sr.id, n.node->id) << ' ' << n.node->id; + out << ' ' << (findNode(n.node->id, AF_INET) || findNode(n.node->id, AF_INET6) ? '*' : ' '); + out << ' ' << (n.candidate ? 'c' : ' '); + out << " [" + << (n.node->isMessagePending(now) ? 'f':' '); + if (n.node->pinged) + out << n.node->pinged; + else + out << ' '; + out << (n.node->isExpired(now) ? 'x' : ' ') << "]"; + + out << " [" + << (n.getStatus.pending(now) ? 'f' : (n.getStatus.expired(now) ? 'x' : ' ')) + << (n.isSynced(now) ? 's' : '-') + << ((n.getStatus.reply_time > last_get) ? 'u' : '-') << "] "; + + if (not sr.listeners.empty()) { + if (n.listenStatus.request_time == time_point{}) + out << " "; + else + out << "[" + << (n.listenStatus.pending(now) ? 'f' : (n.listenStatus.expired(now) ? 'x' : ' ')) + << (n.isListening(now) ? 'l' : '-') << "] "; } - out << " [g "; - if (n.getStatus.request_time != TIME_INVALID) - out << duration_cast(now - n.getStatus.request_time).count() << "s "; - out << duration_cast(now - n.getStatus.reply_time).count() << "s]"; - if (findNode(n.node->id, AF_INET) || findNode(n.node->id, AF_INET6)) - out << " [known]"; - if (n.node->reply_time != TIME_INVALID) - out << " [replied]"; - if (n.listenStatus.reply_time != TIME_INVALID) - out << " [list]"; - out << (n.isSynced(now) ? " [synced]" : (n.node->isExpired(now) ? " [expired]" : " [not synced]")); + + if (not sr.announce.empty()) { + if (n.acked.empty()) { + out << " "; + for (const auto& a : sr.announce) + out << ' '; + } else { + out << "["; + for (const auto& a : sr.announce) { + auto ack = n.acked.find(a.value->id); + if (ack == n.acked.end()) { + out << ' '; + } else { + auto& astatus = ack->second; + if (astatus.reply_time + getType(a.value->type).expiration > now) + out << 'a'; + else if (astatus.pending(now)) + out << 'f'; + else if (astatus.expired(now)) + out << 'x'; + else + out << ' '; + } + } + out << "] "; + } + } + out << print_addr(n.node->ss, n.node->sslen); out << std::endl; } } @@ -1899,11 +1956,13 @@ Dht::getRoutingTablesLog(sa_family_t af) const } std::string -Dht::getSearchesLog(sa_family_t) const +Dht::getSearchesLog(sa_family_t af) const { std::stringstream out; + out << "s:synched, u:updated, a:announced, c:candidate, f:cur req, x:expired, *:known" << std::endl; for (const auto& sr : searches) - dumpSearch(sr, out); + if (af == 0 or sr.af == af) + dumpSearch(sr, out); return out.str(); } @@ -2120,6 +2179,8 @@ Dht::processMessage(const uint8_t *buf, size_t buflen, const sockaddr *from, soc } } + //std::cout << "Message from " << id << " IPv" << (from->sa_family==AF_INET?'4':'6') << std::endl; + switch (message) { case MessageType::Error: if (tid.length != 4) return; @@ -2208,14 +2269,9 @@ Dht::processMessage(const uint8_t *buf, size_t buflen, const sockaddr *from, soc /* Since we received a reply, the number of requests in flight has decreased. Let's push another request. */ - /*if (sr->isSynced(now)) { - DHT_DEBUG("Trying to accelerate search!"); - search_time = now; - //sr->step_time = 0; - } else {*/ - if (searchSendGetValues(*sr)) - sr->get_step_time = now; - //} + //std::cout << "Received reply from " << id << ", sending new message..." << std::endl; + if (searchSendGetValues(*sr)) + sr->get_step_time = now; } } if (sr) { @@ -2416,19 +2472,17 @@ Dht::periodic(const uint8_t *buf, size_t buflen, expireSearches(); } - if (search_time != TIME_INVALID && now > search_time) { - search_time = TIME_INVALID; + if (now > search_time) { + search_time = time_point::max(); for (auto& sr : searches) { - auto tm = sr.getNextStepTime(types, now); - if (tm == TIME_INVALID) continue; - if (tm <= now) { + auto step = sr.getNextStepTime(types, now); + if (step <= now) { searchStep(sr); - tm = sr.getNextStepTime(types, now); + step = sr.getNextStepTime(types, now); } - if (tm != TIME_INVALID && (search_time == TIME_INVALID || search_time > tm)) - search_time = tm; + search_time = std::min(search_time, step); } - if (search_time == TIME_INVALID) + if (search_time == time_point::max()) DHT_DEBUG("next search time : (none)"); else DHT_DEBUG("next search time : %lf s%s", print_dt(search_time-now), (search_time < now)?" (ASAP)":""); @@ -2459,11 +2513,7 @@ Dht::periodic(const uint8_t *buf, size_t buflen, dumpTables(); } - time_point tosleep = confirm_nodes_time; - if (search_time != TIME_INVALID) - tosleep = std::min(tosleep, search_time); - - return tosleep; + return std::min(confirm_nodes_time, search_time); } std::vector