update to version 1.7.4

dhtproxyclient: fix ok when a permanent PUT operations fails

CMakeLists.txt

@@ -1,7 +1,7 @@
 cmake_minimum_required (VERSION 3.1)
 project (opendht)
 set (opendht_VERSION_MAJOR 1)
-set (opendht_VERSION_MINOR 7.2)
+set (opendht_VERSION_MINOR 7.4)
 set (opendht_VERSION ${opendht_VERSION_MAJOR}.${opendht_VERSION_MINOR})
 set (PACKAGE_VERSION ${opendht_VERSION})
 set (VERSION "${opendht_VERSION}")

MSVC/opendht.vcxproj

@@ -146,7 +146,7 @@
       <Optimization>Disabled</Optimization>
       <SDLCheck>true</SDLCheck>
       <AdditionalIncludeDirectories>..\..\include;..\include;..\include\opendht;..\..\msgpack-c\include;</AdditionalIncludeDirectories>
-      <PreprocessorDefinitions>_CRT_SECURE_NO_WARNINGS;WIN32_NATIVE;WIN32_LEAN_AND_MEAN;_MBCS;%(PreprocessorDefinitions)</PreprocessorDefinitions>
+      <PreprocessorDefinitions>_CRT_SECURE_NO_WARNINGS;WIN32_LEAN_AND_MEAN;_MBCS;%(PreprocessorDefinitions)</PreprocessorDefinitions>
       <DisableSpecificWarnings>4804;4800;4101;4267;4244;4503;</DisableSpecificWarnings>
       <AdditionalOptions>-D_SCL_SECURE_NO_WARNINGS %(AdditionalOptions)</AdditionalOptions>
       <SuppressStartupBanner>false</SuppressStartupBanner>
@@ -169,7 +169,7 @@ copy ..\include\opendht.h $(OutDir)\include\</Command>
       <Optimization>Disabled</Optimization>
       <SDLCheck>true</SDLCheck>
       <AdditionalIncludeDirectories>..\..\include;..\include;..\include\opendht;..\..\msgpack-c\include;</AdditionalIncludeDirectories>
-      <PreprocessorDefinitions>_CRT_SECURE_NO_WARNINGS;WIN32_NATIVE;WIN32_LEAN_AND_MEAN;_MBCS;%(PreprocessorDefinitions)</PreprocessorDefinitions>
+      <PreprocessorDefinitions>_CRT_SECURE_NO_WARNINGS;WIN32_LEAN_AND_MEAN;_MBCS;%(PreprocessorDefinitions)</PreprocessorDefinitions>
       <DisableSpecificWarnings>4804;4800;4101;4267;4244;4503;</DisableSpecificWarnings>
       <AdditionalOptions>-D_SCL_SECURE_NO_WARNINGS %(AdditionalOptions)</AdditionalOptions>
     </ClCompile>
@@ -193,7 +193,7 @@ copy ..\include\opendht.h $(OutDir)\include\</Command>
       <IntrinsicFunctions>true</IntrinsicFunctions>
       <SDLCheck>true</SDLCheck>
       <AdditionalIncludeDirectories>..\..\include;..\include;..\include\opendht;..\..\msgpack-c\include;</AdditionalIncludeDirectories>
-      <PreprocessorDefinitions>_CRT_SECURE_NO_WARNINGS;WIN32_NATIVE;WIN32_LEAN_AND_MEAN;_MBCS;%(PreprocessorDefinitions)</PreprocessorDefinitions>
+      <PreprocessorDefinitions>_CRT_SECURE_NO_WARNINGS;WIN32_LEAN_AND_MEAN;_MBCS;%(PreprocessorDefinitions)</PreprocessorDefinitions>
       <DisableSpecificWarnings>4804;4800;4101;4267;4244;4503;</DisableSpecificWarnings>
       <AdditionalOptions>-D_SCL_SECURE_NO_WARNINGS %(AdditionalOptions)</AdditionalOptions>
       <ProgramDataBaseFileName>$(OutDir)\lib\x86\$(TargetName).pdb</ProgramDataBaseFileName>
@@ -223,7 +223,7 @@ copy ..\include\opendht.h $(OutDir)\include\</Command>
       <IntrinsicFunctions>true</IntrinsicFunctions>
       <SDLCheck>true</SDLCheck>
       <AdditionalIncludeDirectories>$(ProjectDir)contrib\build\include;$(ProjectDir)..\include;$(ProjectDir)..\include\opendht;$(ProjectDir)contrib\build\msgpack-c\include;$(ProjectDir)contrib\build\argon2\include;$(ProjectDir)..\..\include;$(ProjectDir)..\..\argon2\include</AdditionalIncludeDirectories>
