update to version 2.1.0

network engine: check for logger
tools: cleanup
2026-11-08 12:28:42 +00:00 · 2020-04-30 14:58:25 -04:00 · 2020-04-29 20:23:19 -04:00 · 2020-04-29 19:22:24 -04:00 · 2020-04-29 17:12:11 -04:00 · 2020-04-29 16:13:36 -04:00
31 changed files with 603 additions and 396 deletions
@@ -4,7 +4,7 @@ include(CMakeDependentOption)

 project (opendht)
 set (opendht_VERSION_MAJOR 2)
-set (opendht_VERSION_MINOR 0.0)
+set (opendht_VERSION_MINOR 1.0)
 set (opendht_VERSION ${opendht_VERSION_MAJOR}.${opendht_VERSION_MINOR})
 set (PACKAGE_VERSION ${opendht_VERSION})
 set (VERSION "${opendht_VERSION}")
@@ -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'))
@@ -1,6 +1,6 @@
 dnl define macros
 m4_define([opendht_major_version], 2)
-m4_define([opendht_minor_version], 0)
+m4_define([opendht_minor_version], 1)
 m4_define([opendht_patch_version], 0)
 m4_define([opendht_version],
 		  [opendht_major_version.opendht_minor_version.opendht_patch_version])
@@ -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]))
 ])

