dhtnode: print metrics

Provides info about internal dht structures

CMakeLists.txt

@@ -1,4 +1,7 @@
 cmake_minimum_required (VERSION 3.1)
+include(CMakePackageConfigHelpers)
+include(CMakeDependentOption)
+
 project (opendht)
 set (opendht_VERSION_MAJOR 2)
 set (opendht_VERSION_MINOR 0.0)
@@ -22,6 +25,7 @@ option (OPENDHT_PROXY_SERVER_IDENTITY "Allow clients to use the node identity" O
 option (OPENDHT_PROXY_CLIENT "Enable DHT proxy client, use Restinio and jsoncpp" OFF)
 option (OPENDHT_PROXY_OPENSSL "Build DHT proxy with OpenSSL" ON)
 option (OPENDHT_PROXY_HTTP_PARSER_FORK "Build DHT proxy with custom http_parser to support old API" OFF)
+CMAKE_DEPENDENT_OPTION(OPENDHT_HTTP "Build embedded http(s) client" OFF "NOT OPENDHT_PROXY_SERVER;NOT OPENDHT_PROXY_CLIENT" ON)
 option (OPENDHT_PEER_DISCOVERY "Enable multicast peer discovery" ON)
 option (OPENDHT_INDEX "Build DHT indexation feature" OFF)
 option (OPENDHT_TESTS "Add unit tests executable" OFF)
@@ -61,7 +65,7 @@ if (Jsoncpp_FOUND)
     )
 endif()
 
-if (OPENDHT_PROXY_SERVER OR OPENDHT_PROXY_CLIENT)
+if (OPENDHT_HTTP)
     find_package(Restinio REQUIRED)
     if (Restinio_FOUND)
         find_library(FMT_LIBRARY fmt)
@@ -88,7 +92,7 @@ if (OPENDHT_PROXY_SERVER OR OPENDHT_PROXY_CLIENT)
 else()
     set(OPENDHT_PROXY_OPENSSL OFF)
 endif()
-if (OPENDHT_PROXY_SERVER OR OPENDHT_PROXY_CLIENT OR OPENDHT_PEER_DISCOVERY)
+if (OPENDHT_HTTP OR OPENDHT_PEER_DISCOVERY)
     add_definitions(-DASIO_STANDALONE)
 endif()
 
@@ -251,7 +255,7 @@ if (OPENDHT_PROXY_CLIENT)
   )
 endif ()
 
-if (OPENDHT_PROXY_SERVER OR OPENDHT_PROXY_CLIENT)
+if (OPENDHT_HTTP)
   if (OPENDHT_PUSH_NOTIFICATIONS)
     message("Using push notification")
     add_definitions(-DOPENDHT_PUSH_NOTIFICATIONS)
@@ -357,7 +361,6 @@ if (OPENDHT_PYTHON)
 endif ()
 
 # CMake module
-include(CMakePackageConfigHelpers)
 write_basic_package_version_file("${CMAKE_CURRENT_BINARY_DIR}/opendhtConfigVersion.cmake"
   VERSION ${opendht_VERSION}
   COMPATIBILITY AnyNewerVersion

README.md

@@ -14,8 +14,8 @@ Values are distributed over the network, with redundancy.
  * High resilience to network disruption
  * Public key cryptography layer providing optional data signature and encryption (using GnuTLS)
  * IPv4 and IPv6 support
- * Clean and powerful C++14 map API
- * Python 3 bindings
+ * Clean and powerful **C++14** map API
+ * Bindings for **C, Rust & Python 3**
  * REST API with optional HTTP client+server with push notification support
 
 ## Documentation

include/opendht/callbacks.h

@@ -58,6 +58,8 @@ struct OPENDHT_PUBLIC NodeStats {
              cached_nodes {0},
              incoming_nodes {0};
     unsigned table_depth {0};
+    unsigned cachedNodes {0};
+    unsigned searches {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,25 @@ 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)
+};
+
+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
 };
 
 struct OPENDHT_PUBLIC NodeInfo {
@@ -80,6 +100,7 @@ struct OPENDHT_PUBLIC NodeInfo {
     NodeStats ipv4 {};
     NodeStats ipv6 {};
     size_t ongoing_ops {0};
+    NodeMetrics metrics {};
 
 #ifdef OPENDHT_JSONCPP
     /**
@@ -137,6 +158,12 @@ struct OPENDHT_PUBLIC SecureDhtConfig
 {
     Config node_config {};
     crypto::Identity id {};
+
+    /** 
+     * Cache all encountered public keys and certificates,
+     * for use by the certificate store, putEncrypted and putSigned
+     */
+    bool cert_cache_all {false};
 };
 
 static constexpr size_t DEFAULT_STORAGE_LIMIT {1024 * 1024 * 64};

include/opendht/dht.h

@@ -273,6 +273,7 @@ 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;

include/opendht/dht_interface.h

@@ -194,6 +194,7 @@ 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/http.h

@@ -206,7 +206,7 @@ class Request;
 
 struct Response
 {
-    unsigned int status_code {0};
+    unsigned status_code {0};
     std::map<std::string, std::string> headers;
     std::string body;
     bool aborted {false};
@@ -223,10 +223,10 @@ public:
         RECEIVING,
         DONE
     };
-    using OnStatusCb = std::function<void(unsigned int status_code)>;
+    using OnStatusCb = std::function<void(unsigned status_code)>;
     using OnDataCb = std::function<void(const char* at, size_t length)>;
     using OnStateChangeCb = std::function<void(State state, const Response& response)>;
-    using OnJsonCb = std::function<void(Json::Value value, unsigned int status_code)>;
+    using OnJsonCb = std::function<void(Json::Value value, unsigned status_code)>;
     using OnDoneCb = std::function<void(const Response& response)>;
 
     // resolves implicitly
@@ -253,6 +253,12 @@ public:
     inline const Url& get_url() const {
         return resolver_->get_url();
     };
+
+    /** The previous request in case of redirect following */
+    std::shared_ptr<Request> getPrevious() const {
+        return prev_.lock();
+    }
+
     inline std::string& to_string() {
         return request_;
     }
@@ -347,6 +353,12 @@ private:
     std::atomic<bool> finishing_ {false};
     std::unique_ptr<http_parser> parser_;
     std::unique_ptr<http_parser_settings> parser_s_;
+
+    // Next request in case of redirect following
+    std::shared_ptr<Request> next_;
+    std::weak_ptr<Request> prev_;
+    unsigned num_redirect {0};
+    bool follow_redirect {true};
 };
 
 } // namespace http

include/opendht/network_engine.h

@@ -275,7 +275,8 @@ public:
      * @return the request with information concerning its success.
      */
     Sp<Request>
-        sendPing(Sp<Node> n, RequestCb&& on_done, RequestExpiredCb&& on_expired);
+    sendPing(Sp<Node> n, RequestCb&& on_done, RequestExpiredCb&& on_expired);
+    
     /**
      * Send a "ping" request to a given node.
      *
@@ -287,11 +288,12 @@ public:
      * @return the request with information concerning its success.
      */
     Sp<Request>