-      <PreprocessorDefinitions>_CRT_SECURE_NO_WARNINGS;WIN32_NATIVE;WIN32_LEAN_AND_MEAN;_MBCS;%(PreprocessorDefinitions)</PreprocessorDefinitions>
+      <PreprocessorDefinitions>_CRT_SECURE_NO_WARNINGS;WIN32_LEAN_AND_MEAN;_MBCS;%(PreprocessorDefinitions)</PreprocessorDefinitions>
       <DisableSpecificWarnings>4804;4800;4101;4267;4244;4503;4273;</DisableSpecificWarnings>
       <AdditionalOptions>-D_SCL_SECURE_NO_WARNINGS %(AdditionalOptions)</AdditionalOptions>
       <ProgramDataBaseFileName>$(OutDir)\lib\x64\$(TargetName).pdb</ProgramDataBaseFileName>

configure.ac

@@ -1,7 +1,7 @@
 dnl define macros
 m4_define([opendht_major_version], 1)
 m4_define([opendht_minor_version], 7)
-m4_define([opendht_patch_version], 2)
+m4_define([opendht_patch_version], 4)
 m4_define([opendht_version],
 		  [opendht_major_version.opendht_minor_version.opendht_patch_version])
 

include/opendht/callbacks.h

@@ -83,16 +83,6 @@ struct OPENDHT_PUBLIC NodeInfo {
     MSGPACK_DEFINE_MAP(id, node_id, ipv4, ipv6)
 };
 
-struct OPENDHT_PUBLIC MessageStats {
-    unsigned packets {0};
-    unsigned ping    {0};
-    unsigned find    {0};
-    unsigned get     {0};
-    unsigned put     {0};
-    unsigned listen  {0};
-    unsigned refresh {0};
-};
-
 /**
  * Dht configuration.
  */

include/opendht/network_engine.h

@@ -27,7 +27,6 @@
 #include "utils.h"
 #include "rng.h"
 #include "rate_limiter.h"
-#include "callbacks.h"
 
 #include <vector>
 #include <string>
@@ -409,10 +408,6 @@ public:
         return stats;
     }
 
-    MessageStats getMessageStats(bool in) {
-        return in ? in_stats : out_stats;
-    }
-
     void blacklistNode(const Sp<Node>& n);
 
     std::vector<Sp<Node>> getCachedNodes(const InfoHash& id, sa_family_t sa_f, size_t count) {
@@ -462,6 +457,16 @@ 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};
+    };
+
+
     // basic wrapper for socket sendto function
     int send(const char *buf, size_t len, int flags, const SockAddr& addr);
 

include/opendht/rng.h

