dhtnode: print metrics

Provides info about internal dht structures

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
     /**

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/network_engine.h

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

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;
@@ -49,6 +56,7 @@ private:
         void clearBadNodes();
         void setExpired();
         void cleanup();
+        size_t count() const { return size(); }
     private:
         size_t cleanup_counter {0};
     };

include/opendht/scheduler.h

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

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

@@ -1477,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
 {

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,6 +498,7 @@ DhtRunner::getNodeInfo(std::function<void(std::shared_ptr<NodeInfo>)> cb)
         info.node_id = dht.getNodeId();
         info.ipv4 = dht.getNodesStats(AF_INET);
         info.ipv6 = dht.getNodesStats(AF_INET6);
+        info.metrics = dht.getNodeMetrics();
         info.ongoing_ops = ongoing_ops;
         cb(std::move(sinfo));
         opEnded();

src/network_engine.cpp

@@ -281,9 +281,6 @@ 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] {
@@ -1178,8 +1175,9 @@ NetworkEngine::sendAnnounceValue(Sp<Node> n,
             }
         }
     );
-    req->parts = std::move(v);
     sendRequest(req);
+    if (not v.empty())
+        sendValueParts(tid, v, n->getAddr());
     ++out_stats.put;
     return req;
 }

src/request.h

@@ -29,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; }
@@ -115,7 +125,6 @@ private:
         on_done = {};
         on_expired = {};
         msg = {};
-        parts = {};
     }
 
     const Tid tid {0}; /* the request id. */
@@ -131,7 +140,6 @@ private:
     std::function<void(const Request&, bool)> on_expired {};
 
     Blob msg {};                      /* the serialized message. */
-    std::vector<Blob> parts;
     Tid socket;   /* the socket used for further reponses. */
 };
 

src/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}};
 

src/value.cpp

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

tests/dhtproxytester.cpp

@@ -34,13 +34,10 @@ CPPUNIT_TEST_SUITE_REGISTRATION(DhtProxyTester);
 
 void
 DhtProxyTester::setUp() {
-    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);
+    nodePeer.run(0);
 
     nodeProxy = std::make_shared<dht::DhtRunner>();
-    nodeProxy->run(0, clientConfig);
+    nodeProxy->run(0);
     nodeProxy->bootstrap(nodePeer.getBound());
 
     auto serverCAIdentity = dht::crypto::generateEcIdentity("DHT Node CA");
@@ -53,6 +50,8 @@ 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";
 }
@@ -281,7 +280,7 @@ DhtProxyTester::testFuzzy()
         },[&](bool ok){
             CPPUNIT_ASSERT(ok);
         });
-        std::this_thread::sleep_for(5ms);
+        std::this_thread::sleep_for(10ms);
     }
 
     // Assert

tests/dhtrunnertester.cpp

@@ -29,12 +29,8 @@ CPPUNIT_TEST_SUITE_REGISTRATION(DhtRunnerTester);
 
 void
 DhtRunnerTester::setUp() {
-    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);
+    node1.run(42222);
+    node2.run(42232);
     node2.bootstrap(node1.getBound());
 }
 
@@ -88,9 +84,7 @@ DhtRunnerTester::testListen() {
     auto a = dht::InfoHash::get("234");
     auto b = dht::InfoHash::get("2345");
     auto c = dht::InfoHash::get("23456");
-    auto d = dht::InfoHash::get("234567");
-    constexpr unsigned N = 256;
-    constexpr unsigned SZ = 56 * 1024;
+    constexpr unsigned N = 32;
 
     auto ftokena = node1.listen(a, [&](const std::shared_ptr<dht::Value>&) {
         valueCount++;
@@ -107,16 +101,6 @@ 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);
@@ -130,41 +114,29 @@ 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 * 3u; }));
-        CPPUNIT_ASSERT_EQUAL(N * 3u, putOkCount);
+        CPPUNIT_ASSERT(cv.wait_for(lk, 30s, [&]{ return putCount == N * 2u; }));
+        CPPUNIT_ASSERT_EQUAL(N * 2u, 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(tokend);
-    CPPUNIT_ASSERT_EQUAL(N * 2u + 1u, valueCount.load());
+    CPPUNIT_ASSERT_EQUAL(N + 1u, valueCount.load());
 
     node1.cancelListen(a, tokena);
     node1.cancelListen(b, std::move(ftokenb));
     node1.cancelListen(c, tokenc);
-    node1.cancelListen(d, tokend);
 }
 
 }  // namespace test

tools/dhtnode.cpp

@@ -145,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;