search: expire only after candidate expiration

README.md

@@ -27,7 +27,7 @@ Build instructions: <https://github.com/savoirfairelinux/opendht/wiki/Build-the-
 
 #### How-to build a simple client app
 ```bash
-g++ main.cpp -std=c++14 -lopendht -lgnutls
+g++ main.cpp -std=c++14 -lopendht
 ```
 
 ## Examples
@@ -51,7 +51,7 @@ int main()
 
     // Join the network through any running node,
     // here using a known bootstrap node.
-    node.bootstrap("bootstrap.ring.cx", "4222");
+    node.bootstrap("bootstrap.jami.net", "4222");
 
     // put some data on the dht
     std::vector<uint8_t> some_data(5, 10);
@@ -82,7 +82,7 @@ node.run()
 
 # Join the network through any running node,
 # here using a known bootstrap node.
-node.bootstrap("bootstrap.ring.cx", "4222")
+node.bootstrap("bootstrap.jami.net", "4222")
 
 # blocking call (provide callback arguments to make the call non-blocking)
 node.put(dht.InfoHash.get("unique_key"), dht.Value(b'some binary data'))

configure.ac

@@ -140,6 +140,7 @@ AS_IF([test "x$with_jsoncpp" != "xno"],
 AS_IF([test "x$have_jsoncpp" = "xyes"], [
     AC_MSG_NOTICE([Using JsonCpp])
     CPPFLAGS+=" -DOPENDHT_JSONCPP"
+    AC_SUBST(jsoncpp_lib, [", jsoncpp"])
 ], [
     AC_MSG_NOTICE([Not using JsonCpp])
     AM_COND_IF(PROXY_CLIENT_OR_SERVER, AC_MSG_ERROR(["JsonCpp is required for proxy/push notification support"]))
@@ -151,6 +152,7 @@ AS_IF([test "x$with_openssl" != "xno"],
       [have_openssl=no])
 AS_IF([test "x$have_openssl" = "xyes"], [
     AC_MSG_NOTICE([Using OpenSSL])
+    AC_SUBST(openssl_lib, [", openssl"])
 ], [
     AC_MSG_NOTICE([Not using OpenSSL])
 ])
@@ -175,6 +177,7 @@ AM_COND_IF([PROXY_CLIENT_OR_SERVER], [
     # http_parser has no pkgconfig, instead we check with:
     AC_CHECK_LIB(http_parser, exit,, AC_MSG_ERROR([Missing HttpParser library files]))
     AC_CHECK_HEADERS([http_parser.h], [http_parser_headers=yes; break;])
+    AC_SUBST(http_parser_lib, ["-lhttp_parser"])
     AS_IF([test "x$http_parser_headers" != "xyes"], AC_MSG_ERROR([Missing HttpParser headers files]))
 ])
 

include/opendht/callbacks.h

@@ -58,8 +58,8 @@ struct OPENDHT_PUBLIC NodeStats {
              cached_nodes {0},
              incoming_nodes {0};
     unsigned table_depth {0};
-    unsigned cachedNodes {0};
     unsigned searches {0};
+    unsigned node_cache_size {0};
     unsigned getKnownNodes() const { return good_nodes + dubious_nodes; }
     unsigned long getNetworkSizeEstimation() const { return 8 * std::exp2(table_depth); }
     std::string toString() const;
@@ -73,25 +73,7 @@ struct OPENDHT_PUBLIC NodeStats {
     explicit NodeStats(const Json::Value& v);
 #endif
 
-    MSGPACK_DEFINE_MAP(good_nodes, dubious_nodes, cached_nodes, incoming_nodes, table_depth, searches)
-};
-
-struct OPENDHT_PUBLIC NodeMetrics {
-    size_t scheduler_size {0};
-    size_t node_cache_size {0};
-    size_t request_count {0};
-    size_t rate_limiter_size {0};
-    size_t reported_addr_num {0};
-    size_t store_size {0};
-    size_t store_values {0};
-    size_t store_quota_size {0};
-
-#ifdef OPENDHT_JSONCPP
-    /**
-     * Build a json object from a NodeStats
-     */
-    Json::Value toJson() const;
-#endif
+    MSGPACK_DEFINE_MAP(good_nodes, dubious_nodes, cached_nodes, incoming_nodes, table_depth, searches, node_cache_size)
 };
 
 struct OPENDHT_PUBLIC NodeInfo {
@@ -100,7 +82,6 @@ struct OPENDHT_PUBLIC NodeInfo {
     NodeStats ipv4 {};
     NodeStats ipv6 {};
     size_t ongoing_ops {0};
-    NodeMetrics metrics {};
 
 #ifdef OPENDHT_JSONCPP
     /**
@@ -137,12 +118,18 @@ struct OPENDHT_PUBLIC Config {
     /** If set, the dht will load its state from this file on start and save its state in this file on shutdown */
     std::string persist_path {};
 
-    /** If non-0, overrides the default global rate-limit.-1 means no limit. */
+    /** If non-0, overrides the default global rate-limit. -1 means no limit. */
     ssize_t max_req_per_sec {0};
 
     /** If non-0, overrides the default per-IP address rate-limit. -1 means no limit. */
     ssize_t max_peer_req_per_sec {0};
 
+    /* If non-0, overrides the default maximum number of searches. -1 means no limit.  */
+    ssize_t max_searches {0};
+
+    /* If non-0, overrides the default maximum store size. -1 means no limit.  */
+    ssize_t max_store_size {0};
+
     /** 
      * Use appropriate bahavior for a public IP, stable node:
      *   - No connectivity change triggered when a search fails

include/opendht/dht.h

@@ -80,10 +80,14 @@ public:
      */
     inline const InfoHash& getNodeId() const override { return myid; }
 
+    NodeStatus updateStatus(sa_family_t af) override;
+
     /**
      * Get the current status of the node for the given family.
      */
-    NodeStatus getStatus(sa_family_t af) const override;
+    NodeStatus getStatus(sa_family_t af) const override {
+        return dht(af).status;
+    }
 
     NodeStatus getStatus() const override {
         return std::max(getStatus(AF_INET), getStatus(AF_INET6));
@@ -111,6 +115,15 @@ public:
         return types.getType(type_id);
     }
 
+    void addBootstrap(const std::string& host, const std::string& service) override {
+        bootstrap_nodes.emplace_back(host, service);
+        onDisconnected();
+    }
+
+    void clearBootstrap() override {
+        bootstrap_nodes.clear();
+    }
+
     /**
      * Insert a node in the main routing table.
      * The node is not pinged, so this should be
@@ -273,7 +286,6 @@ public:
     void loadState(const std::string& path);
 
     NodeStats getNodesStats(sa_family_t af) const override;
-    NodeMetrics getNodeMetrics() const override;
 
     std::string getStorageLog() const override;
     std::string getStorageLog(const InfoHash&) const override;
@@ -331,10 +343,10 @@ private:
     static constexpr unsigned LISTEN_NODES {4};
 
     /* The maximum number of hashes we're willing to track. */
-    static constexpr unsigned MAX_HASHES {64 * 1024};
+    static constexpr unsigned MAX_HASHES {1024 * 1024};
 
     /* The maximum number of searches we keep data about. */
-    static constexpr unsigned MAX_SEARCHES {64 * 1024};
+    static constexpr unsigned MAX_SEARCHES {1024 * 1024};
 
     static constexpr std::chrono::minutes MAX_STORAGE_MAINTENANCE_EXPIRE_TIME {10};
 
@@ -367,20 +379,33 @@ private:
     // registred types
     TypeStore types;
 
-    // the stuff
-    RoutingTable buckets4 {};
-    RoutingTable buckets6 {};
+    using SearchMap = std::map<InfoHash, Sp<Search>>;
+    struct Kad {
+        RoutingTable buckets {};
+        SearchMap searches {};
+        unsigned pending_pings {0};
+        NodeStatus status;
+
+        NodeStatus getStatus(time_point now) const;
+        NodeStats getNodesStats(time_point now, const InfoHash& myid) const;
+    };
+
+    Kad dht4 {};
+    Kad dht6 {};
+
+    std::vector<std::pair<std::string,std::string>> bootstrap_nodes {};
+    std::chrono::steady_clock::duration bootstrap_period {std::chrono::seconds(10)};
+    Sp<Scheduler::Job> bootstrapJob {};
 
     std::map<InfoHash, Storage> store;
     std::map<SockAddr, StorageBucket, SockAddr::ipCmp> store_quota;
     size_t total_values {0};
     size_t total_store_size {0};
+    size_t max_store_keys {MAX_HASHES};
     size_t max_store_size {DEFAULT_STORAGE_LIMIT};
 
-    using SearchMap = std::map<InfoHash, Sp<Search>>;
-    SearchMap searches4 {};
-    SearchMap searches6 {};
-    uint16_t search_id {0};
+    size_t max_searches {MAX_SEARCHES};
+    size_t search_id {0};
 
     // map a global listen token to IPv4, IPv6 specific listen tokens.
     // 0 is the invalid token.
@@ -395,9 +420,6 @@ private:
     Sp<Scheduler::Job> nextStorageMaintenance {};
 
     net::NetworkEngine network_engine;
-    unsigned pending_pings4 {0};
-    unsigned pending_pings6 {0};
-
     using ReportedAddr = std::pair<unsigned, SockAddr>;
     std::vector<ReportedAddr> reported_addr;
 
@@ -443,8 +465,10 @@ private:
     size_t maintainStorage(decltype(store)::value_type&, bool force=false, const DoneCallback& donecb={});
 
     // Buckets
-    RoutingTable& buckets(sa_family_t af) { return af == AF_INET ? buckets4 : buckets6; }
-    const RoutingTable& buckets(sa_family_t af) const { return af == AF_INET ? buckets4 : buckets6; }
+    Kad& dht(sa_family_t af) { return af == AF_INET ? dht4 : dht6; }
+    const Kad& dht(sa_family_t af) const { return af == AF_INET ? dht4 : dht6; }
+    RoutingTable& buckets(sa_family_t af) { return dht(af).buckets; }
+    const RoutingTable& buckets(sa_family_t af) const { return dht(af).buckets; }
     Bucket* findBucket(const InfoHash& id, sa_family_t af) {
         auto& b = buckets(af);
         auto it = b.findBucket(id);
@@ -466,8 +490,8 @@ private:
 
     // Searches
 
-    inline SearchMap& searches(sa_family_t af) { return af == AF_INET ? searches4 : searches6; }
-    inline const SearchMap& searches(sa_family_t af) const { return af == AF_INET ? searches4 : searches6; }
+    inline SearchMap& searches(sa_family_t af) { return dht(af).searches; }
+    inline const SearchMap& searches(sa_family_t af) const { return dht(af).searches; }
 
     /**
      * Low-level method that will perform a search on the DHT for the specified
@@ -490,6 +514,7 @@ private:
 
     void confirmNodes();
     void expire();
+    void onDisconnected();
 
     /**
      * Generic function to execute when a 'get' request has completed.

include/opendht/dht_interface.h

@@ -42,6 +42,7 @@ public:
     /**
      * Get the current status of the node for the given family.
      */
+    virtual NodeStatus updateStatus(sa_family_t af) { return getStatus(af); };
     virtual NodeStatus getStatus(sa_family_t af) const = 0;
     virtual NodeStatus getStatus() const = 0;
 
@@ -69,6 +70,9 @@ public:
 
     virtual const ValueType& getType(ValueType::Id type_id) const = 0;
 
+    virtual void addBootstrap(const std::string& /*host*/, const std::string& /*service*/) {};
+    virtual void clearBootstrap() {};
+
     /**
      * Insert a node in the main routing table.
      * The node is not pinged, so this should be
@@ -194,7 +198,6 @@ public:
     virtual void importValues(const std::vector<ValuesExport>&) = 0;
 
     virtual NodeStats getNodesStats(sa_family_t af) const = 0;
-    virtual NodeMetrics getNodeMetrics() const { return {}; }
 
     virtual std::string getStorageLog() const = 0;
     virtual std::string getStorageLog(const InfoHash&) const = 0;

include/opendht/dhtrunner.h

@@ -459,14 +459,6 @@ private:
         Stopping
     };
 
-    /**
-     * Will try to resolve the list of hostnames `bootstrap_nodes` on seperate
-     * thread and then queue ping requests. This list should contain reliable
-     * nodes so that the DHT node can recover quickly from losing connection
-     * with the network.
-     */
-    void tryBootstrapContinuously();
-
     time_point loop_();
 
     NodeStatus getStatus() const {
@@ -513,15 +505,6 @@ private:
     net::PacketList rcv {};
     decltype(rcv) rcv_free {};
 
-    /** true if currently actively boostraping */
-    std::atomic_bool bootstraping {false};
-    /* bootstrap nodes given as (host, service) pairs */
-    std::vector<std::pair<std::string,std::string>> bootstrap_nodes_all {};
-    std::vector<std::pair<std::string,std::string>> bootstrap_nodes {};
-    std::thread bootstrap_thread {};
-    std::mutex bootstrap_mtx {};
-    std::condition_variable bootstrap_cv {};
-
     std::queue<std::function<void(SecureDht&)>> pending_ops_prio {};
     std::queue<std::function<void(SecureDht&)>> pending_ops {};
     std::mutex storage_mtx {};

include/opendht/network_engine.h

@@ -276,7 +276,7 @@ public:
      */
     Sp<Request>
     sendPing(Sp<Node> n, RequestCb&& on_done, RequestExpiredCb&& on_expired);
-    
+
     /**
      * Send a "ping" request to a given node.
      *
@@ -293,7 +293,7 @@ public:
                 std::forward<RequestCb>(on_done),
                 std::forward<RequestExpiredCb>(on_expired));
     }
-    
+
     /**
      * Send a "find node" request to a given node.
      *

include/opendht/network_utils.h

@@ -61,6 +61,9 @@ using PacketList = std::list<ReceivedPacket>;
 
 class OPENDHT_PUBLIC DatagramSocket {
 public:
+    /** A function that takes a list of new received packets and
+     *  optionally returns consumed packets for recycling.
+     **/
     using OnReceive = std::function<PacketList(PacketList&& packets)>;
     virtual ~DatagramSocket() {};
 
@@ -85,6 +88,11 @@ public:
 
     virtual const SockAddr& getBoundRef(sa_family_t family = AF_UNSPEC) const = 0;
 
+    /** Virtual resolver mothod allows to implement custom resolver */
+    virtual std::vector<SockAddr> resolve(const std::string& host, const std::string& service = {}) {
+        return SockAddr::resolve(host, service);
+    }
+
     virtual void stop() = 0;
 protected:
 

include/opendht/scheduler.h

@@ -115,8 +115,6 @@ public:
     inline time_point syncTime() { return (now = clock::now()); }
     inline void syncTime(const time_point& n) { now = n; }
 
-    inline size_t size() const { return timers.size(); }
-
 private:
     time_point now {clock::now()};
     std::multimap<time_point, Sp<Job>> timers {}; /* the jobs ordered by time */

include/opendht/securedht.h

@@ -180,9 +180,6 @@ public:
     NodeStats getNodesStats(sa_family_t af) const override {
         return dht_->getNodesStats(af);
     }
-    NodeMetrics getNodeMetrics() const override {
-        return dht_->getNodeMetrics();
-    }
     std::vector<unsigned> getNodeMessageStats(bool in = false) override {
         return dht_->getNodeMessageStats(in);
     }
@@ -204,6 +201,9 @@ public:
     time_point periodic(const uint8_t *buf, size_t buflen, const sockaddr* from, socklen_t fromlen, const time_point& now) override {
         return dht_->periodic(buf, buflen, from, fromlen, now);
     }
+    NodeStatus updateStatus(sa_family_t af) override  {
+        return dht_->updateStatus(af);
+    }
     NodeStatus getStatus(sa_family_t af) const override {
         return dht_->getStatus(af);
     }
@@ -219,6 +219,12 @@ public:
     const ValueType& getType(ValueType::Id type_id) const override {
         return dht_->getType(type_id);
     }
+    void addBootstrap(const std::string& host, const std::string& service) override {
+        dht_->addBootstrap(host, service);
+    }
+    void clearBootstrap() override {
+        dht_->clearBootstrap();
+    }
     void insertNode(const InfoHash& id, const SockAddr& sa) override {
         dht_->insertNode(id, sa);
     }

include/opendht/utils.h

@@ -105,8 +105,7 @@ static std::string
 print_duration(DT d) {
     if (d < std::chrono::seconds(0)) {
         return "-" + print_duration(-d);
-    }
-    if (d < std::chrono::milliseconds(1)) {
+    } else if (d < std::chrono::milliseconds(1)) {
         return to_str(std::chrono::duration_cast<std::chrono::duration<double, std::micro>>(d).count()) +  " us";
     } else if (d < std::chrono::seconds(1)) {
         return to_str(std::chrono::duration_cast<std::chrono::duration<double, std::milli>>(d).count()) +  " ms";

opendht.pc.in

@@ -3,9 +3,9 @@ exec_prefix=@exec_prefix@
 libdir=@libdir@
 includedir=@includedir@
 Name: OpenDHT
-Description: C++ Distributed Hash Table library
+Description: C++14 Distributed Hash Table library
 Version: @VERSION@
 Libs: -L${libdir} -lopendht
-Libs.private: -lpthread
-Requires.private: gnutls >= 3.1@argon2_lib@
+Libs.private: @http_parser_lib@ -pthread
+Requires.private: gnutls >= 3.3, nettle >= 2.4@argon2_lib@@jsoncpp_lib@@openssl_lib@
 Cflags: -I${includedir}

src/callbacks.cpp

@@ -94,6 +94,7 @@ NodeStats::toString() const
 {
     std::stringstream ss;
     ss << "Known nodes: " << good_nodes << " good, " << dubious_nodes << " dubious, " << incoming_nodes << " incoming." << std::endl;
+    ss << searches << " searches, " << node_cache_size << " total cached nodes" << std::endl;
     if (table_depth > 1) {
         ss << "Routing table depth: " << table_depth << std::endl;
         ss << "Network size estimation: " << getNetworkSizeEstimation() << " nodes" << std::endl;
@@ -144,7 +145,6 @@ NodeInfo::toJson() const
     val["ipv4"] = ipv4.toJson();
     val["ipv6"] = ipv6.toJson();
     val["ops"] = Json::Value::LargestUInt(ongoing_ops);
-    val["metrics"] = metrics.toJson();
     return val;
 }
 
@@ -158,21 +158,6 @@ NodeInfo::NodeInfo(const Json::Value& v)
     ongoing_ops = v["ops"].asLargestUInt();
 }
 
-Json::Value
-NodeMetrics::toJson() const
-{
-    Json::Value val;
-    val["scheduler_size"] = Json::Value::LargestUInt(scheduler_size);
-    val["node_cache_size"] = Json::Value::LargestUInt(node_cache_size);
-    val["request_count"] = Json::Value::LargestUInt(request_count);
-    val["rate_limiter_size"] = Json::Value::LargestUInt(rate_limiter_size);
-    val["reported_addr_num"] = Json::Value::LargestUInt(reported_addr_num);
-    val["store_size"] = Json::Value::LargestUInt(store_size);
-    val["store_values"] = Json::Value::LargestUInt(store_values);
-    val["store_quota_size"] = Json::Value::LargestUInt(store_quota_size);
-    return val;
-}
-
 #endif
 
 

src/dht.cpp

@@ -17,7 +17,6 @@
  *  along with this program. If not, see <https://www.gnu.org/licenses/>.
  */
 
-
 #include "dht.h"
 #include "rng.h"
 #include "search.h"
@@ -43,13 +42,41 @@ constexpr duration Dht::REANNOUNCE_MARGIN;
 static constexpr size_t MAX_REQUESTS_PER_SEC {8 * 1024};
 
 NodeStatus
-Dht::getStatus(sa_family_t af) const
+Dht::updateStatus(sa_family_t af)
 {
-    const auto& stats = getNodesStats(af);
-    if (stats.good_nodes)
-        return NodeStatus::Connected;
-    auto& ping = af == AF_INET ? pending_pings4 : pending_pings6;
-    if (ping or stats.getKnownNodes())
+    auto& d = dht(af);
+    auto old = d.status;
+    d.status = d.getStatus(scheduler.time());
+    if (d.status != old) {
+        auto& other = dht(af == AF_INET ? AF_INET6 : AF_INET);
+        if (other.status == NodeStatus::Disconnected && d.status == NodeStatus::Disconnected)
+            onDisconnected();
+        else if (other.status == NodeStatus::Connected || d.status == NodeStatus::Connected) {
+            // On connected
+            if (bootstrapJob) {
+                bootstrapJob->cancel();
+                bootstrapJob.reset();
+            }
+            bootstrap_period = std::chrono::seconds(10);
+        }
+    }
+    return d.status;
+}
+
+NodeStatus
+Dht::Kad::getStatus(time_point now) const
+{
+    unsigned dubious = 0;
+    for (const auto& b : buckets) {
+        for (auto& n : b.nodes) {
+            if (n->isGood(now)) {
+                return NodeStatus::Connected;
+            } else if (not n->isExpired())
+                dubious++;
+        }
+    }
+    auto& ping = pending_pings;
+    if (dubious or ping)
         return NodeStatus::Connecting;
     return NodeStatus::Disconnected;
 }
@@ -218,8 +245,8 @@ Dht::expireSearches()
             return b;
         } else { return false; }
     };
-    erase_if(searches4, expired);
-    erase_if(searches6, expired);
+    erase_if(dht4.searches, expired);
+    erase_if(dht6.searches, expired);
 }
 
 void
@@ -690,8 +717,8 @@ Dht::searchStep(Sp<Search> sr)
 
     while (sr->currentlySolicitedNodeCount() < MAX_REQUESTED_SEARCH_NODES and searchSendGetValues(sr));
 
-    if (sr->getNumberOfConsecutiveBadNodes() >= std::min(sr->nodes.size(),
-                                                             static_cast<size_t>(SEARCH_MAX_BAD_NODES)))
+    
+    if (sr->getNumberOfConsecutiveBadNodes() >= std::min<size_t>(sr->nodes.size(), SEARCH_MAX_BAD_NODES))
     {
         if (logger_)
             logger_->w(sr->id, "[search %s IPv%c] expired", sr->id.toString().c_str(), sr->af == AF_INET ? '4' : '6');
@@ -750,7 +777,7 @@ Dht::search(const InfoHash& id, sa_family_t af, GetCallback gcb, QueryCallback q
         sr->done = false;
         sr->expired = false;
     } else {
-        if (searches4.size() + searches6.size() < MAX_SEARCHES) {
+        if (srs.size() < max_searches) {
             sr = std::make_shared<Search>();
             srs.emplace(id, sr);
         } else {
@@ -798,47 +825,12 @@ Dht::announce(const InfoHash& id,
 {
     auto& srs = searches(af);
     auto srp = srs.find(id);
-    auto sr = srp == srs.end() ? search(id, af) : srp->second;
-    if (!sr) {
-        if (callback)
-            callback(false, {});
-        return;
+    if (auto sr = srp == srs.end() ? search(id, af) : srp->second) {
+        sr->put(value, callback, created, permanent);
+        scheduler.edit(sr->nextSearchStep, scheduler.time());
+    } else if (callback) {
+        callback(false, {});
     }
-    sr->done = false;
-    sr->expired = false;
-    auto a_sr = std::find_if(sr->announce.begin(), sr->announce.end(), [&](const Announce& a){
-        return a.value->id == value->id;
-    });
-    if (a_sr == sr->announce.end()) {
-        sr->announce.emplace_back(Announce {permanent, value, created, callback});
-        for (auto& n : sr->nodes) {
-            n->probe_query.reset();
-            n->acked[value->id].first.reset();
-        }
-    } else {
-        a_sr->permanent = permanent;
-        a_sr->created = created;
-        if (a_sr->value != value) {
-            a_sr->value = value;
-            for (auto& n : sr->nodes) {
-                n->acked[value->id].first.reset();
-                n->probe_query.reset();
-            }
-        }
-        if (sr->isAnnounced(value->id)) {
-            if (a_sr->callback)
-                a_sr->callback(true, {});
-            a_sr->callback = {};
-            if (callback)
-                callback(true, {});
-            return;
-        } else {
-            if (a_sr->callback)
-                a_sr->callback(false, {});
-            a_sr->callback = callback;
-        }
-    }
-    scheduler.edit(sr->nextSearchStep, scheduler.time());
 }
 
 size_t
@@ -872,7 +864,7 @@ Dht::listen(const InfoHash& id, ValueCallback cb, Value::Filter f, Where where)
     auto query = std::make_shared<Query>(Select{}, std::move(where));
     auto filter = f.chain(query->where.getFilter());
     auto st = store.find(id);
-    if (st == store.end() && store.size() < MAX_HASHES)
+    if (st == store.end() && store.size() < max_store_keys)
         st = store.emplace(id, scheduler.time() + MAX_STORAGE_MAINTENANCE_EXPIRE_TIME).first;
 
     size_t tokenlocal = 0;
@@ -918,8 +910,8 @@ Dht::cancelListen(const InfoHash& id, size_t token)
                 srp->second->cancelListen(token, scheduler);
         }
     };
-    searches_cancel_listen(searches4, std::get<1>(it->second));
-    searches_cancel_listen(searches6, std::get<2>(it->second));
+    searches_cancel_listen(dht4.searches, std::get<1>(it->second));
+    searches_cancel_listen(dht6.searches, std::get<2>(it->second));
     listeners.erase(it);
     return true;
 }
@@ -1130,8 +1122,8 @@ Dht::getPut(const InfoHash& id) const
         auto vals = srp->second->getPut();
         ret.insert(ret.end(), vals.begin(), vals.end());
     };
-    find_values(searches4);
-    find_values(searches6);
+    find_values(dht4.searches);
+    find_values(dht6.searches);
     return ret;
 }
 
@@ -1142,9 +1134,9 @@ Dht::getPut(const InfoHash& id, const Value::Id& vid) const
         auto srp = srs.find(id);
         return (srp != srs.end()) ? srp->second->getPut(vid) : Sp<Value> {};
     };
-    if (auto v4 = find_value(searches4))
+    if (auto v4 = find_value(dht4.searches))
         return v4;
-    if (auto v6 = find_value(searches6))
+    if (auto v6 = find_value(dht6.searches))
         return v6;
     return {};
 }
@@ -1157,8 +1149,8 @@ Dht::cancelPut(const InfoHash& id, const Value::Id& vid)
         auto srp = srs.find(id);
         return (srp != srs.end()) ? srp->second->cancelPut(vid) : false;
     };
-    canceled |= sr_cancel_put(searches4);
-    canceled |= sr_cancel_put(searches6);
+    canceled |= sr_cancel_put(dht4.searches);
+    canceled |= sr_cancel_put(dht6.searches);
     if (canceled)
         storageErase(id, vid);
     return canceled;
@@ -1226,7 +1218,7 @@ Dht::storageStore(const InfoHash& id, const Sp<Value>& value, time_point created
 
     auto st = store.find(id);
     if (st == store.end()) {
-        if (store.size() >= MAX_HASHES)
+        if (store.size() >= max_store_keys)
             return false;
         auto st_i = store.emplace(id, now);
         st = st_i.first;
@@ -1272,7 +1264,7 @@ Dht::storageAddListener(const InfoHash& id, const Sp<Node>& node, size_t socket_
     const auto& now = scheduler.time();
     auto st = store.find(id);
     if (st == store.end()) {
-        if (store.size() >= MAX_HASHES)
+        if (store.size() >= max_store_keys)
             return;
         st = store.emplace(id, now).first;
     }
@@ -1282,7 +1274,7 @@ Dht::storageAddListener(const InfoHash& id, const Sp<Node>& node, size_t socket_
         auto vals = st->second.get(query.where.getFilter());
         if (not vals.empty()) {
             network_engine.tellListener(node, socket_id, id, WANT4 | WANT6, makeToken(node->getAddr(), false),
-                    buckets4.findClosestNodes(id, now, TARGET_NODES), buckets6.findClosestNodes(id, now, TARGET_NODES),
+                    dht4.buckets.findClosestNodes(id, now, TARGET_NODES), dht6.buckets.findClosestNodes(id, now, TARGET_NODES),
                     std::move(vals), query);
         }
         node_listeners.emplace(socket_id, Listener {now, std::forward<Query>(query)});
@@ -1460,11 +1452,17 @@ Dht::tokenMatch(const Blob& token, const SockAddr& addr) const
 
 NodeStats
 Dht::getNodesStats(sa_family_t af) const
+{
+    NodeStats stats = dht(af).getNodesStats(scheduler.time(), myid);
+    stats.node_cache_size = network_engine.getNodeCacheSize(af);
+    return stats;
+}
+
+NodeStats
+Dht::Kad::getNodesStats(time_point now, const InfoHash& myid) const
 {
     NodeStats stats {};
-    const auto& now = scheduler.time();
-    const auto& bcks = buckets(af);
-    for (const auto& b : bcks) {
+    for (const auto& b : buckets) {
         for (auto& n : b.nodes) {
             if (n->isGood(now)) {
                 stats.good_nodes++;
@@ -1476,26 +1474,11 @@ Dht::getNodesStats(sa_family_t af) const
         if (b.cached)
             stats.cached_nodes++;
     }
-    stats.table_depth = bcks.depth(bcks.findBucket(myid));
-    stats.searches = searches(af).size();
-    stats.cached_nodes = network_engine.getNodeCacheSize(af);
+    stats.table_depth = buckets.depth(buckets.findBucket(myid));
+    stats.searches = searches.size();
     return stats;
 }
 
-NodeMetrics
-Dht::getNodeMetrics() const
-{
-    NodeMetrics metrics;
-    metrics.scheduler_size = scheduler.size();
-    metrics.node_cache_size = network_engine.getNodeCacheSize();
-    metrics.rate_limiter_size = network_engine.getRateLimiterSize();
-    metrics.store_size = store.size();
-    metrics.store_quota_size = store_quota.size();
-    metrics.reported_addr_num = reported_addr.size();
-    metrics.request_count = net::REQ_COUNT.load();
-    return metrics;
-}
-
 void
 Dht::dumpBucket(const Bucket& b, std::ostream& out) const
 {
@@ -1613,18 +1596,18 @@ Dht::dumpTables() const
     out << "My id " << myid << std::endl;
 
     out << "Buckets IPv4 :" << std::endl;
-    for (const auto& b : buckets4)
+    for (const auto& b : dht4.buckets)
         dumpBucket(b, out);
     out << "Buckets IPv6 :" << std::endl;
-    for (const auto& b : buckets6)
+    for (const auto& b : dht6.buckets)
         dumpBucket(b, out);
 
     auto dump_searches = [&](std::map<InfoHash, Sp<Search>> srs) {
         for (auto& srp : srs)
             dumpSearch(*srp.second, out);
     };
-    dump_searches(searches4);
-    dump_searches(searches6);
+    dump_searches(dht4.searches);
+    dump_searches(dht6.searches);
     out << std::endl;
 
     out << getStorageLog() << std::endl;
@@ -1700,24 +1683,24 @@ std::string
 Dht::getSearchesLog(sa_family_t af) const
 {
     std::stringstream out;
-    auto num_searches = searches4.size() + searches6.size();
+    auto num_searches = dht4.searches.size() + dht6.searches.size();
     if (num_searches > 8) {
         if (not af or af == AF_INET)
-            for (const auto& sr : searches4)
+            for (const auto& sr : dht4.searches)
                 out << "[search " << sr.first << " IPv4]" << std::endl;
         if (not af or af == AF_INET6)
-            for (const auto& sr : searches6)
+            for (const auto& sr : dht6.searches)
                 out << "[search " << sr.first << " IPv6]" << std::endl;
     } else {
         out << "s:synched, u:updated, a:announced, c:candidate, f:cur req, x:expired, *:known" << std::endl;
         if (not af or af == AF_INET)
-            for (const auto& sr : searches4)
+            for (const auto& sr : dht4.searches)
                 dumpSearch(*sr.second, out);
         if (not af or af == AF_INET6)
-            for (const auto& sr : searches6)
+            for (const auto& sr : dht6.searches)
                 dumpSearch(*sr.second, out);
     }
-    out << "Total: " << num_searches << " searches (" << searches4.size() << " IPv4, " << searches6.size() << " IPv6)." << std::endl;
+    out << "Total: " << num_searches << " searches (" << dht4.searches.size() << " IPv4, " << dht6.searches.size() << " IPv6)." << std::endl;
     return out.str();
 }
 
@@ -1738,9 +1721,9 @@ Dht::getSearchLog(const InfoHash& id, sa_family_t af) const
 
 Dht::~Dht()
 {
-    for (auto& s : searches4)
+    for (auto& s : dht4.searches)
         s.second->clear();
-    for (auto& s : searches6)
+    for (auto& s : dht6.searches)
         s.second->clear();
 }
 
@@ -1759,7 +1742,12 @@ fromDhtConfig(const Config& config)
 Dht::Dht() : store(), network_engine(logger_, rd, scheduler, {}) {}
 
 Dht::Dht(std::unique_ptr<net::DatagramSocket>&& sock, const Config& config, const Sp<Logger>& l)
-    : DhtInterface(l), myid(config.node_id ? config.node_id : InfoHash::getRandom()), store(), store_quota(),
+    : DhtInterface(l),
+    myid(config.node_id ? config.node_id : InfoHash::getRandom()),
+    store(),
+    store_quota(),
+    max_store_keys(config.max_store_size ? (int)config.max_store_size : MAX_HASHES),
+    max_searches(config.max_searches ? (int)config.max_searches : MAX_SEARCHES),
     network_engine(myid, fromDhtConfig(config), std::move(sock), logger_, rd, scheduler,
             std::bind(&Dht::onError, this, _1, _2),
             std::bind(&Dht::onNewNode, this, _1, _2),
@@ -1780,12 +1768,12 @@ Dht::Dht(std::unique_ptr<net::DatagramSocket>&& sock, const Config& config, cons
     if (not s or (not s->hasIPv4() and not s->hasIPv6()))
         throw DhtException("Opened socket required");
     if (s->hasIPv4()) {
-        buckets4 = {Bucket {AF_INET}};
-        buckets4.is_client = config.is_bootstrap;
+        dht4.buckets = {Bucket {AF_INET}};
+        dht4.buckets.is_client = config.is_bootstrap;
     }
     if (s->hasIPv6()) {
-        buckets6 = {Bucket {AF_INET6}};
-        buckets6.is_client = config.is_bootstrap;
+        dht6.buckets = {Bucket {AF_INET6}};
+        dht6.buckets.is_client = config.is_bootstrap;
     }
 
     search_id = std::uniform_int_distribution<decltype(search_id)>{}(rd);
@@ -1980,13 +1968,34 @@ 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));
 
-    expireBuckets(buckets4);
-    expireBuckets(buckets6);
+    expireBuckets(dht4.buckets);
+    expireBuckets(dht6.buckets);
     expireStore();
     expireSearches();
     scheduler.add(expire_stuff_time, std::bind(&Dht::expire, this));
 }
 
+void
+Dht::onDisconnected()
+{
+    if (dht4.status != NodeStatus::Disconnected || dht6.status != NodeStatus::Disconnected)
+        return;
+    if (logger_)
+        logger_->d(myid, "Bootstraping");
+    for (const auto& boootstrap : bootstrap_nodes)
+        try {
+            for (const auto& ip : network_engine.getSocket()->resolve(boootstrap.first, boootstrap.second))
+                pingNode(ip);
+        } catch (const std::exception& e) {
+            if (logger_)
+                logger_->e(myid, "Can't resolve %s:%s: %s", boootstrap.first.c_str(), boootstrap.second.c_str(), e.what());
+        }
+    if (bootstrapJob)
+        bootstrapJob->cancel();
+    bootstrapJob = scheduler.add(scheduler.time() + bootstrap_period, std::bind(&Dht::onDisconnected, this));
+    bootstrap_period *= 2;
+}
+
 void
 Dht::confirmNodes()
 {
@@ -1994,25 +2003,25 @@ Dht::confirmNodes()
     bool soon = false;
     const auto& now = scheduler.time();
 
-    if (searches4.empty() and getStatus(AF_INET) == NodeStatus::Connected) {
+    if (dht4.searches.empty() and dht4.status == NodeStatus::Connected) {
         if (logger_)
             logger_->d(myid, "[confirm nodes] initial IPv4 'get' for my id (%s)", myid.toString().c_str());
         search(myid, AF_INET);
     }
-    if (searches6.empty() and getStatus(AF_INET6) == NodeStatus::Connected) {
+    if (dht6.searches.empty() and dht6.status == NodeStatus::Connected) {
         if (logger_)
             logger_->d(myid, "[confirm nodes] initial IPv6 'get' for my id (%s)", myid.toString().c_str());
         search(myid, AF_INET6);
     }
 
-    soon |= bucketMaintenance(buckets4);
-    soon |= bucketMaintenance(buckets6);
+    soon |= bucketMaintenance(dht4.buckets);
+    soon |= bucketMaintenance(dht6.buckets);
 
     if (!soon) {
-        if (buckets4.grow_time >= now - seconds(150))
-            soon |= neighbourhoodMaintenance(buckets4);
-        if (buckets6.grow_time >= now - seconds(150))
-            soon |= neighbourhoodMaintenance(buckets6);
+        if (dht4.buckets.grow_time >= now - seconds(150))
+            soon |= neighbourhoodMaintenance(dht4.buckets);
+        if (dht6.buckets.grow_time >= now - seconds(150))
+            soon |= neighbourhoodMaintenance(dht6.buckets);
     }
 
     /* In order to maintain all buckets' age within 600 seconds, worst
@@ -2097,25 +2106,25 @@ Dht::exportNodes() const
 {
     const auto& now = scheduler.time();
     std::vector<NodeExport> nodes;
-    const auto b4 = buckets4.findBucket(myid);
-    if (b4 != buckets4.end()) {
+    const auto b4 = dht4.buckets.findBucket(myid);
+    if (b4 != dht4.buckets.end()) {
         for (auto& n : b4->nodes)
             if (n->isGood(now))
                 nodes.push_back(n->exportNode());
     }
-    const auto b6 = buckets6.findBucket(myid);
-    if (b6 != buckets6.end()) {
+    const auto b6 = dht6.buckets.findBucket(myid);
+    if (b6 != dht6.buckets.end()) {
         for (auto& n : b6->nodes)
             if (n->isGood(now))
                 nodes.push_back(n->exportNode());
     }
-    for (auto b = buckets4.begin(); b != buckets4.end(); ++b) {
+    for (auto b = dht4.buckets.begin(); b != dht4.buckets.end(); ++b) {
         if (b == b4) continue;
         for (auto& n : b->nodes)
             if (n->isGood(now))
                 nodes.push_back(n->exportNode());
     }
-    for (auto b = buckets6.begin(); b != buckets6.end(); ++b) {
+    for (auto b = dht6.buckets.begin(); b != dht6.buckets.end(); ++b) {
         if (b == b6) continue;
         for (auto& n : b->nodes)
             if (n->isGood(now))
@@ -2139,7 +2148,7 @@ Dht::pingNode(SockAddr sa, DoneCallbackSimple&& cb)
     scheduler.syncTime();
     if (logger_)
         logger_->d("Sending ping to %s", sa.toString().c_str());
-    auto& count = sa.getFamily() == AF_INET ? pending_pings4 : pending_pings6;
+    auto& count = dht(sa.getFamily()).pending_pings;
     count++;
     network_engine.sendPing(std::move(sa), [&count,cb](const net::Request&, net::RequestAnswer&&) {
         count--;
@@ -2200,9 +2209,9 @@ Dht::onFindNode(Sp<Node> node, const InfoHash& target, want_t want)
     net::RequestAnswer answer;
     answer.ntoken = makeToken(node->getAddr(), false);
     if (want & WANT4)
-        answer.nodes4 = buckets4.findClosestNodes(target, now, TARGET_NODES);
+        answer.nodes4 = dht4.buckets.findClosestNodes(target, now, TARGET_NODES);
     if (want & WANT6)
-        answer.nodes6 = buckets6.findClosestNodes(target, now, TARGET_NODES);
+        answer.nodes6 = dht6.buckets.findClosestNodes(target, now, TARGET_NODES);
     return answer;
 }
 
@@ -2221,8 +2230,8 @@ Dht::onGetValues(Sp<Node> node, const InfoHash& hash, want_t, const Query& query
     net::RequestAnswer answer {};
     auto st = store.find(hash);
     answer.ntoken = makeToken(node->getAddr(), false);
-    answer.nodes4 = buckets4.findClosestNodes(hash, now, TARGET_NODES);
-    answer.nodes6 = buckets6.findClosestNodes(hash, now, TARGET_NODES);
+    answer.nodes4 = dht4.buckets.findClosestNodes(hash, now, TARGET_NODES);
+    answer.nodes6 = dht6.buckets.findClosestNodes(hash, now, TARGET_NODES);
     if (st != store.end() && not st->second.empty()) {
         answer.values = st->second.get(query.where.getFilter());
         if (logger_)

src/dhtrunner.cpp

@@ -174,7 +174,7 @@ DhtRunner::run(const Config& config, Context&& context)
                     if (not pending_ops_prio.empty())
                         return true;
                     auto s = getStatus();
-                    if (not pending_ops.empty() and (s == NodeStatus::Connected or (s == NodeStatus::Disconnected and not bootstraping)))
+                    if (not pending_ops.empty() and (s == NodeStatus::Connected or s == NodeStatus::Disconnected))
                         return true;
                 }
                 return false;
@@ -300,7 +300,6 @@ DhtRunner::join()
         if (running.exchange(State::Idle) == State::Idle)
             return;
         cv.notify_all();
-        bootstrap_cv.notify_all();
 #ifdef OPENDHT_PEER_DISCOVERY
         if (peerDiscovery_)
             peerDiscovery_->stop();
@@ -315,9 +314,6 @@ DhtRunner::join()
     if (dht_thread.joinable())
         dht_thread.join();
 
-    if (bootstrap_thread.joinable())
-        bootstrap_thread.join();
-
     {
         std::lock_guard<std::mutex> lck(storage_mtx);
         pending_ops = decltype(pending_ops)();
@@ -479,8 +475,6 @@ DhtRunner::getNodeInfo() const {
         info.node_id = dht->getNodeId();
         info.ipv4 = dht->getNodesStats(AF_INET);
         info.ipv6 = dht->getNodesStats(AF_INET6);
-        info.metrics = dht->getNodeMetrics();
-        info.ongoing_ops = ongoing_ops;
     }
     info.ongoing_ops = ongoing_ops;
     return info;
@@ -498,7 +492,6 @@ DhtRunner::getNodeInfo(std::function<void(std::shared_ptr<NodeInfo>)> cb)
         info.node_id = dht.getNodeId();
         info.ipv4 = dht.getNodesStats(AF_INET);
         info.ipv6 = dht.getNodesStats(AF_INET6);
-        info.metrics = dht.getNodeMetrics();
         info.ongoing_ops = ongoing_ops;
         cb(std::move(sinfo));
         opEnded();
@@ -587,7 +580,7 @@ DhtRunner::loop_()
     {
         std::lock_guard<std::mutex> lck(storage_mtx);
         auto s = getStatus();
-        ops = (pending_ops_prio.empty() && (s == NodeStatus::Connected or (s == NodeStatus::Disconnected and not bootstraping))) ?
+        ops = (pending_ops_prio.empty() && (s == NodeStatus::Connected or s == NodeStatus::Disconnected)) ?
                std::move(pending_ops) : std::move(pending_ops_prio);
     }
     while (not ops.empty()) {
@@ -642,20 +635,11 @@ DhtRunner::loop_()
     if (dropped)
         std::cerr << "Dropped " << dropped << " packets with high delay" << std::endl;
 
-    NodeStatus nstatus4 = dht->getStatus(AF_INET);
-    NodeStatus nstatus6 = dht->getStatus(AF_INET6);
+    NodeStatus nstatus4 = dht->updateStatus(AF_INET);
+    NodeStatus nstatus6 = dht->updateStatus(AF_INET6);
     if (nstatus4 != status4 || nstatus6 != status6) {
         status4 = nstatus4;
         status6 = nstatus6;
-        if (status4 == NodeStatus::Disconnected and status6 == NodeStatus::Disconnected) {
-            // We have lost connection with the DHT.  Try to recover using bootstrap nodes.
-            std::unique_lock<std::mutex> lck(bootstrap_mtx);
-            bootstrap_nodes = bootstrap_nodes_all;
-            tryBootstrapContinuously();
-        } else {
-            std::unique_lock<std::mutex> lck(bootstrap_mtx);
-            bootstrap_nodes.clear();
-        }
         if (statusCb)
             statusCb(status4, status6);
     }
@@ -902,89 +886,34 @@ DhtRunner::putEncrypted(const std::string& key, InfoHash to, Value&& value, Done
     putEncrypted(InfoHash::get(key), to, std::forward<Value>(value), std::move(cb), permanent);
 }
 
-void
-DhtRunner::tryBootstrapContinuously()
-{
-    if (bootstrap_thread.joinable()) {
-        if (bootstraping)
-            return; // already running
-        else
-            bootstrap_thread.join();
-    }
-    bootstraping = true;
-    bootstrap_thread = std::thread([this]() {
-        auto next = clock::now();
-        do {
-            decltype(bootstrap_nodes) nodes;
-            {
-                std::lock_guard<std::mutex> lck(bootstrap_mtx);
-                nodes = bootstrap_nodes;
-            }
-
-            next += BOOTSTRAP_PERIOD;
-            {
-                std::mutex mtx;
-                std::unique_lock<std::mutex> blck(mtx);
-                unsigned ping_count(0);
-                // Reverse: try last inserted bootstrap nodes first
-                for (auto it = nodes.rbegin(); it != nodes.rend(); it++) {
-                    ++ping_count;
-                    try {
-                        bootstrap(SockAddr::resolve(it->first, it->second), [&](bool) {
-                            if (running != State::Running)
-                                return;
-                            {
-                                std::unique_lock<std::mutex> blck(mtx);
-                                --ping_count;
-                            }
-                            bootstrap_cv.notify_all();
-                        });
-                    } catch (std::invalid_argument& e) {
-                        --ping_count;
-                        std::cerr << e.what() << std::endl;
-                    }
-                }
-                // wait at least until the next BOOTSTRAP_PERIOD
-                bootstrap_cv.wait_until(blck, next, [&]() { return running != State::Running; });
-                // wait for bootstrap requests to end.
-                if (running != State::Running)
-                   bootstrap_cv.wait(blck, [&]() { return running != State::Running or ping_count == 0; });
-            }
-            // update state
-            {
-                std::lock_guard<std::mutex> lck(dht_mtx);
-                bootstraping = running == State::Running and
-                               status4 == NodeStatus::Disconnected and
-                               status6 == NodeStatus::Disconnected;
-            }
-        } while (bootstraping);
-    });
-}
-
 void
 DhtRunner::bootstrap(const std::string& host, const std::string& service)
 {
-    std::lock_guard<std::mutex> lck(bootstrap_mtx);
-    bootstrap_nodes_all.emplace_back(host, service);
-    bootstrap_nodes.emplace_back(host, service);
-    tryBootstrapContinuously();
+    std::lock_guard<std::mutex> lck(storage_mtx);
+    pending_ops_prio.emplace([host, service] (SecureDht& dht) mutable {
+        dht.addBootstrap(host, service);
+    });
+    cv.notify_all();
 }
 
 void
 DhtRunner::bootstrap(const std::string& hostService)
 {
-    std::lock_guard<std::mutex> lck(bootstrap_mtx);
-    auto host_service = splitPort(hostService);
-    bootstrap_nodes_all.emplace_back(host_service.first, host_service.second);
-    bootstrap_nodes.emplace_back(std::move(host_service.first), std::move(host_service.second));
-    tryBootstrapContinuously();
+    std::lock_guard<std::mutex> lck(storage_mtx);
+    pending_ops_prio.emplace([host_service = splitPort(hostService)] (SecureDht& dht) mutable {
+        dht.addBootstrap(host_service.first, host_service.second);
+    });
+    cv.notify_all();
 }
 
 void
 DhtRunner::clearBootstrap()
 {
-    std::lock_guard<std::mutex> lck(bootstrap_mtx);
-    bootstrap_nodes_all.clear();
+    std::lock_guard<std::mutex> lck(storage_mtx);
+    pending_ops_prio.emplace([] (SecureDht& dht) mutable {
+        dht.clearBootstrap();
+    });
+    cv.notify_all();
 }
 
 void

src/network_engine.cpp

@@ -281,6 +281,9 @@ NetworkEngine::requestStep(Sp<Request> sreq)
             ++req.attempt_count;
             req.attempt_duration +=
                 req.attempt_duration + uniform_duration_distribution<>(0ms, ((duration)Node::MAX_RESPONSE_TIME)/4)(rd);
+            if (not req.parts.empty()){
+                sendValueParts(req.tid, req.parts, node.getAddr());
+            }
         }
         std::weak_ptr<Request> wreq = sreq;
         scheduler.add(req.last_try + req.attempt_duration, [this,wreq] {
@@ -1175,9 +1178,8 @@ NetworkEngine::sendAnnounceValue(Sp<Node> n,
             }
         }
     );
+    req->parts = std::move(v);
     sendRequest(req);
-    if (not v.empty())
-        sendValueParts(tid, v, n->getAddr());
     ++out_stats.put;
     return req;
 }

src/request.h

@@ -29,8 +29,6 @@ namespace net {
 class NetworkEngine;
 struct ParsedMessage;
 
-extern std::atomic<size_t> REQ_COUNT;
-
 /*!
  * @class   Request
  * @brief   An atomic request destined to a node.
@@ -69,22 +67,14 @@ struct Request {
         }
     }
 
-    Request(State state = State::PENDING) : state_(state) {
-        REQ_COUNT++;
-    }
+    Request(State state = State::PENDING) : state_(state) {}
     Request(MessageType type, Tid tid,
             Sp<Node> node,
             Blob&& msg,
             std::function<void(const Request&, ParsedMessage&&)> on_done,
             std::function<void(const Request&, bool)> on_expired,
             Tid socket = 0) :
-        node(node), tid(tid), type(type), on_done(on_done), on_expired(on_expired), msg(std::move(msg)), socket(socket) {
-            REQ_COUNT++;
-        }
-
-    ~Request() {
-        REQ_COUNT--;
-    }
+        node(node), tid(tid), type(type), on_done(on_done), on_expired(on_expired), msg(std::move(msg)), socket(socket) { }
 
     Tid getTid() const { return tid; }
     MessageType getType() const { return type; }
@@ -125,6 +115,7 @@ private:
         on_done = {};
         on_expired = {};
         msg = {};
+        parts = {};
     }
 
     const Tid tid {0}; /* the request id. */
@@ -140,6 +131,7 @@ private:
     std::function<void(const Request&, bool)> on_expired {};
 
     Blob msg {};                      /* the serialized message. */
+    std::vector<Blob> parts;
     Tid socket;   /* the socket used for further reponses. */
 };
 

src/search.h

@@ -568,6 +568,43 @@ struct Dht::Search {
         return canceled;
     }
 
+    void put(const Sp<Value>& value, DoneCallback callback, time_point created, bool permanent) {
+        done = false;
+        expired = false;
+        auto a_sr = std::find_if(announce.begin(), announce.end(), [&](const Announce& a){
+            return a.value->id == value->id;
+        });
+        if (a_sr == announce.end()) {
+            announce.emplace_back(Announce {permanent, value, created, callback});
+            for (auto& n : nodes) {
+                n->probe_query.reset();
+                n->acked[value->id].first.reset();
+            }
+        } else {
+            a_sr->permanent = permanent;
+            a_sr->created = created;
+            if (a_sr->value != value) {
+                a_sr->value = value;
+                for (auto& n : nodes) {
+                    n->acked[value->id].first.reset();
+                    n->probe_query.reset();
+                }
+            }
+            if (isAnnounced(value->id)) {
+                if (a_sr->callback)
+                    a_sr->callback(true, {});
+                a_sr->callback = {};
+                if (callback)
+                    callback(true, {});
+                return;
+            } else {
+                if (a_sr->callback)
+                    a_sr->callback(false, {});
+                a_sr->callback = callback;
+            }
+        }
+    }
+
     /**
      * @return The number of non-good search nodes.
      */
@@ -579,7 +616,7 @@ struct Dht::Search {
     unsigned getNumberOfConsecutiveBadNodes() const {
         unsigned count = 0;
         std::find_if(nodes.begin(), nodes.end(), [&count](const std::unique_ptr<SearchNode>& sn) {
-            if (not sn->isBad())
+            if (not sn->node->isExpired())
                 return true;
             ++count;
             return false;

src/utils.cpp

@@ -34,9 +34,6 @@
 #endif
 
 namespace dht {
-namespace net {
-std::atomic<size_t> REQ_COUNT {0};
-}
 
 static constexpr std::array<uint8_t, 12> MAPPED_IPV4_PREFIX {{0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0xff, 0xff}};
 
@@ -79,11 +76,8 @@ SockAddr::resolve(const std::string& host, const std::string& service)
     if(rc != 0)
         throw std::invalid_argument(std::string("Error: `") + host + ":" + service + "`: " + gai_strerror(rc));
 
-    addrinfo* infop = info;
-    while (infop) {
+    for (addrinfo* infop = info; infop; infop = infop->ai_next)
         ips.emplace_back(infop->ai_addr, infop->ai_addrlen);
-        infop = infop->ai_next;
-    }
     freeaddrinfo(info);
     return ips;
 }

src/value.cpp

@@ -75,11 +75,13 @@ std::ostream& operator<< (std::ostream& s, const Value& v)
                 s << '"';
                 s.write((const char*)v.data.data(), v.data.size());
                 s << '"';
-            } else {
+            } else if (v.data.size() < 1024) {
                 s << std::hex;
                 for (auto i : v.data)
                     s << std::setfill('0') << std::setw(2) << (unsigned)i;
                 s << std::dec;
+            } else {
+                s << v.data.size() << " bytes";
             }
         }
     }

tests/dhtproxytester.cpp

@@ -34,10 +34,13 @@ CPPUNIT_TEST_SUITE_REGISTRATION(DhtProxyTester);
 
 void
 DhtProxyTester::setUp() {
-    nodePeer.run(0);
+    clientConfig.dht_config.node_config.max_peer_req_per_sec = -1;
+    clientConfig.dht_config.node_config.max_req_per_sec = -1;
+
+    nodePeer.run(0, clientConfig);
 
     nodeProxy = std::make_shared<dht::DhtRunner>();
-    nodeProxy->run(0);
+    nodeProxy->run(0, clientConfig);
     nodeProxy->bootstrap(nodePeer.getBound());
 
     auto serverCAIdentity = dht::crypto::generateEcIdentity("DHT Node CA");
@@ -50,8 +53,6 @@ DhtProxyTester::setUp() {
 
     clientConfig.server_ca = serverCAIdentity.second;
     clientConfig.client_identity = dht::crypto::generateIdentity("DhtProxyTester");
-    clientConfig.dht_config.node_config.maintain_storage = false;
-    clientConfig.threaded = true;
     clientConfig.push_node_id = "dhtnode";
     clientConfig.proxy_server = "https://127.0.0.1:8080";
 }
@@ -280,7 +281,7 @@ DhtProxyTester::testFuzzy()
         },[&](bool ok){
             CPPUNIT_ASSERT(ok);
         });
-        std::this_thread::sleep_for(10ms);
+        std::this_thread::sleep_for(5ms);
     }
 
     // Assert

tests/dhtrunnertester.cpp

@@ -29,8 +29,12 @@ CPPUNIT_TEST_SUITE_REGISTRATION(DhtRunnerTester);
 
 void
 DhtRunnerTester::setUp() {
-    node1.run(42222);
-    node2.run(42232);
+    dht::DhtRunner::Config config;
+    config.dht_config.node_config.max_peer_req_per_sec = -1;
+    config.dht_config.node_config.max_req_per_sec = -1;
+
+    node1.run(42222, config);
+    node2.run(42232, config);
     node2.bootstrap(node1.getBound());
 }
 
@@ -84,7 +88,9 @@ DhtRunnerTester::testListen() {
     auto a = dht::InfoHash::get("234");
     auto b = dht::InfoHash::get("2345");
     auto c = dht::InfoHash::get("23456");
-    constexpr unsigned N = 32;
+    auto d = dht::InfoHash::get("234567");
+    constexpr unsigned N = 256;
+    constexpr unsigned SZ = 56 * 1024;
 
     auto ftokena = node1.listen(a, [&](const std::shared_ptr<dht::Value>&) {
         valueCount++;
@@ -101,6 +107,16 @@ DhtRunnerTester::testListen() {
         return true;
     });
 
+    auto ftokend = node1.listen(d, [&](const std::shared_ptr<dht::Value>&) {
+        valueCount++;
+        return true;
+    });
+
+    std::vector<uint8_t> mtu;
+    mtu.reserve(SZ);
+    for (size_t i = 0; i < SZ; i++)
+        mtu.emplace_back((i % 2) ? 'T' : 'M');
+
     for (unsigned i=0; i<N; i++) {
         node2.put(a, dht::Value("v1"), [&](bool ok) {
             std::lock_guard<std::mutex> lock(mutex);
@@ -114,29 +130,41 @@ DhtRunnerTester::testListen() {
             if (ok) putOkCount++;
             cv.notify_all();
         });
+        auto bigVal = std::make_shared<dht::Value>();
+        bigVal->data = mtu;
+        node2.put(c, bigVal, [&](bool ok) {
+            std::lock_guard<std::mutex> lock(mutex);
+            putCount++;
+            if (ok) putOkCount++;
+            cv.notify_all();
+        });
     }
 
     {
         std::unique_lock<std::mutex> lk(mutex);
-        CPPUNIT_ASSERT(cv.wait_for(lk, 30s, [&]{ return putCount == N * 2u; }));
-        CPPUNIT_ASSERT_EQUAL(N * 2u, putOkCount);
+        CPPUNIT_ASSERT(cv.wait_for(lk, 30s, [&]{ return putCount == N * 3u; }));
+        CPPUNIT_ASSERT_EQUAL(N * 3u, putOkCount);
     }
 
     CPPUNIT_ASSERT(ftokena.valid());
     CPPUNIT_ASSERT(ftokenb.valid());
     CPPUNIT_ASSERT(ftokenc.valid());
+    CPPUNIT_ASSERT(ftokend.valid());
 
     auto tokena = ftokena.get();
     auto tokenc = ftokenc.get();
+    auto tokend = ftokend.get();
     // tokenb might be 0 since the callback returns false.
 
     CPPUNIT_ASSERT(tokena);
     CPPUNIT_ASSERT(tokenc);
-    CPPUNIT_ASSERT_EQUAL(N + 1u, valueCount.load());
+    CPPUNIT_ASSERT(tokend);
+    CPPUNIT_ASSERT_EQUAL(N * 2u + 1u, valueCount.load());
 
     node1.cancelListen(a, tokena);
     node1.cancelListen(b, std::move(ftokenb));
     node1.cancelListen(c, tokenc);
+    node1.cancelListen(d, tokend);
 }
 
 }  // namespace test

tools/dhtnode.cpp

@@ -145,14 +145,6 @@ void cmd_loop(std::shared_ptr<DhtRunner>& node, dht_params& params
         } else if (op == "ll") {
             auto nodeInfo = node->getNodeInfo();
             print_node_info(*node, params);
-            std::cout << "Scheduler size:" << nodeInfo.metrics.scheduler_size << std::endl;
-            std::cout << "RateLimiter size:" << nodeInfo.metrics.rate_limiter_size << std::endl;
-            std::cout << "ReportedAddr count:" << nodeInfo.metrics.reported_addr_num << std::endl;
-            std::cout << "Node cache size:" << nodeInfo.metrics.node_cache_size << std::endl;
-            std::cout << "Store size:" << nodeInfo.metrics.store_size << std::endl;
-            std::cout << "Store quota size:" << nodeInfo.metrics.store_quota_size << std::endl;
-            std::cout << "Global request count:" << nodeInfo.metrics.request_count << std::endl;
-
             std::cout << nodeInfo.ongoing_ops << " ongoing operations" << std::endl;
             std::cout << "IPv4 stats:" << std::endl;
             std::cout << nodeInfo.ipv4.toString() << std::endl;

tools/tools_common.h

@@ -170,6 +170,8 @@ getDhtConfig(dht_params& params)
     if (params.no_rate_limit) {
         config.dht_config.node_config.max_req_per_sec = -1;
         config.dht_config.node_config.max_peer_req_per_sec = -1;
+        config.dht_config.node_config.max_searches = -1;
+        config.dht_config.node_config.max_store_size = -1;
     }
 
     dht::DhtRunner::Context context {};