@@ -25,7 +25,7 @@
 namespace dht {
 namespace crypto {
 
-#ifndef WIN32_NATIVE
+#ifndef _MSC_VER
 #ifdef _WIN32
 
 /**

src/dht.cpp

@@ -1585,11 +1585,6 @@ Dht::getStorageLog() const
     std::stringstream out;
     for (const auto& s : store)
         out << printStorageLog(s);
-    out << "Total " << store.size() << " storages, " << total_values << " values (";
-    if (total_store_size < 1024)
-        out << total_store_size << " bytes)";
-    else
-        out << (total_store_size/1024) << " KB)";
     out << std::endl << std::endl;
     std::multimap<size_t, const SockAddr*> q_map;
     for (const auto& ip : store_quota)
@@ -1598,6 +1593,12 @@ Dht::getStorageLog() const
     for (auto ip = q_map.rbegin(); ip != q_map.rend(); ++ip)
         out << "IP " << ip->second->toString() << " uses " << ip->first << " bytes" << std::endl;
     out << std::endl;
+    out << "Total " << store.size() << " storages, " << total_values << " values (";
+    if (total_store_size < 1024)
+        out << total_store_size << " bytes)";
+    else
+        out << (total_store_size/1024) << " / " << (max_store_size/1024) << " KB)";
+    out << std::endl;
     return out.str();
 }
 

src/dht_proxy_client.cpp

@@ -342,7 +342,7 @@ DhtProxyClient::put(const InfoHash& key, Sp<Value> val, DoneCallback cb, time_po
                 if (p != s->second.puts.end()) {
                     doPut(key, p->second.value,
                     [ok](bool result, const std::vector<std::shared_ptr<dht::Node> >&){
-                        *ok = !result;
+                        *ok = result;
                     }, time_point::max(), true);
                     scheduler.edit(p->second.refreshJob, scheduler.time() + proxy::OP_TIMEOUT - proxy::OP_MARGIN);
                 }
@@ -951,11 +951,11 @@ DhtProxyClient::restartListeners()
     DHT_LOG.d("Refresh permanent puts");
     for (auto& search : searches_) {
         for (auto& put : search.second.puts) {
-            if (!put.second.ok) {
+            if (!*put.second.ok) {
                 auto ok = put.second.ok;
                 doPut(search.first, put.second.value,
                 [ok](bool result, const std::vector<std::shared_ptr<dht::Node> >&){
-                    *ok = !result;
+                    *ok = result;
                 }, time_point::max(), true);
                 scheduler.edit(put.second.refreshJob, scheduler.time() + proxy::OP_TIMEOUT - proxy::OP_MARGIN);
             }

src/dhtrunner.cpp

@@ -410,25 +410,15 @@ DhtRunner::loop_()
         received = std::move(rcv);
     }
     if (not received.empty()) {
-        size_t packet_num = received.size();
-        auto start = clock::now();
         while (not received.empty()) {
             auto& pck = received.front();
-            auto now = clock::now();
-            auto delay = now - pck.received;
-            if (delay > std::chrono::seconds(1))
+            auto delay = clock::now() - pck.received;
+            if (delay > std::chrono::milliseconds(500))
                 std::cerr << "Dropping packet with high delay: " << print_dt(delay) << std::endl;
-            else {
+            else
                 wakeup = dht->periodic(pck.data.data(), pck.data.size()-1, pck.from);
-                auto end = clock::now();
-                std::cerr << "periodic took " << print_dt(end - now)*1000 << std::endl;
-            }
             received.pop();
         }
-        if (packet_num > 1) {
-            auto time_all = print_dt(clock::now() - start);
-            std::cerr << "Treating " << packet_num << " packets took " << time_all << " - " << (packet_num/time_all) << " packets/sec" << std::endl;
-        }
     } else {
         wakeup = dht->periodic(nullptr, 0, nullptr, 0);
     }
@@ -536,7 +526,7 @@ DhtRunner::startNetwork(const SockAddr sin4, const SockAddr sin6)
                             std::lock_guard<std::mutex> lck(sock_mtx);
                             if (rcv.size() >= RX_QUEUE_MAX_SIZE) {
                                 std::cerr << "Dropping packet: queue is full!" << std::endl;
-                                continue;
+                                rcv.pop();
                             }
                             rcv.emplace(ReceivedPacket {Blob {buf.begin(), buf.begin()+rc+1}, SockAddr(from, from_len), clock::now()});
                         }
@@ -593,6 +583,7 @@ DhtRunner::listen(InfoHash hash, ValueCallback vcb, Value::Filter f, Where w)
     {
         std::lock_guard<std::mutex> lck(storage_mtx);
         pending_ops.emplace([=](SecureDht& dht) mutable {
+#if OPENDHT_PROXY_CLIENT
             auto tokenbGlobal = listener_token_++;
             Listener listener {};
             listener.hash = hash;
@@ -600,21 +591,20 @@ DhtRunner::listen(InfoHash hash, ValueCallback vcb, Value::Filter f, Where w)
             listener.w = std::move(w);
             listener.gcb = [hash,vcb,tokenbGlobal,this](const std::vector<Sp<Value>>& vals, bool expired){
                 if (not vcb(vals, expired)) {
-#if OPENDHT_PROXY_CLIENT
                     cancelListen(hash, tokenbGlobal);
-#endif
                     return false;
                 }
                 return true;
             };
-#if OPENDHT_PROXY_CLIENT
             if (use_proxy)
                 listener.tokenProxyDht = dht.listen(hash, listener.gcb, listener.f, listener.w);
             else
-#endif
                 listener.tokenClassicDht = dht.listen(hash, listener.gcb, listener.f, listener.w);
             listeners_.emplace(tokenbGlobal, std::move(listener));
             ret_token->set_value(tokenbGlobal);
+#else
+            ret_token->set_value(dht.listen(hash, vcb, f, w));
+#endif
         });
     }
     cv.notify_all();
@@ -908,8 +898,8 @@ DhtRunner::findCertificate(InfoHash hash, std::function<void(const std::shared_p
 void
 DhtRunner::resetDht()
 {
-    listeners_.clear();
 #if OPENDHT_PROXY_CLIENT
+    listeners_.clear();
     dht_via_proxy_.reset();
 #endif // OPENDHT_PROXY_CLIENT
     dht_.reset();

src/network_engine.cpp

@@ -403,7 +403,6 @@ NetworkEngine::isNodeBlacklisted(const SockAddr& addr) const
 void
 NetworkEngine::processMessage(const uint8_t *buf, size_t buflen, const SockAddr& from)
 {
-    in_stats.packets++;
     if (isMartian(from)) {
         DHT_LOG.w("Received packet from martian node %s", from.toString().c_str());
         return;
@@ -664,8 +663,6 @@ NetworkEngine::send(const char *buf, size_t len, int flags, const SockAddr& addr
 #ifdef MSG_NOSIGNAL
     flags |= MSG_NOSIGNAL;
 #endif
-
-    out_stats.packets++;
     if (sendto(s, buf, len, flags, addr.get(), addr.getLength()) == -1) {
         int err = errno;
         DHT_LOG.e("Can't send message to %s: %s", addr.toString().c_str(), strerror(err));

tests/dhtproxytester.cpp

@@ -116,10 +116,7 @@ DhtProxyTester::testListen() {
     // All values
     dht::Value secondVal {"You're a monster"};
     auto secondVal_data = secondVal.data;
-    nodePeer.put(key, std::move(secondVal), [&](bool) {
-        done = true;
-        cv.notify_all();
-    });
+    nodePeer.put(key, std::move(secondVal));
     cv.wait_for(lk, std::chrono::seconds(10), [&]{ return done; });
     // Here values should contains 3 values
     CPPUNIT_ASSERT_EQUAL(static_cast<int>(values.size()), 2);