-        sendPing(SockAddr&& sa, RequestCb&& on_done, RequestExpiredCb&& on_expired) {
-            return sendPing(std::make_shared<Node>(zeroes, std::move(sa)),
-                    std::forward<RequestCb>(on_done),
-                    std::forward<RequestExpiredCb>(on_expired));
-        }
+    sendPing(SockAddr&& sa, RequestCb&& on_done, RequestExpiredCb&& on_expired) {
+        return sendPing(std::make_shared<Node>(zeroes, std::move(sa), rd),
+                std::forward<RequestCb>(on_done),
+                std::forward<RequestExpiredCb>(on_expired));
+    }
+    
     /**
      * Send a "find node" request to a given node.
      *
@@ -430,6 +432,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;
@@ -529,7 +546,7 @@ private:
     Sp<Logger> logger_;
     std::mt19937_64& rd;
 
-    NodeCache cache {};
+    NodeCache cache;
 
     // global limiting should be triggered by at least 8 different IPs
     using IpLimiter = RateLimiter;

include/opendht/node.h

@@ -47,10 +47,10 @@ struct Socket {
 struct Node {
     const InfoHash id;
 
-    Node(const InfoHash& id, const SockAddr& addr, bool client=false);
-    Node(const InfoHash& id, SockAddr&& addr, bool client=false);
-    Node(const InfoHash& id, const sockaddr* sa, socklen_t salen)
-        : Node(id, SockAddr(sa, salen)) {}
+    Node(const InfoHash& id, const SockAddr& addr, std::mt19937_64& rd, bool client=false);
+    Node(const InfoHash& id, SockAddr&& addr, std::mt19937_64& rd, bool client=false);
+    Node(const InfoHash& id, const sockaddr* sa, socklen_t salen, std::mt19937_64& rd)
+        : Node(id, SockAddr(sa, salen), rd) {}
 
     InfoHash getId() const {
         return id;

include/opendht/node_cache.h

@@ -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;
@@ -37,17 +44,19 @@ struct NodeCache {
      */
     void clearBadNodes(sa_family_t family = 0);
 
+    NodeCache(std::mt19937_64& r) : rd(r) {};
     ~NodeCache();
 
 private:
     class NodeMap : private std::map<InfoHash, std::weak_ptr<Node>> {
     public:
         Sp<Node> getNode(const InfoHash& id);
-        Sp<Node> getNode(const InfoHash& id, const SockAddr&, time_point now, bool confirmed, bool client);
+        Sp<Node> getNode(const InfoHash& id, const SockAddr&, time_point now, bool confirmed, bool client, std::mt19937_64& rd);
         std::vector<Sp<Node>> getCachedNodes(const InfoHash& id, size_t count) const;
         void clearBadNodes();
         void setExpired();
         void cleanup();
+        size_t count() const { return size(); }
     private:
         size_t cleanup_counter {0};
     };
@@ -56,6 +65,7 @@ private:
     NodeMap& cache(sa_family_t af) { return af == AF_INET ? cache_4 : cache_6; }
     NodeMap cache_4;
     NodeMap cache_6;
+    std::mt19937_64& rd;
 };
 
 }

include/opendht/routing_table.h

