request: cleanup

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

src/network_engine.cpp

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

src/request.h

@@ -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. */
 };
 

src/value.cpp

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

tests/dhtproxytester.cpp

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

tests/dhtrunnertester.cpp

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