@@ -58,6 +58,8 @@ struct OPENDHT_PUBLIC NodeStats {
             cached_nodes {0},
             incoming_nodes {0};
    unsigned table_depth {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;
@@ -71,7 +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)
+    MSGPACK_DEFINE_MAP(good_nodes, dubious_nodes, cached_nodes, incoming_nodes, table_depth, searches, node_cache_size)
 };

 struct OPENDHT_PUBLIC NodeInfo {
@@ -80,6 +82,8 @@ struct OPENDHT_PUBLIC NodeInfo {
    NodeStats ipv4 {};
    NodeStats ipv6 {};
    size_t ongoing_ops {0};
+    in_port_t bound4 {0};
+    in_port_t bound6 {0};

 #ifdef OPENDHT_JSONCPP
    /**
@@ -116,12 +120,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
@@ -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
@@ -330,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};

@@ -366,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.
@@ -394,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;

@@ -422,7 +445,7 @@ private:
    void reportedAddr(const SockAddr&);

    // Storage
-    void storageAddListener(const InfoHash& id, const Sp<Node>& node, size_t tid, Query&& = {});
+    void storageAddListener(const InfoHash& id, const Sp<Node>& node, size_t tid, Query&& = {}, int version = 0);
    bool storageStore(const InfoHash& id, const Sp<Value>& value, time_point created, const SockAddr& sa = {}, bool permanent = false);
    bool storageErase(const InfoHash& id, Value::Id vid);
    bool storageRefresh(const InfoHash& id, Value::Id vid);
@@ -442,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);
@@ -465,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
@@ -489,6 +514,7 @@ private:

    void confirmNodes();
    void expire();
+    void onDisconnected();

    /**
     * Generic function to execute when a 'get' request has completed.
@@ -574,7 +600,8 @@ private:
            const InfoHash& hash,
            const Blob& token,
            size_t socket_id,
-            const Query& query);
+            const Query& query,
+            int version = 0);
    void onListenDone(const Sp<Node>& status,
            net::RequestAnswer& a,
            Sp<Search>& sr);
@@ -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
@@ -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 {};
@@ -180,7 +180,8 @@ private:
            const InfoHash&,
            const Blob&,
            Tid,
-            const Query&)> onListen {};
+            const Query&,
+            int)> onListen {};
    /**
     * Called on announce request.
     *
@@ -237,9 +238,10 @@ public:
    void clear();

    /**
-     * Sends values (with closest nodes) to a listenner.
+     * Sends values (with closest nodes) to a listener.
     *
-     * @param sa          The address of the listenner.
+     * @deprecated
+     * @param sa          The address of the listener.
     * @param sslen       The length of the sockaddr structure.
     * @param socket_id  The tid to use to write to the request socket.
     * @param hash        The hash key of the value.
@@ -248,10 +250,11 @@ public:
     * @param nodes       The ipv4 closest nodes.
     * @param nodes6      The ipv6 closest nodes.
     * @param values      The values to send.
+     * @param version     If version = 1, a request will be used to answer to the listener
     */
    void tellListener(Sp<Node> n, Tid socket_id, const InfoHash& hash, want_t want, const Blob& ntoken,
            std::vector<Sp<Node>>&& nodes, std::vector<Sp<Node>>&& nodes6,
-            std::vector<Sp<Value>>&& values, const Query& q);
+            std::vector<Sp<Value>>&& values, const Query& q, int version);

    void tellListenerRefreshed(Sp<Node> n, Tid socket_id, const InfoHash& hash, const Blob& ntoken, const std::vector<Value::Id>& values);
    void tellListenerExpired(Sp<Node> n, Tid socket_id, const InfoHash& hash, const Blob& ntoken, const std::vector<Value::Id>& values);
@@ -276,7 +279,7 @@ public:
     */
    Sp<Request>
    sendPing(Sp<Node> n, RequestCb&& on_done, RequestExpiredCb&& on_expired);
-    
+
    /**
     * Send a "ping" request to a given node.
     *
@@ -293,7 +296,7 @@ public:
                std::forward<RequestCb>(on_done),
                std::forward<RequestExpiredCb>(on_expired));
    }
-    
+
    /**
     * Send a "find node" request to a given node.
     *
@@ -401,6 +404,24 @@ public:
                                 const Blob& token,
                                 RequestCb&& on_done,
                                 RequestExpiredCb&& on_expired);
+    /**
+     * Send a "update" request to a given node. Used for Listen operations
+     *
+     * @param n           The node.
+     * @param hash        The target hash.
+     * @param values      The values.
+     * @param created     Time id.
+     * @param token       A security token.
+     * @param sid         The socket id.
+     *
+     * @return the request with information concerning its success.
+     */
+    Sp<Request> sendUpdateValues(Sp<Node> n,
+                                 const InfoHash& infohash,
+                                 const std::vector<Sp<Value>>& values,
+                                 time_point created,
+                                 const Blob& token,
+                                 const size_t& sid);

    /**
     * Parses a message and calls appropriate callbacks.
@@ -432,6 +453,21 @@ public:
        return cache.getCachedNodes(id, sa_f, count);
    }

+    size_t getNodeCacheSize() const {
+        return cache.size();
+    }
+    size_t getNodeCacheSize(sa_family_t af) const {
+        return cache.size(af);
+    }
+
+    size_t getRateLimiterSize() const {
+        return address_rate_limiter.size();
+    }
+
+    size_t getPartialCount() const {
+        return partial_messages.size();
+    }
+
 private:

    struct PartialMessage;
@@ -475,19 +511,20 @@ private:
    void sendRequest(const Sp<Request>& request);

    struct MessageStats {
-        unsigned ping    {0};
-        unsigned find    {0};
-        unsigned get     {0};
-        unsigned put     {0};
-        unsigned listen  {0};
-        unsigned refresh {0};
+        unsigned ping         {0};
+        unsigned find         {0};
+        unsigned get          {0};
+        unsigned put          {0};
+        unsigned listen       {0};
+        unsigned updateValue  {0};
+        unsigned refresh      {0};
    };


    // basic wrapper for socket sendto function
    int send(const SockAddr& addr, const char *buf, size_t len, bool confirmed = false);

-    void sendValueParts(const TransId& tid, const std::vector<Blob>& svals, const SockAddr& addr);
+    void sendValueParts(Tid tid, const std::vector<Blob>& svals, const SockAddr& addr);
    std::vector<Blob> packValueHeader(msgpack::sbuffer&, const std::vector<Sp<Value>>&);
    void maintainRxBuffer(Tid tid);

@@ -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:

@@ -26,6 +26,13 @@
 namespace dht {

 struct NodeCache {
+    size_t size(sa_family_t family) const {
+        return cache(family).count();
+    }
+    size_t size() const {
+        return size(AF_INET) + size(AF_INET6);
+    }
+
    Sp<Node> getNode(const InfoHash& id, sa_family_t family);
    Sp<Node> getNode(const InfoHash& id, const SockAddr&, time_point now, bool confirmed, bool client=false);
    std::vector<Sp<Node>> getCachedNodes(const InfoHash& id, sa_family_t sa_f, size_t count) const;
@@ -49,6 +56,7 @@ private:
        void clearBadNodes();
        void setExpired();
        void cleanup();
+        size_t count() const { return size(); }
    private:
        size_t cleanup_counter {0};
    };
@@ -201,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);
    }
@@ -216,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);
    }
@@ -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";
@@ -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}
@@ -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;
@@ -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;
@@ -1209,7 +1201,7 @@ Dht::storageChanged(const InfoHash& id, Storage& st, ValueStorage& v, bool newVa
                vals.push_back(v.data);
                Blob ntoken = makeToken(node_listeners.first->getAddr(), false);
                network_engine.tellListener(node_listeners.first, l.first, id, 0, ntoken, {}, {},
-                        std::move(vals), l.second.query);
+                        std::move(vals), l.second.query, l.second.version);
            }
        }
    }
@@ -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;
@@ -1267,12 +1259,12 @@ Dht::storageErase(const InfoHash& id, Value::Id vid)
 }

 void
-Dht::storageAddListener(const InfoHash& id, const Sp<Node>& node, size_t socket_id, Query&& query)
+Dht::storageAddListener(const InfoHash& id, const Sp<Node>& node, size_t socket_id, Query&& query, int version)
 {
    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,10 +1274,10 @@ 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),
-                    std::move(vals), query);
+                    dht4.buckets.findClosestNodes(id, now, TARGET_NODES), dht6.buckets.findClosestNodes(id, now, TARGET_NODES),
+                    std::move(vals), query, version);
        }
-        node_listeners.emplace(socket_id, Listener {now, std::forward<Query>(query)});
+        node_listeners.emplace(socket_id, Listener {now, std::forward<Query>(query), version});
    }
    else
        l->second.refresh(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,7 +1474,8 @@ Dht::getNodesStats(sa_family_t af) const
        if (b.cached)
            stats.cached_nodes++;
    }
-    stats.table_depth = bcks.depth(bcks.findBucket(myid));
+    stats.table_depth = buckets.depth(buckets.findBucket(myid));
+    stats.searches = searches.size();
    return stats;
 }

@@ -1597,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;
@@ -1684,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();
 }

@@ -1722,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();
 }

@@ -1743,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),
@@ -1751,7 +1755,7 @@ Dht::Dht(std::unique_ptr<net::DatagramSocket>&& sock, const Config& config, cons
            std::bind(&Dht::onPing, this, _1),
            std::bind(&Dht::onFindNode, this, _1, _2, _3),
            std::bind(&Dht::onGetValues, this, _1, _2, _3, _4),
-            std::bind(&Dht::onListen, this, _1, _2, _3, _4, _5),
+            std::bind(&Dht::onListen, this, _1, _2, _3, _4, _5, _6),
            std::bind(&Dht::onAnnounce, this, _1, _2, _3, _4, _5),
            std::bind(&Dht::onRefresh, this, _1, _2, _3, _4)),
    persistPath(config.persist_path),
@@ -1764,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);
@@ -1964,13 +1968,39 @@ 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 {
+            auto ips = network_engine.getSocket()->resolve(boootstrap.first, boootstrap.second);
+            for (auto& ip : ips) {
+                if (ip.getPort() == 0)
+                    ip.setPort(net::DHT_DEFAULT_PORT);
+                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()
 {
@@ -1978,25 +2008,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
@@ -2081,25 +2111,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))
@@ -2123,7 +2153,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--;
@@ -2184,9 +2214,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;
 }

@@ -2205,8 +2235,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_)
@@ -2295,7 +2325,7 @@ void Dht::onGetValuesDone(const Sp<Node>& node,
 }

 net::RequestAnswer
-Dht::onListen(Sp<Node> node, const InfoHash& hash, const Blob& token, size_t socket_id, const Query& query)
+Dht::onListen(Sp<Node> node, const InfoHash& hash, const Blob& token, size_t socket_id, const Query& query, int version)
 {
    if (not hash) {
        if (logger_)
@@ -2311,7 +2341,7 @@ Dht::onListen(Sp<Node> node, const InfoHash& hash, const Blob& token, size_t soc
        throw net::DhtProtocolException {net::DhtProtocolException::UNAUTHORIZED, net::DhtProtocolException::LISTEN_WRONG_TOKEN};
    }
    Query q = query;
-    storageAddListener(hash, node, socket_id, std::move(q));
+    storageAddListener(hash, node, socket_id, std::move(q), version);
    return {};
 }

@@ -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,6 +475,10 @@ DhtRunner::getNodeInfo() const {
        info.node_id = dht->getNodeId();
        info.ipv4 = dht->getNodesStats(AF_INET);
        info.ipv6 = dht->getNodesStats(AF_INET6);
+        if (auto sock = dht->getSocket()) {
+            info.bound4 = sock->getBoundRef(AF_INET).getPort();
+            info.bound6 = sock->getBoundRef(AF_INET6).getPort();
+        }
    }
    info.ongoing_ops = ongoing_ops;
    return info;
@@ -496,6 +496,10 @@ 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);
+        if (auto sock = dht.getSocket()) {
+            info.bound4 = sock->getBoundRef(AF_INET).getPort();
+            info.bound6 = sock->getBoundRef(AF_INET6).getPort();
+        }
        info.ongoing_ops = ongoing_ops;
        cb(std::move(sinfo));
        opEnded();
@@ -584,7 +588,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()) {
@@ -639,20 +643,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);
    }
@@ -899,89 +894,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
@@ -30,8 +30,9 @@ namespace dht {
 struct Listener {
    time_point time;
    Query query;
+    int version;

-    Listener(time_point t, Query&& q) : time(t), query(std::move(q)) {}
+    Listener(time_point t, Query&& q, int version = 0) : time(t), query(std::move(q)), version(version) {}

    void refresh(time_point t, Query&& q) {
        time = t;
@@ -31,7 +31,8 @@ enum class MessageType {
    Refresh,
    Listen,
    ValueData,
-    ValueUpdate
+    ValueUpdate,
+    UpdateValue
 };

 } /* namespace net */
@@ -48,23 +48,6 @@ static constexpr uint8_t v4prefix[16] = {0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0xFF, 0xF
 constexpr unsigned SEND_NODES {8};


-/* Transaction-ids are 4-bytes long, with the first two bytes identifying
- * the kind of request, and the remaining two a sequence number in
- * host order.
- */
-struct TransId final : public std::array<uint8_t, 4> {
-    TransId() { std::fill(begin(), end(), 0); }
-    TransId(const std::array<char, 4>& o) { std::copy(o.begin(), o.end(), begin()); }
-    TransId(uint32_t id) {
-        *reinterpret_cast<uint32_t*>(data()) = htonl(id);
-    }
-
-    uint32_t toInt() const {
-        return ntohl(*reinterpret_cast<const uint32_t*>(&(*this)[0]));
-    }
-};
-
-
 struct NetworkEngine::PartialMessage {
    SockAddr from;
    time_point start;
@@ -140,11 +123,15 @@ void
 NetworkEngine::tellListener(Sp<Node> node, Tid socket_id, const InfoHash& hash, want_t want,
        const Blob& ntoken, std::vector<Sp<Node>>&& nodes,
        std::vector<Sp<Node>>&& nodes6, std::vector<Sp<Value>>&& values,
-        const Query& query)
+        const Query& query, int version)
 {
    auto nnodes = bufferNodes(node->getFamily(), hash, want, nodes, nodes6);
    try {
-        sendNodesValues(node->getAddr(), socket_id, nnodes.first, nnodes.second, values, query, ntoken);
+        if (version == 1) {
+            sendUpdateValues(node, hash, values, scheduler.time(), ntoken, socket_id);
+        } else {
+            sendNodesValues(node->getAddr(), socket_id, nnodes.first, nnodes.second, values, query, ntoken);
+        }
    } catch (const std::overflow_error& e) {
        if (logger_)
            logger_->e("Can't send value: buffer not large enough !");
@@ -281,6 +268,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] {
@@ -614,18 +604,28 @@ NetworkEngine::process(std::unique_ptr<ParsedMessage>&& msg, const SockAddr& fro
                if (logIncoming_ and logger_)
                    logger_->d(msg->info_hash, node->id, "[node %s] got 'listen' request for %s", node->toString().c_str(), msg->info_hash.toString().c_str());
                ++in_stats.listen;
-                RequestAnswer answer = onListen(node, msg->info_hash, msg->token, msg->socket_id, std::move(msg->query));
+                RequestAnswer answer = onListen(node, msg->info_hash, msg->token, msg->socket_id, std::move(msg->query), msg->version);
                auto nnodes = bufferNodes(from.getFamily(), msg->info_hash, msg->want, answer.nodes4, answer.nodes6);
                sendListenConfirmation(from, msg->tid);
                break;
            }
+            case MessageType::UpdateValue: {
+                if (logIncoming_ and logger_)
+                    logger_->d(msg->info_hash, node->id, "[node %s] got 'update' request for %s", node->toString().c_str(), msg->info_hash.toString().c_str());
+                ++in_stats.updateValue;
+                if (auto rsocket = node->getSocket(msg->socket_id))
+                    rsocket->on_receive(node, std::move(*msg));
+                else if (logger_)
+                    logger_->e(msg->info_hash, node->id, "[node %s] 'update' request without socket for %s", node->toString().c_str(), msg->info_hash.toString().c_str());
+                break;
+            }
            default:
                break;
            }
        } catch (const std::overflow_error& e) {
            if (logger_)
                logger_->e("Can't send value: buffer not large enough !");
-        } catch (DhtProtocolException& e) {
+        } catch (const DhtProtocolException& e) {
            sendError(from, msg->tid, e.getCode(), e.getMsg().c_str(), true);
        }
    }
@@ -651,7 +651,7 @@ NetworkEngine::send(const SockAddr& addr, const char *buf, size_t len, bool conf

 Sp<Request>
 NetworkEngine::sendPing(Sp<Node> node, RequestCb&& on_done, RequestExpiredCb&& on_expired) {
-    TransId tid (node->getNewTid());
+    Tid tid (node->getNewTid());
    msgpack::sbuffer buffer;
    msgpack::packer<msgpack::sbuffer> pk(&buffer);
    pk.pack_map(5+(config.network?1:0));
@@ -660,15 +660,14 @@ NetworkEngine::sendPing(Sp<Node> node, RequestCb&& on_done, RequestExpiredCb&& o
     pk.pack(KEY_REQ_ID); pk.pack(myid);

    pk.pack(KEY_Q); pk.pack(QUERY_PING);
-    pk.pack(KEY_TID); pk.pack_bin(tid.size());
-                              pk.pack_bin_body((const char*)tid.data(), tid.size());
+    pk.pack(KEY_TID); pk.pack(tid);
    pk.pack(KEY_Y); pk.pack(KEY_Q);
    pk.pack(KEY_UA); pk.pack(my_v);
    if (config.network) {
        pk.pack(KEY_NETID); pk.pack(config.network);
    }

-    auto req = std::make_shared<Request>(MessageType::Ping, tid.toInt(), node,
+    auto req = std::make_shared<Request>(MessageType::Ping, tid, node,
        Blob(buffer.data(), buffer.data() + buffer.size()),
        [=](const Request& req_status, ParsedMessage&&) {
            if (logger_)
@@ -698,9 +697,7 @@ NetworkEngine::sendPong(const SockAddr& addr, Tid tid) {
      pk.pack(KEY_REQ_ID); pk.pack(myid);
      insertAddr(pk, addr);

-    TransId t (tid);
-    pk.pack(KEY_TID); pk.pack_bin(t.size());
-                               pk.pack_bin_body((const char*)t.data(), t.size());
+    pk.pack(KEY_TID); pk.pack(tid);
    pk.pack(KEY_Y); pk.pack(KEY_R);
    pk.pack(KEY_UA); pk.pack(my_v);
    if (config.network) {
@@ -713,7 +710,7 @@ NetworkEngine::sendPong(const SockAddr& addr, Tid tid) {
 Sp<Request>
 NetworkEngine::sendFindNode(Sp<Node> n, const InfoHash& target, want_t want,
        RequestCb&& on_done, RequestExpiredCb&& on_expired) {
-    TransId tid (n->getNewTid());
+    Tid tid (n->getNewTid());
    msgpack::sbuffer buffer;
    msgpack::packer<msgpack::sbuffer> pk(&buffer);
    pk.pack_map(5+(config.network?1:0));
@@ -729,15 +726,14 @@ NetworkEngine::sendFindNode(Sp<Node> n, const InfoHash& target, want_t want,
    }

    pk.pack(KEY_Q); pk.pack(QUERY_FIND);
-    pk.pack(KEY_TID); pk.pack_bin(tid.size());
-                               pk.pack_bin_body((const char*)tid.data(), tid.size());
+    pk.pack(KEY_TID); pk.pack(tid);
    pk.pack(KEY_Y); pk.pack(KEY_Q);
    pk.pack(KEY_UA); pk.pack(my_v);
    if (config.network) {
        pk.pack(KEY_NETID); pk.pack(config.network);
    }

-    auto req = std::make_shared<Request>(MessageType::FindNode, tid.toInt(), n,
+    auto req = std::make_shared<Request>(MessageType::FindNode, tid, n,
        Blob(buffer.data(), buffer.data() + buffer.size()),
        [=](const Request& req_status, ParsedMessage&& msg) { /* on done */
            if (on_done) {
@@ -759,7 +755,7 @@ NetworkEngine::sendFindNode(Sp<Node> n, const InfoHash& target, want_t want,
 Sp<Request>
 NetworkEngine::sendGetValues(Sp<Node> n, const InfoHash& info_hash, const Query& query, want_t want,
        RequestCb&& on_done, RequestExpiredCb&& on_expired) {
-    TransId tid (n->getNewTid());
+    Tid tid (n->getNewTid());
    msgpack::sbuffer buffer;
    msgpack::packer<msgpack::sbuffer> pk(&buffer);
    pk.pack_map(5+(config.network?1:0));
@@ -783,15 +779,14 @@ NetworkEngine::sendGetValues(Sp<Node> n, const InfoHash& info_hash, const Query&
      }

    pk.pack(KEY_Q); pk.pack(QUERY_GET);
-    pk.pack(KEY_TID); pk.pack_bin(tid.size());
-                      pk.pack_bin_body((const char*)tid.data(), tid.size());
+    pk.pack(KEY_TID); pk.pack(tid);
    pk.pack(KEY_Y); pk.pack(KEY_Q);
    pk.pack(KEY_UA); pk.pack(my_v);
    if (config.network) {
        pk.pack(KEY_NETID); pk.pack(config.network);
    }

-    auto req = std::make_shared<Request>(MessageType::GetValues, tid.toInt(), n,
+    auto req = std::make_shared<Request>(MessageType::GetValues, tid, n,
        Blob(buffer.data(), buffer.data() + buffer.size()),
        [=](const Request& req_status, ParsedMessage&& msg) { /* on done */
            if (on_done) {
@@ -893,7 +888,7 @@ NetworkEngine::packValueHeader(msgpack::sbuffer& buffer, const std::vector<Sp<Va
 }

 void
-NetworkEngine::sendValueParts(const TransId& tid, const std::vector<Blob>& svals, const SockAddr& addr)
+NetworkEngine::sendValueParts(Tid tid, const std::vector<Blob>& svals, const SockAddr& addr)
 {
    msgpack::sbuffer buffer;
    unsigned i=0;
@@ -908,8 +903,7 @@ NetworkEngine::sendValueParts(const TransId& tid, const std::vector<Blob>& svals
                pk.pack(KEY_NETID); pk.pack(config.network);
            }
            pk.pack(KEY_Y); pk.pack(KEY_V);
-            pk.pack(KEY_TID); pk.pack_bin(tid.size());
-                                       pk.pack_bin_body((const char*)tid.data(), tid.size());
+            pk.pack(KEY_TID); pk.pack(tid);
            pk.pack(KEY_V); pk.pack_map(1);
                pk.pack(i); pk.pack_map(2);
                    pk.pack(std::string("o")); pk.pack(start);
@@ -964,9 +958,7 @@ NetworkEngine::sendNodesValues(const SockAddr& addr, Tid tid, const Blob& nodes,
        }
    }

-    TransId t (tid);
-    pk.pack(KEY_TID); pk.pack_bin(t.size());
-                      pk.pack_bin_body((const char*)t.data(), t.size());
+    pk.pack(KEY_TID); pk.pack(tid);
    pk.pack(KEY_Y); pk.pack(KEY_R);
    pk.pack(KEY_UA); pk.pack(my_v);
    if (config.network) {
@@ -1042,7 +1034,7 @@ NetworkEngine::sendListen(Sp<Node> n,
        SocketCb&& socket_cb)
 {
    Tid socket;
-    TransId tid (n->getNewTid());
+    Tid tid (n->getNewTid());
    if (previous and previous->node == n) {
        socket = previous->getSocket();
    } else {
@@ -1057,33 +1049,30 @@ NetworkEngine::sendListen(Sp<Node> n,
            logger_->e(hash, "[node %s] unable to get a valid socket for listen. Aborting listen", n->toString().c_str());
        return {};
    }
-    TransId sid(socket);
-
    msgpack::sbuffer buffer;
    msgpack::packer<msgpack::sbuffer> pk(&buffer);
    pk.pack_map(5+(config.network?1:0));

    auto has_query = not query.where.empty() or not query.select.empty();
-    pk.pack(KEY_A); pk.pack_map(4 + has_query);
+    pk.pack(KEY_A); pk.pack_map(5 + has_query);
      pk.pack(KEY_REQ_ID);    pk.pack(myid);
+      pk.pack(KEY_VERSION);   pk.pack(1);
      pk.pack(KEY_REQ_H);     pk.pack(hash);
      pk.pack(KEY_REQ_TOKEN); packToken(pk, token);
-      pk.pack(KEY_REQ_SID);   pk.pack_bin(sid.size());
-                              pk.pack_bin_body((const char*)sid.data(), sid.size());
+      pk.pack(KEY_REQ_SID);   pk.pack(socket);
      if (has_query) {
          pk.pack(KEY_REQ_QUERY); pk.pack(query);
      }

    pk.pack(KEY_Q); pk.pack(QUERY_LISTEN);
-    pk.pack(KEY_TID); pk.pack_bin(tid.size());
-                               pk.pack_bin_body((const char*)tid.data(), tid.size());
+    pk.pack(KEY_TID); pk.pack(tid);
    pk.pack(KEY_Y); pk.pack(KEY_Q);
    pk.pack(KEY_UA); pk.pack(my_v);
    if (config.network) {
        pk.pack(KEY_NETID); pk.pack(config.network);
    }

-    auto req = std::make_shared<Request>(MessageType::Listen, tid.toInt(), n,
+    auto req = std::make_shared<Request>(MessageType::Listen, tid, n,
        Blob(buffer.data(), buffer.data() + buffer.size()),
        [=](const Request& req_status, ParsedMessage&& msg) { /* on done */
            if (on_done)
@@ -1110,9 +1099,7 @@ NetworkEngine::sendListenConfirmation(const SockAddr& addr, Tid tid) {
      pk.pack(KEY_REQ_ID); pk.pack(myid);
      insertAddr(pk, addr);

-    TransId t (tid);
-    pk.pack(KEY_TID); pk.pack_bin(t.size());
-                               pk.pack_bin_body((const char*)t.data(), t.size());
+    pk.pack(KEY_TID); pk.pack(tid);
    pk.pack(KEY_Y); pk.pack(KEY_R);
    pk.pack(KEY_UA); pk.pack(my_v);
    if (config.network) {
@@ -1131,7 +1118,7 @@ NetworkEngine::sendAnnounceValue(Sp<Node> n,
        RequestCb&& on_done,
        RequestExpiredCb&& on_expired)
 {
-    TransId tid (n->getNewTid());
+    Tid tid (n->getNewTid());
    msgpack::sbuffer buffer;
    msgpack::packer<msgpack::sbuffer> pk(&buffer);
    pk.pack_map(5+(config.network?1:0));
@@ -1147,15 +1134,14 @@ NetworkEngine::sendAnnounceValue(Sp<Node> n,
      pk.pack(KEY_REQ_TOKEN);  pk.pack(token);

    pk.pack(KEY_Q);   pk.pack(QUERY_PUT);
-    pk.pack(KEY_TID); pk.pack_bin(tid.size());
-                      pk.pack_bin_body((const char*)tid.data(), tid.size());
+    pk.pack(KEY_TID); pk.pack(tid);
    pk.pack(KEY_Y);   pk.pack(KEY_Q);
    pk.pack(KEY_UA);  pk.pack(my_v);
    if (config.network) {
        pk.pack(KEY_NETID); pk.pack(config.network);
    }

-    auto req = std::make_shared<Request>(MessageType::AnnounceValue, tid.toInt(), n,
+    auto req = std::make_shared<Request>(MessageType::AnnounceValue, tid, n,
        Blob(buffer.data(), buffer.data() + buffer.size()),
        [=](const Request& req_status, ParsedMessage&& msg) { /* on done */
            if (msg.value_id == Value::INVALID_ID) {
@@ -1175,13 +1161,58 @@ 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;
 }

+Sp<Request>
+NetworkEngine::sendUpdateValues(Sp<Node> n,
+                                const InfoHash& infohash,
+                                const std::vector<Sp<Value>>& values,
+                                time_point created,
+                                const Blob& token,
+                                const size_t& socket_id)
+{
+    Tid tid (n->getNewTid());
+    Tid sid (socket_id);
+
+    msgpack::sbuffer buffer;
+    msgpack::packer<msgpack::sbuffer> pk(&buffer);
+    pk.pack_map(5+(config.network?1:0));
+
+    pk.pack(KEY_A); pk.pack_map((created < scheduler.time() ? 7 : 6));
+      pk.pack(KEY_REQ_ID);     pk.pack(myid);
+      pk.pack(KEY_VERSION);    pk.pack(1);
+      pk.pack(KEY_REQ_H);      pk.pack(infohash);
+      pk.pack(KEY_REQ_SID);   pk.pack(sid);
+      auto v = packValueHeader(buffer, values);
+      if (created < scheduler.time()) {
+          pk.pack(KEY_REQ_CREATION);
+          pk.pack(to_time_t(created));
+      }
+      pk.pack(KEY_REQ_TOKEN);  pk.pack(token);
+
+    pk.pack(KEY_Q);   pk.pack(QUERY_UPDATE);
+    pk.pack(KEY_TID); pk.pack(tid);
+    pk.pack(KEY_Y);   pk.pack(KEY_Q);
+    pk.pack(KEY_UA);  pk.pack(my_v);
+    if (config.network) {
+        pk.pack(KEY_NETID); pk.pack(config.network);
+    }
+
+    auto req = std::make_shared<Request>(MessageType::UpdateValue, tid, n,
+        Blob(buffer.data(), buffer.data() + buffer.size()),
+        [=](const Request&, ParsedMessage&&) { /* on done */ },
+        [=](const Request&, bool) { /* on expired */ }
+    );
+    req->parts = std::move(v);
+    sendRequest(req);
+    ++out_stats.updateValue;
+    return req;
+}
+
 Sp<Request>
 NetworkEngine::sendRefreshValue(Sp<Node> n,
                const InfoHash& infohash,
@@ -1190,7 +1221,7 @@ NetworkEngine::sendRefreshValue(Sp<Node> n,
                RequestCb&& on_done,
                RequestExpiredCb&& on_expired)
 {
-    TransId tid (n->getNewTid());
+    Tid tid (n->getNewTid());
    msgpack::sbuffer buffer;
    msgpack::packer<msgpack::sbuffer> pk(&buffer);
    pk.pack_map(5+(config.network?1:0));
@@ -1202,15 +1233,14 @@ NetworkEngine::sendRefreshValue(Sp<Node> n,
      pk.pack(KEY_REQ_TOKEN);    pk.pack(token);

    pk.pack(KEY_Q); pk.pack(QUERY_REFRESH);
-    pk.pack(KEY_TID); pk.pack_bin(tid.size());
-                               pk.pack_bin_body((const char*)tid.data(), tid.size());
+    pk.pack(KEY_TID); pk.pack(tid);
    pk.pack(KEY_Y); pk.pack(KEY_Q);
    pk.pack(KEY_UA); pk.pack(my_v);
    if (config.network) {
        pk.pack(KEY_NETID); pk.pack(config.network);
    }

-    auto req = std::make_shared<Request>(MessageType::Refresh, tid.toInt(), n,
+    auto req = std::make_shared<Request>(MessageType::Refresh, tid, n,
        Blob(buffer.data(), buffer.data() + buffer.size()),
        [=](const Request& req_status, ParsedMessage&& msg) { /* on done */
            if (msg.value_id == Value::INVALID_ID) {
@@ -1246,9 +1276,7 @@ NetworkEngine::sendValueAnnounced(const SockAddr& addr, Tid tid, Value::Id vid)
      pk.pack(KEY_REQ_VALUE_ID); pk.pack(vid);
      insertAddr(pk, addr);

-    TransId t(tid);
-    pk.pack(KEY_TID); pk.pack_bin(t.size());
-                               pk.pack_bin_body((const char*)t.data(), t.size());
+    pk.pack(KEY_TID); pk.pack(tid);
    pk.pack(KEY_Y); pk.pack(KEY_R);
    pk.pack(KEY_UA); pk.pack(my_v);
    if (config.network) {
@@ -1278,9 +1306,7 @@ NetworkEngine::sendError(const SockAddr& addr,
          pk.pack(KEY_REQ_ID); pk.pack(myid);
    }

-    TransId t(tid);
-    pk.pack(KEY_TID); pk.pack_bin(t.size());
-                               pk.pack_bin_body((const char*)t.data(), t.size());
+    pk.pack(KEY_TID); pk.pack(tid);
    pk.pack(KEY_Y); pk.pack(KEY_E);
    pk.pack(KEY_UA); pk.pack(my_v);
    if (config.network) {
@@ -54,10 +54,12 @@ static const std::string KEY_REQ_EXPIRED {"exp"};
 static const std::string KEY_REQ_REFRESHED {"re"};
 static const std::string KEY_REQ_FIELDS {"fileds"};
 static const std::string KEY_REQ_WANT {"w"};
+static const std::string KEY_VERSION {"ve"};

 static const std::string QUERY_PING {"ping"};
 static const std::string QUERY_FIND {"find"};
 static const std::string QUERY_GET {"get"};
+static const std::string QUERY_UPDATE {"update"};
 static const std::string QUERY_PUT {"put"};
 static const std::string QUERY_LISTEN {"listen"};
 static const std::string QUERY_REFRESH {"refresh"};
@@ -113,6 +115,7 @@ struct ParsedMessage {
    uint16_t error_code;
    /* reported address by the distant node */
    std::string ua;
+    int version {0};
    SockAddr addr;
    void msgpack_unpack(const msgpack::object& o);

@@ -224,6 +227,8 @@ ParsedMessage::msgpack_unpack(const msgpack::object& msg)
        type = MessageType::AnnounceValue;
    else if (parsed.q == QUERY_REFRESH)
        type = MessageType::Refresh;
+    else if (parsed.q == QUERY_UPDATE)
+        type = MessageType::UpdateValue;
    else
        throw msgpack::type_error();

@@ -295,6 +300,8 @@ ParsedMessage::msgpack_unpack(const msgpack::object& msg)
            parsedReq.fields = &o.val;
        else if (key == KEY_REQ_WANT)
            parsedReq.want = &o.val;
+        else if (key == KEY_VERSION)
+            version = o.val.as<int>();
    }

    if (parsedReq.sa) {
@@ -115,6 +115,7 @@ private:
        on_done = {};
        on_expired = {};
        msg = {};
+        parts = {};
    }

    const Tid tid {0}; /* the request id. */
@@ -130,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. */
 };

@@ -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;
@@ -76,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;
 }
@@ -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";
            }
        }
    }
@@ -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
@@ -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,88 @@ 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);
+}
+
+void
+DhtRunnerTester::testListenLotOfBytes() {
+    std::mutex mutex;
+    std::condition_variable cv;
+    std::atomic_uint valueCount(0);
+    unsigned putCount(0);
+    unsigned putOkCount(0);
+
+    std::string data(10000, 'a');
+
+    auto foo = dht::InfoHash::get("foo");
+    constexpr unsigned N = 50;
+
+    for (unsigned i=0; i<N; i++) {
+        node2.put(foo, data, [&](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; }));
+    }
+
+    dht::DhtRunner node3 {};
+    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;
+    node3.run(42242, config);
+    node3.bootstrap(node1.getBound());
+
+    auto ftokenfoo = node3.listen(foo, [&](const std::shared_ptr<dht::Value>&) {
+        valueCount++;
+        cv.notify_all();
+        return true;
+    });
+
+    {
+        std::unique_lock<std::mutex> lk(mutex);
+        CPPUNIT_ASSERT(cv.wait_for(lk, 30s, [&]{ return valueCount == N; }));
+    }
+
+    node3.cancelListen(foo, ftokenfoo.get());
 }

 }  // namespace test
@@ -32,6 +32,7 @@ class DhtRunnerTester : public CppUnit::TestFixture {
    CPPUNIT_TEST(testConstructors);
    CPPUNIT_TEST(testGetPut);
    CPPUNIT_TEST(testListen);
+    CPPUNIT_TEST(testListenLotOfBytes);
    CPPUNIT_TEST_SUITE_END();

    dht::DhtRunner node1 {};
@@ -57,6 +58,10 @@ class DhtRunnerTester : public CppUnit::TestFixture {
     * Test listen method
     */
    void testListen();
+    /**
+     * Test listen method with lot of datas
+     */
+    void testListenLotOfBytes();
 };

 }  // namespace test
@@ -64,10 +64,10 @@ main(int argc, char **argv)
        auto dhtConf = getDhtConfig(params);
        dht.run(params.port, dhtConf.first, std::move(dhtConf.second));

-        if (not params.bootstrap.first.empty())
-            dht.bootstrap(params.bootstrap.first.c_str(), params.bootstrap.second.c_str());
+        if (not params.bootstrap.empty())
+            dht.bootstrap(params.bootstrap);

-        print_node_info(dht, params);
+        print_node_info(dht.getNodeInfo());
        std::cout << "  type 'c {hash}' to join a channel" << std::endl << std::endl;

        bool connected {false};
@@ -114,7 +114,7 @@ void cmd_loop(std::shared_ptr<DhtRunner>& node, dht_params& params
 #endif
 )
 {
-    print_node_info(*node, params);
+    print_node_info(node->getNodeInfo());
    std::cout << " (type 'h' or 'help' for a list of possible commands)" << std::endl << std::endl;

 #ifndef WIN32_NATIVE
@@ -143,22 +143,23 @@ void cmd_loop(std::shared_ptr<DhtRunner>& node, dht_params& params
            print_help();
            continue;
        } else if (op == "ll") {
-            auto nodeInfo = node->getNodeInfo();
-            print_node_info(*node, params);
-            std::cout << nodeInfo.ongoing_ops << " ongoing operations" << std::endl;
-            std::cout << "IPv4 stats:" << std::endl;
-            std::cout << nodeInfo.ipv4.toString() << std::endl;
-            std::cout << "IPv6 stats:" << std::endl;
-            std::cout << nodeInfo.ipv6.toString() << std::endl;
+            node->getNodeInfo([&](const std::shared_ptr<dht::NodeInfo>& nodeInfo) {
+                print_node_info(*nodeInfo);
+                std::cout << nodeInfo->ongoing_ops << " ongoing operations" << std::endl;
+                std::cout << "IPv4 stats:" << std::endl;
+                std::cout << nodeInfo->ipv4.toString() << std::endl;
+                std::cout << "IPv6 stats:" << std::endl;
+                std::cout << nodeInfo->ipv6.toString() << std::endl;
 #ifdef OPENDHT_PROXY_SERVER
-            for (const auto& proxy : proxies) {
-                std::cout << "Stats for proxy server on port " << proxy.first << std::endl;
-                if (auto stats = proxy.second->stats())
-                    std::cout << "  " << stats->toString() << std::endl;
-                else
-                    std::cout << "  (stats not available yet)" << std::endl;                
-            }
+                for (const auto& proxy : proxies) {
+                    std::cout << "Stats for proxy server on port " << proxy.first << std::endl;
+                    if (auto stats = proxy.second->stats())
+                        std::cout << "  " << stats->toString() << std::endl;
+                    else
+                        std::cout << "  (stats not available yet)" << std::endl;                
+                }
 #endif
+            });
            continue;
        } else if (op == "lr") {
            std::cout << "IPv4 routing table:" << std::endl;
@@ -190,14 +191,10 @@ void cmd_loop(std::shared_ptr<DhtRunner>& node, dht_params& params
            continue;
        } else if (op == "b") {
            iss >> idstr;
-            try {
-                auto addr = splitPort(idstr);
-                if (not addr.first.empty() and addr.second.empty())
-                    addr.second = std::to_string(dht::net::DHT_DEFAULT_PORT);
-                node->bootstrap(addr.first.c_str(), addr.second.c_str());
-            } catch (const std::exception& e) {
-                std::cerr << e.what() << std::endl;
-            }
+            if (not idstr.empty())
+                node->bootstrap(idstr);
+            else
+                std::cerr << "No service provided. Syntax: b hostname:port" << std::endl;
            continue;
        } else if (op == "log") {
            iss >> idstr;
@@ -537,9 +534,9 @@ main(int argc, char **argv)
        auto dhtConf = getDhtConfig(params);
        node->run(params.port, dhtConf.first, std::move(dhtConf.second));

-        if (not params.bootstrap.first.empty()) {
-            std::cout << "Bootstrap: " << params.bootstrap.first << ":" << params.bootstrap.second << std::endl;
-            node->bootstrap(params.bootstrap.first.c_str(), params.bootstrap.second.c_str());
+        if (not params.bootstrap.empty()) {
+            std::cout << "Bootstrap: " << params.bootstrap << std::endl;
+            node->bootstrap(params.bootstrap);
        }

 #ifdef OPENDHT_PROXY_SERVER
@@ -91,10 +91,10 @@ main(int argc, char **argv)
        auto dhtConf = getDhtConfig(params);
        dht.run(params.port, dhtConf.first, std::move(dhtConf.second));

-        if (not params.bootstrap.first.empty())
-            dht.bootstrap(params.bootstrap.first.c_str(), params.bootstrap.second.c_str());
+        if (not params.bootstrap.empty())
+            dht.bootstrap(params.bootstrap);

-        print_node_info(dht, params);
+        print_node_info(dht.getNodeInfo());
        std::cout << "Scanning network..." << std::endl;
        auto all_nodes = std::make_shared<NodeSet>();

@@ -124,7 +124,7 @@ struct dht_params {
    bool log {false};
    bool syslog {false};
    std::string logfile {};
-    std::pair<std::string, std::string> bootstrap {};
+    std::string bootstrap {};
    dht::NetId network {0};
    in_port_t port {0};
    in_port_t proxyserver {0};
@@ -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 {};
@@ -189,16 +191,14 @@ getDhtConfig(dht_params& params)
    return {std::move(config), std::move(context)};
 }

-void print_node_info(const dht::DhtRunner& node, const dht_params& params) {
-    std::cout << "OpenDHT node " << node.getNodeId() << " running on ";
-    auto port4 = node.getBoundPort(AF_INET);
-    auto port6 = node.getBoundPort(AF_INET6);
-    if (port4 == port6)
-        std::cout << "port " << port4 << std::endl;
+void print_node_info(const dht::NodeInfo& info) {
+    std::cout << "OpenDHT node " << info.node_id << " running on ";
+    if (info.bound4 == info.bound6)
+        std::cout << "port " << info.bound4 << std::endl;
    else
-        std::cout << "IPv4 port " << port4 << ", IPv6 port " << port6 << std::endl;
-    if (params.id.first)
-        std::cout << "Public key ID " << node.getId() << std::endl;
+        std::cout << "IPv4 port " << info.bound4 << ", IPv6 port " << info.bound6 << std::endl;
+    if (info.id)
+        std::cout << "Public key ID " << info.id << std::endl;
 }

 static const constexpr struct option long_options[] = {
@@ -289,10 +289,7 @@ parseArgs(int argc, char **argv) {
            params.public_stable = true;
            break;
        case 'b':
-            params.bootstrap = dht::splitPort((optarg[0] == '=') ? optarg+1 : optarg);
-            if (not params.bootstrap.first.empty() and params.bootstrap.second.empty()) {
-                params.bootstrap.second = std::to_string(dht::net::DHT_DEFAULT_PORT);
-            }
+            params.bootstrap = (optarg[0] == '=') ? optarg+1 : optarg;
            break;
        case 'V':
            params.version = true;
Author	SHA1	Message	Date
Adrien Béraud	c754b5cfce	update to version 2.1.0	2020-04-30 14:58:25 -04:00
Adrien Béraud	b0c1249710	network engine: check for logger	2020-04-29 20:23:19 -04:00
Adrien Béraud	c41ee417fb	tools: cleanup	2020-04-29 19:22:24 -04:00
Adrien Béraud	47ba0415ca	tools: use bootstrap split/default port from dht	2020-04-29 17:12:11 -04:00
Adrien Béraud	99857e25cc	dhtnode: use async version of getNodeInfo	2020-04-29 16:13:36 -04:00
Adrien Béraud	7f832ba9d5	NodeInfo: add bound ports	2020-04-29 16:00:02 -04:00
Sébastien Blin	8c4a894f93	misc: replace TransId to Tid	2020-04-28 13:55:29 -04:00
Sébastien Blin	dd3de412fb	dht: move updateValues logic into network_engine	2020-04-28 13:55:29 -04:00
Sébastien Blin	4ad5e6c9ba	misc: clean	2020-04-28 13:55:29 -04:00
Sébastien Blin	b445818521	tests: enable test for listening with lot of values	2020-04-28 13:55:29 -04:00
Sébastien Blin	7c14ea1730	network_engine: fix pack for sendUpdateValues	2020-04-28 13:55:29 -04:00
Sébastien Blin	7bf9f861a5	dht: send update values for listeners	2020-04-28 13:55:29 -04:00
Sébastien Blin	4cf5c87761	network_engin: move to pack(socket)	2020-04-28 13:55:29 -04:00
Sébastien Blin	e1c6fec581	dht: starts to add update and version support	2020-04-28 13:55:29 -04:00
Sébastien Blin	31da5b91fa	[WIP]net: add updateValues to replace Listen operations Actually, packets are not retransmitted for listen operations. This leads to some problems when they are already several values present at one hash. Ideally, packets should be transmitted inside a request to avoid packet loss and get all datas. This patch breaks the compatibility with previous versions	2020-04-28 13:55:29 -04:00
Adrien Béraud	01bcb1a750	network engine: completely remove TransId	2020-04-24 18:46:46 -04:00
Adrien Béraud	35348bf77f	dht: use default port for bootstraping	2020-04-24 17:45:40 -04:00
Adrien Béraud	ae58ea8312	search: expire only after candidate expiration	2020-04-24 16:32:07 -04:00
Adrien Béraud	f30d21e96f	move bootstrap logic from DhtRunner to Dht	2020-04-24 16:32:07 -04:00
Vladimir Stoiakin	7628aab230	opendht.pc: change -lpthread to -pthread	2020-04-19 14:13:56 -04:00
Adrien Béraud	f4cba56b3c	dht: optimize getStatus()	2020-04-17 02:09:57 -04:00
Adrien Béraud	7155dc4fb7	stats: add search, cached node count	2020-04-16 02:08:00 -04:00
Adrien Béraud	c961e2e99f	dht: allow to configure max searches & store size	2020-04-16 02:08:00 -04:00
Adrien Béraud	2fb8e1c7ee	Update README.md	2020-04-12 15:56:23 -04:00
Vladimir Stoiakin	4d689ef51a	update opendht.pc	2020-04-10 09:24:31 -04:00
Adrien Béraud	face9f12b2	request: cleanup	2020-03-30 18:39:28 -04:00
Adrien Béraud	c92f553a24	network engine: retransmit value parts	2020-03-30 17:54:50 -04:00
Adrien Béraud	8564cf3b86	tests/dhtrunner: test put/listen for large values	2020-03-30 17:54:47 -04:00
Adrien Béraud	e7c810d5d6	value: don't print large values	2020-03-30 17:54:42 -04:00
Adrien Béraud	e76c80b8be	tests: disable rate limiting	2020-03-30 17:54:34 -04:00