@@ -39,7 +39,7 @@ struct Bucket {
     Sp<Node> cached;                    /* the address of a likely candidate */
 
     /** Return a random node in a bucket. */
-    Sp<Node> randomNode();
+    Sp<Node> randomNode(std::mt19937_64& rd);
 
     void sendCachedPing(net::NetworkEngine& ne);
     void connectivityChanged() {
@@ -89,7 +89,7 @@ public:
     /**
      * Return a random id in the bucket's range.
      */
-    InfoHash randomId(const RoutingTable::const_iterator& bucket) const;
+    InfoHash randomId(const RoutingTable::const_iterator& bucket, std::mt19937_64& rd) const;
 
     unsigned depth(const RoutingTable::const_iterator& bucket) const;
 

include/opendht/scheduler.h

@@ -84,7 +84,6 @@ public:
      * @return The time for the next job to run.
      */
     time_point run() {
-        syncTime();
         while (not timers.empty()) {
             auto timer = timers.begin();
             /*
@@ -116,6 +115,8 @@ 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,6 +180,9 @@ 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);
     }
@@ -350,6 +353,7 @@ private:
     std::map<InfoHash, Sp<const crypto::PublicKey>> nodesPubKeys_ {};
 
     std::atomic_bool forward_all_ {false};
+    bool enableCache_ {false};
 };
 
 const ValueType CERTIFICATE_TYPE = {

include/opendht/utils.h

@@ -110,8 +110,12 @@ print_duration(DT d) {
         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";
+    } else if (d < std::chrono::minutes(1)) {
+        return to_str(std::chrono::duration_cast<std::chrono::duration<double>>(d).count()) +  " s";
+    } else if (d < std::chrono::hours(1)) {
+        return to_str(std::chrono::duration_cast<std::chrono::duration<double, std::ratio<60>>>(d).count()) +  " min";
     } else {
-        return to_str(print_dt(d)) + " s";
+        return to_str(std::chrono::duration_cast<std::chrono::duration<double, std::ratio<3600>>>(d).count()) +  " h";
     }
 }
 

python/tools/pingpong.py

@@ -35,13 +35,21 @@ pong_node.ping(ping_node.getBound())
 loc_ping = dht.InfoHash.get("toto99")
 loc_pong = dht.InfoHash.get(str(loc_ping))
 
+net = []
+for i in range(1,10):
+	node = dht.DhtRunner()
+	node.run(config=config)
+	node.ping(ping_node.getBound())
+	net.append(node)
+
 i = 0
 MAX = 2048
 
 loop = asyncio.get_event_loop()
 
 def done(h, ok):
-	print(h, "over", ok)
+	pass
+	#print(h, "over", ok)
 
 def ping(node, h):
 	global i
@@ -52,13 +60,20 @@ def ping(node, h):
 		loop.stop()
 
 def pong(node, h):
-	print(node.getNodeId().decode(), "got ping", h, i)
+	#print(node.getNodeId().decode(), "got ping", h, i)
 	loop.call_soon_threadsafe(ping, node, h)
 	return True
 
+t1 = time.time()
+
 ping_node.listen(loc_ping, lambda v, e: pong(pong_node, loc_pong) if not e else True)
 pong_node.listen(loc_pong, lambda v, e: pong(ping_node, loc_ping) if not e else True)
 
 ping(pong_node, loc_ping)
 
 loop.run_forever()
+
+t2 = time.time()
+
+print(MAX, "ping-pong done, took", t2 - t1, "s")
+print(1000 * (t2 - t1)/MAX, "ms per rt", MAX/(t2 - t1), "rt per s")

src/callbacks.cpp

@@ -144,6 +144,7 @@ NodeInfo::toJson() const
     val["ipv4"] = ipv4.toJson();
     val["ipv6"] = ipv6.toJson();
     val["ops"] = Json::Value::LargestUInt(ongoing_ops);
+    val["metrics"] = metrics.toJson();
     return val;
 }
 
@@ -157,6 +158,21 @@ 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

@@ -487,15 +487,16 @@ void Dht::searchSendAnnounceValue(const Sp<Search>& sr) {
         }
     };
 
-    Sp<Query> probe_query {};
+    static const auto PROBE_QUERY = std::make_shared<Query>(Select {}.field(Value::Field::Id).field(Value::Field::SeqNum));
+
     const auto& now = scheduler.time();
     for (auto& np : sr->nodes) {
         auto& n = *np;
         if (not n.isSynced(now))
             continue;
 
-        const auto& gs = n.probe_query ? n.getStatus.find(n.probe_query) : n.getStatus.cend();
-        if (gs != n.getStatus.cend() and gs->second and gs->second->pending()) {
+        auto gs = n.probe_query ? n.getStatus.find(n.probe_query) : n.getStatus.end();
+        if (gs != n.getStatus.end() and gs->second and gs->second->pending()) {
             continue;
         }
 
@@ -517,18 +518,17 @@ void Dht::searchSendAnnounceValue(const Sp<Search>& sr) {
         }
 
         if (sendQuery) {
-            if (not probe_query)
-                probe_query = std::make_shared<Query>(Select {}.field(Value::Field::Id).field(Value::Field::SeqNum));
+            n.probe_query = PROBE_QUERY;
             if (logger_)
                 logger_->d(sr->id, n.node->id, "[search %s] [node %s] sending %s",
-                    sr->id.toString().c_str(), n.node->toString().c_str(), probe_query->toString().c_str());
-            n.probe_query = probe_query;
-            n.getStatus[probe_query] = network_engine.sendGetValues(n.node,
+                    sr->id.toString().c_str(), n.node->toString().c_str(), n.probe_query->toString().c_str());
+            auto req = network_engine.sendGetValues(n.node,
                     sr->id,
-                    *probe_query,
+                    *PROBE_QUERY,
                     -1,
                     onSelectDone,
-                    std::bind(&Dht::searchNodeGetExpired, this, _1, _2, ws, probe_query));
+                    std::bind(&Dht::searchNodeGetExpired, this, _1, _2, ws, PROBE_QUERY));
+            n.getStatus[PROBE_QUERY] = std::move(req);
         }
         if (not n.candidate and ++i == TARGET_NODES)
             break;
@@ -1409,10 +1409,7 @@ void
 Dht::rotateSecrets()
 {
     oldsecret = secret;
-    {
-        crypto::random_device rdev;
-        secret = std::uniform_int_distribution<uint64_t>{}(rdev);
-    }
+    secret = std::uniform_int_distribution<uint64_t>{}(rd);
     uniform_duration_distribution<> time_dist(std::chrono::minutes(15), std::chrono::minutes(45));
     auto rotate_secrets_time = scheduler.time() + time_dist(rd);
     scheduler.add(rotate_secrets_time, std::bind(&Dht::rotateSecrets, this));
@@ -1480,9 +1477,25 @@ Dht::getNodesStats(sa_family_t af) const
             stats.cached_nodes++;
     }
     stats.table_depth = bcks.depth(bcks.findBucket(myid));
+    stats.searches = searches(af).size();
+    stats.cached_nodes = network_engine.getNodeCacheSize(af);
     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
 {
@@ -1781,10 +1794,7 @@ Dht::Dht(std::unique_ptr<net::DatagramSocket>&& sock, const Config& config, cons
     nextNodesConfirmation = scheduler.add(scheduler.time() + time_dis(rd), std::bind(&Dht::confirmNodes, this));
 
     // Fill old secret
-    {
-        crypto::random_device rdev;
-        secret = std::uniform_int_distribution<uint64_t>{}(rdev);
-    }
+    secret = std::uniform_int_distribution<uint64_t>{}(rd);
     rotateSecrets();
 
     expire();
@@ -1822,7 +1832,7 @@ Dht::neighbourhoodMaintenance(RoutingTable& list)
             q = r;
     }
 
-    auto n = q->randomNode();
+    auto n = q->randomNode(rd);
     if (n) {
         if (logger_)
             logger_->d(id, n->id, "[node %s] sending [find %s] for neighborhood maintenance",
@@ -1847,7 +1857,7 @@ Dht::bucketMaintenance(RoutingTable& list)
             /* This bucket hasn't seen any positive confirmation for a long
                time. Pick a random id in this bucket's range, and send a request
                to a random node. */
-            InfoHash id = list.randomId(b);
+            InfoHash id = list.randomId(b, rd);
             auto q = b;
             /* If the bucket is empty, we try to fill it from a neighbour.
                We also sometimes do it gratuitiously to recover from
@@ -1860,7 +1870,7 @@ Dht::bucketMaintenance(RoutingTable& list)
                     q = r;
             }
 
-            auto n = q->randomNode();
+            auto n = q->randomNode(rd);
             if (n and not n->isPendingMessage()) {
                 want_t want = -1;
 
@@ -1918,37 +1928,25 @@ Dht::maintainStorage(decltype(store)::value_type& storage, bool force, const Don
     const auto& now = scheduler.time();
     size_t announce_per_af = 0;
 
-    bool want4 = true, want6 = true;
-
-    auto nodes = buckets4.findClosestNodes(storage.first, now);
-    if (!nodes.empty()) {
-        if (force || storage.first.xorCmp(nodes.back()->id, myid) < 0) {
-            for (auto &value : storage.second.getValues()) {
-                const auto& vt = getType(value.data->type);
-                if (force || value.created + vt.expiration > now + MAX_STORAGE_MAINTENANCE_EXPIRE_TIME) {
-                    // gotta put that value there
-                    announce(storage.first, AF_INET, value.data, donecb, value.created);
-                    ++announce_per_af;
+    auto maintain = [&](sa_family_t af){
+        bool want = true;
+        auto nodes = buckets(af).findClosestNodes(storage.first, now);
+        if (!nodes.empty()) {
+            if (force || storage.first.xorCmp(nodes.back()->id, myid) < 0) {
+                for (auto &value : storage.second.getValues()) {
+                    const auto& vt = getType(value.data->type);
+                    if (force || value.created + vt.expiration > now + MAX_STORAGE_MAINTENANCE_EXPIRE_TIME) {
+                        // gotta put that value there
+                        announce(storage.first, af, value.data, donecb, value.created);
+                        ++announce_per_af;
+                    }
                 }
+                want = false;
             }
-            want4 = false;
         }
-    }
-
-    auto nodes6 = buckets6.findClosestNodes(storage.first, now);
-    if (!nodes6.empty()) {
-        if (force || storage.first.xorCmp(nodes6.back()->id, myid) < 0) {
-            for (auto &value : storage.second.getValues()) {
-                const auto& vt = getType(value.data->type);
-                if (force || value.created + vt.expiration > now + MAX_STORAGE_MAINTENANCE_EXPIRE_TIME) {
-                    // gotta put that value there
-                    announce(storage.first, AF_INET6, value.data, donecb, value.created);
-                    ++announce_per_af;
-                }
-            }
-            want6 = false;
-        }
-    }
+        return want;
+    };
+    bool want4 = maintain(AF_INET), want6 = maintain(AF_INET6);
 
     if (not want4 and not want6) {
         if (logger_)

src/dht_proxy_server.cpp

@@ -341,6 +341,7 @@ DhtProxyServer::addServerSettings(ServerSettings& settings, const unsigned int m
     // socket options
     settings.socket_options_setter([](auto & options){
         options.set_option(asio::ip::tcp::no_delay{true});
+        options.set_option(asio::socket_base::keep_alive{true});
     });
     settings.connection_state_listener(connListener_);
 }
@@ -360,7 +361,7 @@ DhtProxyServer::updateStats(std::shared_ptr<NodeInfo> info) const
 #endif
     stats.putCount = puts_.size();
     stats.listenCount = listeners_.size();
-    stats.nodeInfo = info;
+    stats.nodeInfo = std::move(info);
     return sstats;
 }
 
@@ -945,7 +946,10 @@ DhtProxyServer::put(restinio::request_handle_t request,
                 auto& sPuts = puts_[infoHash];
                 if (value->id == Value::INVALID_ID) {
                     for (auto& pp : sPuts.puts) {
-                        if (pp.second.pushToken == pushToken and pp.second.value->contentEquals(*value)) {
+                        if (pp.second.pushToken == pushToken 
+                            and pp.second.clientId == clientId 
+                            and pp.second.value->contentEquals(*value))
+                        {
                             pp.second.expireTimer->expires_at(timeout);
                             pp.second.expireTimer->async_wait(std::bind(&DhtProxyServer::handleCancelPermamentPut, this,
                                                         std::placeholders::_1, infoHash, pp.second.value->id));
@@ -972,6 +976,7 @@ DhtProxyServer::put(restinio::request_handle_t request,
 #ifdef OPENDHT_PUSH_NOTIFICATIONS
                     if (not pushToken.empty()){
                         pput.pushToken = pushToken;
+                        pput.clientId = clientId;
                         pput.sessionCtx = std::make_shared<PushSessionContext>();
                         pput.sessionCtx->sessionId = sessionId;
                         // notify push listen expire

src/dhtrunner.cpp

@@ -479,6 +479,8 @@ 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;
@@ -496,8 +498,9 @@ 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(sinfo);
+        cb(std::move(sinfo));
         opEnded();
     });
     cv.notify_all();

src/http.cpp

@@ -31,6 +31,7 @@ namespace http {
 
 constexpr char HTTP_HEADER_CONTENT_TYPE_JSON[] = "application/json";
 constexpr char HTTP_HEADER_DELIM[] = "\r\n\r\n";
+constexpr unsigned MAX_REDIRECTS {5};
 
 Url::Url(const std::string& url): url(url)
 {
@@ -976,18 +977,44 @@ void
 Request::onHeadersComplete() {
     notify_state_change(State::HEADER_RECEIVED);
 
-    auto expect_it = headers_.find(restinio::http_field_t::expect);
-    if (expect_it != headers_.end() and (expect_it->second == "100-continue") and response_.status_code != 200){
-        notify_state_change(State::SENDING);
-        request_.append(body_);
-        std::ostream request_stream(&conn_->input());
-        request_stream << body_ << "\r\n";
-        std::weak_ptr<Request> wthis = shared_from_this();
-        conn_->async_write([wthis](const asio::error_code& ec, size_t) {
-            if (auto sthis = wthis.lock())
-                sthis->handle_request(ec);
-        });
-        return;
+    if (response_.status_code == restinio::status_code::moved_permanently.raw_code() or
+        response_.status_code == restinio::status_code::found.raw_code())
+    {
+        auto location_it = response_.headers.find(restinio::field_to_string(restinio::http_field_t::location));
+        if (location_it == response_.headers.end()){
+            if (logger_)
+                logger_->e("[http:client] [request:%i] got redirect without location", id_);
+            terminate(asio::error::connection_aborted);
+        }
+
+        if (follow_redirect and num_redirect < MAX_REDIRECTS) {
+            auto next = std::make_shared<Request>(ctx_, location_it->second, logger_);
+            next->set_method(header_.method());
+            next->headers_ = std::move(headers_);
+            next->body_ = std::move(body_);
+            next->cbs_ = std::move(cbs_);
+            next->num_redirect = num_redirect + 1;
+            next_ = next;
+            next->prev_ = shared_from_this();
+            next->send();
+        } else {
+            if (logger_)
+                logger_->e("[http:client] [request:%i] got redirect without location", id_);
+            terminate(asio::error::connection_aborted);
+        }
+    } else {
+        auto expect_it = headers_.find(restinio::http_field_t::expect);
+        if (expect_it != headers_.end() and (expect_it->second == "100-continue") and response_.status_code != 200){
+            notify_state_change(State::SENDING);
+            request_.append(body_);
+            std::ostream request_stream(&conn_->input());
+            request_stream << body_ << "\r\n";
+            std::weak_ptr<Request> wthis = shared_from_this();
+            conn_->async_write([wthis](const asio::error_code& ec, size_t) {
+                if (auto sthis = wthis.lock())
+                    sthis->handle_request(ec);
+            });
+        }
     }
 }
 

src/network_engine.cpp

@@ -100,7 +100,7 @@ RequestAnswer::RequestAnswer(ParsedMessage&& msg)
 {}
 
 NetworkEngine::NetworkEngine(const Sp<Logger>& log, std::mt19937_64& rand, Scheduler& scheduler, std::unique_ptr<DatagramSocket>&& sock)
-    : myid(zeroes), dht_socket(std::move(sock)), logger_(log), rd(rand), rate_limiter((size_t)-1), scheduler(scheduler)
+    : myid(zeroes), dht_socket(std::move(sock)), logger_(log), rd(rand), cache(rd), rate_limiter((size_t)-1), scheduler(scheduler)
 {}
 
 NetworkEngine::NetworkEngine(InfoHash& myid, NetworkConfig c,
@@ -127,6 +127,7 @@ NetworkEngine::NetworkEngine(InfoHash& myid, NetworkConfig c,
     onAnnounce(std::move(onAnnounce)),
     onRefresh(std::move(onRefresh)),
     myid(myid), config(c), dht_socket(std::move(sock)), logger_(log), rd(rand),
+    cache(rd),
     rate_limiter(config.max_req_per_sec),
     scheduler(scheduler)
 {}

src/network_utils.cpp

@@ -288,30 +288,36 @@ UdpSocket::openSockets(const SockAddr& bind4, const SockAddr& bind6)
                             logger->e("Error receiving packet: %s", strerror(errno));
                         int err = errno;
                         if (err == EPIPE || err == ENOTCONN || err == ECONNRESET) {
-                            std::lock_guard<std::mutex> lk(lock);
-                            if (ls4 >= 0) {
-                                close(ls4);
-                                try {
-                                    ls4 = bindSocket(bound4, bound4);
-                                } catch (const DhtException& e) {
-                                    if (logger)
-                                        logger->e("Can't bind inet socket: %s", e.what());
+                            if (not running) break;
+                            std::unique_lock<std::mutex> lk(lock, std::try_to_lock);
+                            if (lk.owns_lock()) {
+                                if (not running) break;
+                                if (ls4 >= 0) {
+                                    close(ls4);
+                                    try {
+                                        ls4 = bindSocket(bound4, bound4);
+                                    } catch (const DhtException& e) {
+                                        if (logger)
+                                            logger->e("Can't bind inet socket: %s", e.what());
+                                    }
                                 }
-                            }
-                            if (ls6 >= 0) {
-                                close(ls6);
-                                try {
-                                    ls6 = bindSocket(bound6, bound6);
-                                } catch (const DhtException& e) {
-                                    if (logger)
-                                        logger->e("Can't bind inet6 socket: %s", e.what());
+                                if (ls6 >= 0) {
+                                    close(ls6);
+                                    try {
+                                        ls6 = bindSocket(bound6, bound6);
+                                    } catch (const DhtException& e) {
+                                        if (logger)
+                                            logger->e("Can't bind inet6 socket: %s", e.what());
+                                    }
                                 }
-                            }
-                            if (ls4 < 0 && ls6 < 0)
+                                if (ls4 < 0 && ls6 < 0)
+                                    break;
+                                s4 = ls4;
+                                s6 = ls6;
+                                selectFd = std::max({ls4, ls6, stop_readfd}) + 1;
+                            } else {
                                 break;
-                            s4 = ls4;
-                            s6 = ls6;
-                            selectFd = std::max({ls4, ls6, stop_readfd}) + 1;
+                            }
                         }
                     }
                 }
@@ -326,14 +332,16 @@ UdpSocket::openSockets(const SockAddr& bind4, const SockAddr& bind6)
             close(ls6);
         if (stop_readfd != -1)
             close(stop_readfd);
-        std::lock_guard<std::mutex> lk(lock);
-        s4 = -1;
-        s6 = -1;
-        bound4 = {};
-        bound6 = {};
         if (stopfd != -1)
             close(stopfd);
-        stopfd = -1;
+        std::unique_lock<std::mutex> lk(lock, std::try_to_lock);
+        if (lk.owns_lock()) {
+            s4 = -1;
+            s6 = -1;
+            bound4 = {};
+            bound6 = {};
+            stopfd = -1;
+        }
     });
 }
 

src/node.cpp

@@ -30,17 +30,15 @@ constexpr std::chrono::minutes Node::NODE_EXPIRE_TIME;
 constexpr std::chrono::minutes Node::NODE_GOOD_TIME;
 constexpr std::chrono::seconds Node::MAX_RESPONSE_TIME;
 
-Node::Node(const InfoHash& id, const SockAddr& addr, bool client)
+Node::Node(const InfoHash& id, const SockAddr& addr, std::mt19937_64& rd, bool client)
 : id(id), addr(addr), is_client(client), sockets_()
 {
-    thread_local crypto::random_device rd {};
     transaction_id = std::uniform_int_distribution<Tid>{1}(rd);
 }
 
-Node::Node(const InfoHash& id, SockAddr&& addr, bool client)
+Node::Node(const InfoHash& id, SockAddr&& addr, std::mt19937_64& rd, bool client)
 : id(id), addr(std::move(addr)), is_client(client), sockets_()
 {
-    thread_local crypto::random_device rd {};
     transaction_id = std::uniform_int_distribution<Tid>{1}(rd);
 }
 

src/node_cache.cpp

@@ -37,8 +37,8 @@ NodeCache::getNode(const InfoHash& id, sa_family_t family) {
 Sp<Node>
 NodeCache::getNode(const InfoHash& id, const SockAddr& addr, time_point now, bool confirm, bool client) {
     if (not id)
-        return std::make_shared<Node>(id, addr);
-    return cache(addr.getFamily()).getNode(id, addr, now, confirm, client);
+        return std::make_shared<Node>(id, addr, rd, client);
+    return cache(addr.getFamily()).getNode(id, addr, now, confirm, client, rd);
 }
 
 std::vector<Sp<Node>>
@@ -104,13 +104,13 @@ NodeCache::NodeMap::getNode(const InfoHash& id)
 }
 
 Sp<Node>
-NodeCache::NodeMap::getNode(const InfoHash& id, const SockAddr& addr, time_point now, bool confirm, bool client)
+NodeCache::NodeMap::getNode(const InfoHash& id, const SockAddr& addr, time_point now, bool confirm, bool client, std::mt19937_64& rd)
 {
-    auto it = emplace(id, std::weak_ptr<Node>{});
-    auto node = it.first->second.lock();
+    auto& nref = (*this)[id];
+    auto node = nref.lock();
     if (not node) {
-        node = std::make_shared<Node>(id, addr, client);
-        it.first->second = node;
+        node = std::make_shared<Node>(id, addr, rd, client);
+        nref = node;
         if (cleanup_counter++ == CLEANUP_FREQ) {
             cleanup();
             cleanup_counter = 0;

src/request.h

@@ -29,6 +29,8 @@ namespace net {
 class NetworkEngine;
 struct ParsedMessage;
 
+extern std::atomic<size_t> REQ_COUNT;
+
 /*!
  * @class   Request
  * @brief   An atomic request destined to a node.
@@ -67,14 +69,22 @@ struct Request {
         }
     }
 
-    Request(State state = State::PENDING) : state_(state) {}
+    Request(State state = State::PENDING) : state_(state) {
+        REQ_COUNT++;
+    }
     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) { }
+        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--;
+    }
 
     Tid getTid() const { return tid; }
     MessageType getType() const { return type; }

src/rng.cpp

@@ -25,7 +25,7 @@ namespace dht {
 namespace crypto {
 
 random_device::random_device() :
-   gen(std::chrono::system_clock::now().time_since_epoch().count())
+   gen(std::chrono::system_clock::now().time_since_epoch().count() ^ std::chrono::high_resolution_clock::now().count())
 {}
 
 random_device::result_type

src/routing_table.cpp

@@ -25,15 +25,8 @@
 
 namespace dht {
 
-static std::mt19937 rd{ dht::crypto::random_device{}() };
-#ifdef _WIN32
-static std::uniform_int_distribution<int> rand_byte{ 0, std::numeric_limits<uint8_t>::max() };
-#else
-static std::uniform_int_distribution<uint8_t> rand_byte;
-#endif
-
 Sp<Node>
-Bucket::randomNode()
+Bucket::randomNode(std::mt19937_64& rd)
 {
     if (nodes.empty())
         return nullptr;
@@ -65,7 +58,7 @@ void Bucket::sendCachedPing(net::NetworkEngine& ne)
 }
 
 InfoHash
-RoutingTable::randomId(const RoutingTable::const_iterator& it) const
+RoutingTable::randomId(const RoutingTable::const_iterator& it, std::mt19937_64& rd) const
 {
     int bit1 = it->first.lowbit();
     int bit2 = std::next(it) != end() ? std::next(it)->first.lowbit() : -1;
@@ -74,6 +67,12 @@ RoutingTable::randomId(const RoutingTable::const_iterator& it) const
     if (bit >= 8*(int)HASH_LEN)
         return it->first;
 
+#ifdef _WIN32
+    std::uniform_int_distribution<int> rand_byte{ 0, std::numeric_limits<uint8_t>::max() };
+#else
+    std::uniform_int_distribution<uint8_t> rand_byte;
+#endif
+
     int b = bit/8;
     InfoHash id_return;
     std::copy_n(it->first.cbegin(), b, id_return.begin());

src/search.h

@@ -557,6 +557,14 @@ struct Dht::Search {
             else
                 ++it;
         }
+        for (auto& n : nodes) {
+            auto ackIt = n->acked.find(vid);
+            if (ackIt != n->acked.end()) {
+                if (ackIt->second.first)
+                    ackIt->second.first->cancel();
+                n->acked.erase(ackIt);
+            }
+        }
         return canceled;
     }
 
@@ -662,8 +670,14 @@ struct Dht::Search {
         });
         // remove acked for cleared annouces
         for (auto it = announced; it != announce.end(); ++it) {
-            for (auto& n : nodes)
-                n->acked.erase(it->value->id);
+            for (auto& n : nodes) {
+                auto ackIt = n->acked.find(it->value->id);
+                if (ackIt != n->acked.end()) {
+                    if (ackIt->second.first)
+                        ackIt->second.first->cancel();
+                    n->acked.erase(ackIt);
+                }
+            }
         }
         announce.erase(announced, announce.end());
     }

src/securedht.cpp

@@ -37,7 +37,7 @@ extern "C" {
 namespace dht {
 
 SecureDht::SecureDht(std::unique_ptr<DhtInterface> dht, SecureDht::Config conf)
-: dht_(std::move(dht)), key_(conf.id.first), certificate_(conf.id.second)
+: dht_(std::move(dht)), key_(conf.id.first), certificate_(conf.id.second), enableCache_(conf.cert_cache_all)
 {
     if (!dht_) return;
     for (const auto& type : DEFAULT_TYPES)
@@ -280,11 +280,12 @@ SecureDht::checkValue(const Sp<Value>& v)
         v->signatureChecked = true;
         if (v->owner and v->owner->checkSignature(v->getToSign(), v->signature)) {
             v->signatureValid = true;
-            nodesPubKeys_[v->owner->getId()] = v->owner;
+            if (enableCache_)
+                nodesPubKeys_[v->owner->getId()] = v->owner;
             return v;
         }
         else if (logger_)
-                logger_->w("Signature verification failed for %s", v->toString().c_str());
+            logger_->w("Signature verification failed for %s", v->toString().c_str());
     }
     // Forward normal values
     else {
@@ -352,6 +353,11 @@ SecureDht::listen(const InfoHash& id, GetCallback cb, Value::Filter f, Where w)
 void
 SecureDht::putSigned(const InfoHash& hash, Sp<Value> val, DoneCallback callback, bool permanent)
 {
+    if (not key_)  {
+        if (callback)
+            callback(false, {});
+        return;
+    }
     if (val->id == Value::INVALID_ID) {
         crypto::random_device rdev;
         std::uniform_int_distribution<Value::Id> rand_id;
@@ -393,6 +399,11 @@ SecureDht::putSigned(const InfoHash& hash, Sp<Value> val, DoneCallback callback,
 void
 SecureDht::putEncrypted(const InfoHash& hash, const InfoHash& to, Sp<Value> val, DoneCallback callback, bool permanent)
 {
+    if (not key_)  {
+        if (callback)
+            callback(false, {});
+        return;
+    }
     findPublicKey(to, [=](const Sp<const crypto::PublicKey>& pk) {
         if(!pk || !*pk) {
             if (callback)

src/utils.cpp

@@ -34,6 +34,9 @@
 #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}};
 
@@ -236,13 +239,23 @@ bool operator==(const SockAddr& a, const SockAddr& b) {
 }
 
 time_point from_time_t(std::time_t t) {
-    return clock::now() + (std::chrono::system_clock::from_time_t(t) - std::chrono::system_clock::now());
+    auto dt = system_clock::from_time_t(t) - system_clock::now();
+    auto now = clock::now();
+    if (dt > system_clock::duration(0) and now > time_point::max() - dt)
+        return time_point::max();
+    else if (dt < system_clock::duration(0) and now < time_point::min() - dt)
+        return time_point::min();
+    return now + dt;
 }
 
 std::time_t to_time_t(time_point t) {
-    return std::chrono::system_clock::to_time_t(
-            std::chrono::system_clock::now() +
-            std::chrono::duration_cast<std::chrono::system_clock::duration>(t - clock::now()));
+    auto dt = t - clock::now();
+    auto now = system_clock::now();
+    if (dt > duration(0) and now >= system_clock::time_point::max() - dt)
+        return system_clock::to_time_t(system_clock::time_point::max());
+    else if (dt < duration(0) and now <= system_clock::time_point::min() - dt)
+        return system_clock::to_time_t(system_clock::time_point::min());
+    return system_clock::to_time_t(now + std::chrono::duration_cast<system_clock::duration>(dt));
 }
 
 Blob
@@ -274,7 +287,7 @@ findMapValue(const msgpack::object& map, const char* key) {
     if (map.type != msgpack::type::MAP) throw msgpack::type_error();
     for (unsigned i = 0; i < map.via.map.size; i++) {
         auto& o = map.via.map.ptr[i];
-        if (o.key.type == msgpack::type::STR 
+        if (o.key.type == msgpack::type::STR
             && std::strncmp(o.key.via.str.ptr, key, o.key.via.str.size) == 0)
             return &o.val;
     }
@@ -286,7 +299,7 @@ findMapValue(const msgpack::object& map, const std::string& key) {
     if (map.type != msgpack::type::MAP) throw msgpack::type_error();
     for (unsigned i = 0; i < map.via.map.size; i++) {
         auto& o = map.via.map.ptr[i];
-        if (o.key.type == msgpack::type::STR 
+        if (o.key.type == msgpack::type::STR
             && key.size() == o.key.via.str.size
             && std::strncmp(o.key.via.str.ptr, key.data(), o.key.via.str.size) == 0)
             return &o.val;

src/value.cpp

@@ -261,18 +261,18 @@ FieldValue::operator==(const FieldValue& vfd) const
     if (field != vfd.field)
         return false;
     switch (field) {
-        case Value::Field::Id:
-        case Value::Field::ValueType:
-        case Value::Field::SeqNum:
-            return intValue == vfd.intValue;
-        case Value::Field::OwnerPk:
-            return hashValue == vfd.hashValue;
-        case Value::Field::UserType:
-            return blobValue == vfd.blobValue;
-        case Value::Field::None:
-            return true;
-        default:
-            return false;
+    case Value::Field::Id:
+    case Value::Field::ValueType:
+    case Value::Field::SeqNum:
+        return intValue == vfd.intValue;
+    case Value::Field::OwnerPk:
+        return hashValue == vfd.hashValue;
+    case Value::Field::UserType:
+        return blobValue == vfd.blobValue;
+    case Value::Field::None:
+        return true;
+    default:
+        return false;
     }
 }
 
@@ -280,18 +280,18 @@ Value::Filter
 FieldValue::getLocalFilter() const
 {
     switch (field) {
-        case Value::Field::Id:
-            return Value::IdFilter(intValue);
-        case Value::Field::ValueType:
-            return Value::TypeFilter(intValue);
-        case Value::Field::OwnerPk:
-            return Value::OwnerFilter(hashValue);
-        case Value::Field::SeqNum:
-            return Value::SeqNumFilter(intValue);
-        case Value::Field::UserType:
-            return Value::UserTypeFilter(std::string {blobValue.begin(), blobValue.end()});
-        default:
-            return {};
+    case Value::Field::Id:
+        return Value::IdFilter(intValue);
+    case Value::Field::ValueType:
+        return Value::TypeFilter(intValue);
+    case Value::Field::OwnerPk:
+        return Value::OwnerFilter(hashValue);
+    case Value::Field::SeqNum:
+        return Value::SeqNumFilter(intValue);
+    case Value::Field::UserType:
+        return Value::UserTypeFilter(std::string {blobValue.begin(), blobValue.end()});
+    default:
+        return {};
     }
 }
 
@@ -311,23 +311,23 @@ FieldValueIndex::FieldValueIndex(const Value& v, const Select& s)
     for (const auto& fvp : index) {
         const auto& f = fvp.first;
         switch (f) {
-            case Value::Field::Id:
-                index[f] = {f, v.id};
-                break;
-            case Value::Field::ValueType:
-                index[f] = {f, v.type};
-                break;
-            case Value::Field::OwnerPk:
-                index[f] = {f, v.owner ? v.owner->getId() : InfoHash() };
-                break;
-            case Value::Field::SeqNum:
-                index[f] = {f, v.seq};
-                break;
-            case Value::Field::UserType:
-                index[f] = {f, Blob {v.user_type.begin(), v.user_type.end()}};
-                break;
-            default:
-                break;
+        case Value::Field::Id:
+            index[f] = {f, v.id};
+            break;
+        case Value::Field::ValueType:
+            index[f] = {f, v.type};
+            break;
+        case Value::Field::OwnerPk:
+            index[f] = {f, v.owner ? v.owner->getId() : InfoHash() };
+            break;
+        case Value::Field::SeqNum:
+            index[f] = {f, v.seq};
+            break;
+        case Value::Field::UserType:
+            index[f] = {f, Blob {v.user_type.begin(), v.user_type.end()}};
+            break;
+        default:
+            break;
         }
     }
 }
@@ -347,28 +347,28 @@ std::ostream& operator<<(std::ostream& os, const FieldValueIndex& fvi) {
     os << "Index[";
     for (auto v = fvi.index.begin(); v != fvi.index.end(); ++v) {
         switch (v->first) {
-            case Value::Field::Id: {
-                auto flags(os.flags());
-                os << "Id:" << std::hex << v->second.getInt();
-                os.flags(flags);
-                break;
-            }
-            case Value::Field::ValueType:
-                os << "ValueType:" << v->second.getInt();
-                break;
-            case Value::Field::OwnerPk:
-                os << "Owner:" << v->second.getHash().toString();
-                break;
-            case Value::Field::SeqNum:
-                os << "Seq:" << v->second.getInt();
-                break;
-            case Value::Field::UserType: {
-                auto ut = v->second.getBlob();
-                os << "UserType:" << std::string(ut.begin(), ut.end());
-                break;
-            }
-            default:
-                break;
+        case Value::Field::Id: {
+            auto flags(os.flags());
+            os << "Id:" << std::hex << v->second.getInt();
+            os.flags(flags);
+            break;
+        }
+        case Value::Field::ValueType:
+            os << "ValueType:" << v->second.getInt();
+            break;
+        case Value::Field::OwnerPk:
+            os << "Owner:" << v->second.getHash();
+            break;
+        case Value::Field::SeqNum:
+            os << "Seq:" << v->second.getInt();
+            break;
+        case Value::Field::UserType: {
+            auto ut = v->second.getBlob();
+            os << "UserType:" << std::string(ut.begin(), ut.end());
+            break;
+        }
+        default:
+            break;
         }
         os << (std::next(v) != fvi.index.end() ? "," : "");
     }
@@ -384,19 +384,19 @@ FieldValueIndex::msgpack_unpack_fields(const std::set<Value::Field>& fields, con
     for (const auto& field : fields) {
         auto& field_value = o.via.array.ptr[offset+(j++)];
         switch (field) {
-            case Value::Field::Id:
-            case Value::Field::ValueType:
-            case Value::Field::SeqNum:
-                index[field] = FieldValue(field, field_value.as<uint64_t>());
-                break;
-            case Value::Field::OwnerPk:
-                index[field] = FieldValue(field, field_value.as<InfoHash>());
-                break;
-            case Value::Field::UserType:
-                index[field] = FieldValue(field, field_value.as<Blob>());
-                break;
-            default:
-                throw msgpack::type_error();
+        case Value::Field::Id:
+        case Value::Field::ValueType:
+        case Value::Field::SeqNum:
+            index[field] = FieldValue(field, field_value.as<uint64_t>());
+            break;
+        case Value::Field::OwnerPk:
+            index[field] = FieldValue(field, field_value.as<InfoHash>());
+            break;
+        case Value::Field::UserType:
+            index[field] = FieldValue(field, field_value.as<Blob>());
+            break;
+        default:
+            throw msgpack::type_error();
         }
     }
 }
@@ -522,9 +522,12 @@ bool Query::isSatisfiedBy(const Query& q) const {
 }
 
 std::ostream& operator<<(std::ostream& s, const dht::Select& select) {
-    s << "SELECT " << (select.fieldSelection_.empty() ? "*" : "");
-    for (auto fs = select.fieldSelection_.begin() ; fs != select.fieldSelection_.end() ; ++fs) {
-        switch (*fs) {
+    s << "SELECT ";
+    if (select.fieldSelection_.empty())
+        s << '*';
+    else
+        for (auto fs = select.fieldSelection_.begin(); fs != select.fieldSelection_.end();) {
+            switch (*fs) {
             case Value::Field::Id:
                 s << VALUE_KEY_ID;
                 break;
@@ -542,9 +545,10 @@ std::ostream& operator<<(std::ostream& s, const dht::Select& select) {
                 break;
             default:
                 break;
+            }
+            if (++fs != select.fieldSelection_.end())
+                s << ',';
         }
-        s << (std::next(fs) != select.fieldSelection_.end() ? "," : "");
-    }
     return s;
 }
 
@@ -553,25 +557,25 @@ std::ostream& operator<<(std::ostream& s, const dht::Where& where) {
         s << "WHERE ";
         for (auto f = where.filters_.begin() ; f != where.filters_.end() ; ++f) {
             switch (f->getField()) {
-                case Value::Field::Id:
-                    s << "id=" << f->getInt();
-                    break;
-                case Value::Field::ValueType:
-                    s << "value_type=" << f->getInt();
-                    break;
-                case Value::Field::OwnerPk:
-                    s << "owner_pk_hash=" << f->getHash().toString();
-                    break;
-                case Value::Field::SeqNum:
-                    s << "seq=" << f->getInt();
-                    break;
-                case Value::Field::UserType: {
-                    auto b = f->getBlob();
-                    s << "user_type=" << std::string {b.begin(), b.end()};
-                    break;
-                }
-                default:
-                    break;
+            case Value::Field::Id:
+                s << VALUE_KEY_ID << '=' << f->getInt();
+                break;
+            case Value::Field::ValueType:
+                s << "value_type=" << f->getInt();
+                break;
+            case Value::Field::OwnerPk:
+                s << "owner_pk_hash=" << f->getHash();
+                break;
+            case Value::Field::SeqNum:
+                s << VALUE_KEY_SEQ << '=' << f->getInt();
+                break;
+            case Value::Field::UserType: {
+                auto b = f->getBlob();
+                s << "user_type=" << std::string {b.begin(), b.end()};
+                break;
+            }
+            default:
+                break;
             }
             s << (std::next(f) != where.filters_.end() ? "," : "");
         }

tools/CMakeLists.txt

@@ -2,21 +2,34 @@
 add_executable (dhtnode dhtnode.cpp tools_common.h)
 add_executable (dhtscanner dhtscanner.cpp tools_common.h)
 add_executable (dhtchat dhtchat.cpp tools_common.h)
+add_executable (perftest perftest.cpp tools_common.h)
 
 target_link_libraries (dhtnode LINK_PUBLIC ${READLINE_LIBRARIES})
 target_link_libraries (dhtscanner LINK_PUBLIC ${READLINE_LIBRARIES})
 target_link_libraries (dhtchat LINK_PUBLIC ${READLINE_LIBRARIES})
+target_link_libraries (perftest LINK_PUBLIC ${READLINE_LIBRARIES})
 
 if (OPENDHT_SHARED)
 	target_link_libraries (dhtnode LINK_PUBLIC opendht)
 	target_link_libraries (dhtscanner LINK_PUBLIC opendht)
 	target_link_libraries (dhtchat LINK_PUBLIC opendht)
+	target_link_libraries (perftest LINK_PUBLIC opendht)
 else ()
 	target_link_libraries (dhtnode LINK_PUBLIC opendht-static)
 	target_link_libraries (dhtscanner LINK_PUBLIC opendht-static)
 	target_link_libraries (dhtchat LINK_PUBLIC opendht-static)
+	target_link_libraries (perftest LINK_PUBLIC opendht-static)
 endif ()
 
+if (OPENDHT_HTTP)
+	add_executable (durl durl.cpp)
+	if (OPENDHT_SHARED)
+		target_link_libraries (durl LINK_PUBLIC opendht)
+	else ()
+		target_link_libraries (durl LINK_PUBLIC opendht-static)
+	endif ()
+endif()
+
 if (OPENDHT_C)
 	add_executable (dhtcnode dhtcnode.c)
 	target_link_libraries (dhtcnode LINK_PUBLIC opendht-c ${READLINE_LIBRARIES})

tools/dhtnode.cpp

@@ -145,6 +145,14 @@ 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/durl.cpp

@@ -0,0 +1,120 @@
+/*
+ *  Copyright (C) 2014-2020 Savoir-faire Linux Inc.
+ *
+ *  Authors: Adrien Béraud <adrien.beraud@savoirfairelinux.com>
+ *
+ *  This program is free software; you can redistribute it and/or modify
+ *  it under the terms of the GNU General Public License as published by
+ *  the Free Software Foundation; either version 3 of the License, or
+ *  (at your option) any later version.
+ *
+ *  This program is distributed in the hope that it will be useful,
+ *  but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *  GNU General Public License for more details.
+ *
+ *  You should have received a copy of the GNU General Public License
+ *  along with this program. If not, see <https://www.gnu.org/licenses/>.
+ */
+#include <opendht/http.h>
+#include <opendht/log.h>
+
+#include <asio/io_context.hpp>
+#include <getopt.h>
+
+using namespace dht;
+
+void print_info() {
+    std::cout << "durl, a simple http(s) client." << std::endl;
+    std::cout << "Report bugs to: https://opendht.net" << std::endl;
+}
+
+void print_version() {
+    std::cout << "OpenDHT version " << dht::version() << std::endl;
+    print_info();
+}
+
+void print_usage() {
+    std::cout << "Usage: durl url" << std::endl << std::endl;
+    print_info();
+}
+
+static const constexpr struct option long_options[] = {
+    {"help",                no_argument      , nullptr, 'h'},
+    {"headers",             no_argument      , nullptr, 'H'},
+    {"verbose",             no_argument      , nullptr, 'v'},
+    {"version",             no_argument      , nullptr, 'V'},
+    {nullptr,               0                , nullptr,  0}
+};
+
+struct durl_params {
+    std::string url;
+    bool help {false};
+    bool log {false};
+    bool headers {false};
+    bool version {false};
+};
+
+durl_params
+parseArgs(int argc, char **argv) {
+    durl_params params;
+    int opt;
+    while ((opt = getopt_long(argc, argv, "hvVH", long_options, nullptr)) != -1) {
+        switch (opt) {
+        case 'V':
+            params.version = true;
+            break;
+        case 'h':
+            params.help = true;
+            break;
+        case 'H':
+            params.headers = true;
+            break;
+        case 'v':
+            params.log = true;
+            break;
+        default:
+            break;
+        }
+    }
+    
+    if (optind < argc) {
+        params.url = argv[optind++];
+    }
+
+    return params;
+}
+
+int
+main(int argc, char **argv)
+{
+    auto params = parseArgs(argc, argv);
+    if (params.help) {
+        print_usage();
+        return 0;
+    }
+    if (params.version) {
+        print_version();
+        return 0;
+    }
+
+    std::shared_ptr<dht::Logger> logger;
+    if (params.log) {
+        logger = dht::log::getStdLogger();
+    }
+
+    asio::io_context ctx;
+    auto request = std::make_shared<dht::http::Request>(ctx, params.url, [&](const dht::http::Response& response){
+        if (params.headers) {
+            for (const auto& header : response.headers)
+                std::cout << header.first << ": " << header.second << std::endl;
+            std::cout << std::endl;
+        }
+        std::cout << response.body << std::endl;
+        ctx.stop();
+    }, logger);
+    request->send();
+    auto work = asio::make_work_guard(ctx);
+    ctx.run();
+    return 0;
+}

tools/perftest.cpp

@@ -0,0 +1,180 @@
+/*
+ *  Copyright (C) 2014-2020 Savoir-faire Linux Inc.
+ *
+ *  Author: Adrien Béraud <adrien.beraud@savoirfairelinux.com>
+ *
+ *  This program is free software; you can redistribute it and/or modify
+ *  it under the terms of the GNU General Public License as published by
+ *  the Free Software Foundation; either version 3 of the License, or
+ *  (at your option) any later version.
+ *
+ *  This program is distributed in the hope that it will be useful,
+ *  but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *  GNU General Public License for more details.
+ *
+ *  You should have received a copy of the GNU General Public License
+ *  along with this program. If not, see <https://www.gnu.org/licenses/>.
+ */
+#ifdef HAVE_CONFIG_H
+#include "config.h"
+#endif
+
+#include "tools_common.h"
+#include <opendht/node.h>
+
+extern "C" {
+#include <gnutls/gnutls.h>
+}
+#include <condition_variable>
+#include <mutex>
+#include <atomic>
+
+void print_usage() {
+    std::cout << "Usage: perftest" << std::endl << std::endl;
+    std::cout << "perftest, a simple OpenDHT basic performance tester." << std::endl;
+    std::cout << "Report bugs to: https://opendht.net" << std::endl;
+}
+constexpr unsigned PINGPONG_MAX = 2048;
+using namespace dht;
+
+namespace tests {
+
+using clock = std::chrono::high_resolution_clock;
+using duration = clock::duration;
+
+duration
+benchPingPong(unsigned netSize, unsigned n_parallel) {
+    DhtRunner::Config config {};
+    config.dht_config.node_config.max_peer_req_per_sec = -1;
+    config.dht_config.node_config.max_req_per_sec = -1;
+
+    DhtRunner ping_node, pong_node;
+
+    std::vector<std::pair<InfoHash, InfoHash>> locs;
+    locs.reserve(n_parallel);
+    for (unsigned i=0; i<n_parallel; i++) {
+        auto loc_ping = InfoHash::get("toto" + std::to_string(i));
+        locs.emplace_back(loc_ping, InfoHash::get(loc_ping.toString()));
+    }
+
+    ping_node.run(0, config);
+    pong_node.run(0, config);
+    auto bindAddr = ping_node.getBound();
+    pong_node.bootstrap(bindAddr);
+
+    std::vector<std::unique_ptr<DhtRunner>> nodes;
+    nodes.reserve(netSize);
+    for (unsigned i=0; i<netSize; i++) {
+        auto node = std::make_unique<DhtRunner>();
+        node->run(0, config);
+        node->bootstrap(bindAddr);
+        nodes.emplace_back(std::move(node));
+    }
+
+    // std::this_thread::sleep_for(std::chrono::seconds(1));
+
+    std::condition_variable cv;
+    unsigned i {0};
+    std::mutex m;
+
+    unsigned max = PINGPONG_MAX * n_parallel;
+
+    auto ping = [&](DhtRunner& node, InfoHash h) {
+        std::lock_guard<std::mutex> lk(m);
+        if (i < max)  {
+            i++;
+            node.put(h, Value("hey"));
+        }
+        cv.notify_one();
+    };
+
+    for (unsigned i=0; i<n_parallel; i++)  {
+        ping_node.listen(locs[i].first, [&,i](const std::shared_ptr<Value>&){
+            ping(pong_node, locs[i].second);
+            return true;
+        });
+        pong_node.listen(locs[i].second, [&,i](const std::shared_ptr<Value>&){
+            ping(ping_node, locs[i].first);
+            return true;
+        });
+    }
+
+    auto start = clock::now();
+
+    for (unsigned i=0; i<n_parallel; i++) 
+        ping(pong_node, locs[i].first);
+
+    {
+        std::unique_lock<std::mutex> lk(m);
+        if (not cv.wait_for(lk, std::chrono::minutes(1), [&](){ return i == max; })) {
+            throw std::runtime_error(std::string("Timeout: ") + std::to_string(i));
+        }
+    }
+
+    auto end = clock::now();
+
+    for (auto& node : nodes)
+        node->shutdown();
+    ping_node.shutdown();
+    pong_node.shutdown();
+
+    for (auto& node : nodes)
+        node->join();
+    ping_node.join();
+    pong_node.join();
+
+    return end-start;
+}
+
+}
+
+int
+main(int argc, char **argv)
+{
+#ifdef WIN32_NATIVE
+    gnutls_global_init();
+#endif
+    auto params = parseArgs(argc, argv);
+    if (params.help) {
+        print_usage();
+        return 0;
+    }
+
+    duration totalTime {0};
+    unsigned totalOps {0};
+
+    for (unsigned nparallel = 1; nparallel <= 32; nparallel *= 2)  {
+        unsigned max = PINGPONG_MAX * nparallel;
+        std::vector<duration> results {};
+        results.reserve(8);
+        duration total {0};
+        for (unsigned i=2; i<32; i *= 2) {
+            auto dt = tests::benchPingPong(i - 2, nparallel);
+            std::cout << "Network size: " << i << std::endl;
+            std::cout << max << " ping-pong done, took " << print_duration(dt) << std::endl;
+            std::cout << print_duration(dt/max) << " per rt, "
+                    << max/std::chrono::duration<double>(dt).count() << " ping per s" << std::endl << std::endl;
+            total += dt;
+            totalOps += max;
+            results.emplace_back(dt);
+        }
+
+        totalTime += total;
+
+        std::cout << "Total for " << nparallel << std::endl;
+        auto totNum = max*results.size();
+        std::cout << totNum << " ping-pong done, took " << print_duration(total) << std::endl;
+        std::cout << print_duration(total/totNum) << " per rt, "
+                << totNum/std::chrono::duration<double>(total).count() << " ping per s" << std::endl << std::endl;
+    }
+
+    std::cout << std::endl << "Grand total: " << print_duration(totalTime) << " for " << totalOps << std::endl;
+    std::cout << print_duration(totalTime/totalOps) << " per rt, "
+            << totalOps/std::chrono::duration<double>(totalTime).count() << " ping per s" << std::endl << std::endl;
+
+#ifdef WIN32_NATIVE
+    gnutls_global_deinit();
+#endif
+    return 0;
+}

tools/tools_common.h

@@ -160,6 +160,7 @@ getDhtConfig(dht_params& params)
     config.dht_config.node_config.persist_path = params.persist_path;
     config.dht_config.node_config.public_stable = params.public_stable;
     config.dht_config.id = params.id;
+    config.dht_config.cert_cache_all = static_cast<bool>(params.id.first);
     config.threaded = true;
     config.proxy_server = params.proxyclient;
     config.push_node_id = "dhtnode";