diff --git a/include/opendht/dht.h b/include/opendht/dht.h index 2f10a0ce5eadae956e8b28c86dc0ac71a497f6cb..ebbdc56515b5e426aaed851da69e4eebbbdc271b 100644 --- a/include/opendht/dht.h +++ b/include/opendht/dht.h @@ -609,6 +609,7 @@ private: bool expired {false}; /* no node, or all nodes expired */ bool done {false}; /* search is over, cached for later */ + bool refilled {false}; std::vector<SearchNode> nodes {}; std::vector<Announce> announce {}; std::vector<Get> callbacks {}; diff --git a/python/tools/benchmark.py b/python/tools/benchmark.py index f671bb257caadf46337576fb3502642790c6a073..5bdfc321df9656d9e23c3313aa6098be8d6ae050 100755 --- a/python/tools/benchmark.py +++ b/python/tools/benchmark.py @@ -134,7 +134,7 @@ class DhtNetworkSubProcess(NSPopen): SHUTDOWN_CLUSTER_REQ = b"sdc" DUMP_STORAGE_REQ = b"strl" MESSAGE_STATS = b"gms" - + # tokens NOTIFY_TOKEN = 'notify' @@ -515,13 +515,13 @@ class PersistenceTest(FeatureTest): DhtNetwork.log("Waiting 15 seconds for packets to work their way effectively.") time.sleep(15) ops_count.append(cluster_ops_count/self.wb.node_num) - + # checking if values were transfered to new nodes foreign_nodes_before_delete = PersistenceTest.foreign_nodes DhtNetwork.log('[GET]: trying to fetch persistent values') self._dhtGet(consumer, myhash) new_nodes = set(PersistenceTest.foreign_nodes) - set(foreign_nodes_before_delete) - + self._result(local_values, new_nodes) if self._plot: @@ -602,7 +602,7 @@ class PersistenceTest(FeatureTest): nodes = set([]) # prevents garbage collecting of unused flood nodes during the test. - flood_nodes = [] + flood_nodes = [] def gottaGetThemAllPokeNodes(nodes=None): nonlocal consumer, hashes diff --git a/src/dht.cpp b/src/dht.cpp index 0b16eafee2b5775094cdadd8664bbbc64e8ca981..1efc9bef361717aa8695f583e9189d32128361c3 100644 --- a/src/dht.cpp +++ b/src/dht.cpp @@ -1048,13 +1048,19 @@ Dht::searchStep(Search& sr) return sn.candidate or sn.node->isExpired(now); }) == sr.nodes.size()) { - unsigned added = sr.refill(sr.af == AF_INET ? buckets : buckets6, now); + unsigned added = 0; + if (not sr.refilled) { + added = sr.refill(sr.af == AF_INET ? buckets : buckets6, now); + sr.refilled = true; + } if (added) { DHT_WARN("[search %s IPv%c] refilled with %u nodes", sr.id.toString().c_str(), (sr.af == AF_INET) ? '4' : '6', added); } else { DHT_ERROR("[search %s IPv%c] expired", sr.id.toString().c_str(), sr.af == AF_INET ? '4' : '6'); // no nodes or all expired nodes sr.expired = true; + // reset refilled since the search is now expired. + sr.refilled = false; if (sr.announce.empty() && sr.listeners.empty()) { // Listening or announcing requires keeping the cluster up to date. sr.done = true; @@ -2622,10 +2628,9 @@ Dht::processMessage(const uint8_t *buf, size_t buflen, const sockaddr *from, soc } { // We store a value only if we think we're part of the - // 2*TARGET_NODES nodes around the target id. - auto closest_nodes = (from->sa_family == AF_INET ? buckets : buckets6).findClosestNodes(msg.info_hash, 2*TARGET_NODES); + // SEARCH_NODES nodes around the target id. + auto closest_nodes = (from->sa_family == AF_INET ? buckets : buckets6).findClosestNodes(msg.info_hash, SEARCH_NODES); if (msg.info_hash.xorCmp(closest_nodes.back()->id, myid) < 0) { - std::cerr << "my id [" << myid << "] is too far from " << msg.info_hash << ". Dropping value." << std::endl; DHT_WARN("[node %s %s] announce too far from the target id. Dropping value.", msg.id.toString().c_str(), print_addr(from, fromlen).c_str()); for (auto& v : msg.values) {