Revert "search: use pointer to SearchNode"

This reverts commit 3a41f2ce56.
search: improve canGet performance
2026-11-08 20:37:23 +00:00 · 2018-06-20 14:40:41 -04:00 · 2018-06-20 14:01:48 -04:00 · 2018-06-20 14:01:48 -04:00 · 2018-06-20 14:01:48 -04:00
37 changed files with 368 additions and 709 deletions
@@ -1,4 +1,3 @@
-dist: xenial
 sudo: required

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

 if (OPENDHT_PROXY_SERVER OR OPENDHT_PROXY_CLIENT)
    find_package(Restbed REQUIRED)
-    if (NOT Jsoncpp_FOUND)
+    if (!Jsoncpp_FOUND)
        message(SEND_ERROR "Jsoncpp is required for DHT proxy support")
    endif()
 endif()
@@ -99,8 +99,8 @@ endif ()
 if (Nettle_INCLUDE_DIRS)
    include_directories (SYSTEM "${Nettle_INCLUDE_DIRS}")
 endif ()
-if (Restbed_INCLUDE_DIR)
-    include_directories (SYSTEM "${Restbed_INCLUDE_DIR}")
+if (RESTBED_INCLUDE)
+    include_directories (SYSTEM "${RESTBED_INCLUDE}")
 endif ()
 if (Jsoncpp_INCLUDE_DIRS)
    include_directories (SYSTEM "${Jsoncpp_INCLUDE_DIRS}")
@@ -246,14 +246,16 @@ if (OPENDHT_STATIC)
        ${opendht_HEADERS}
    )
    set_target_properties (opendht-static PROPERTIES OUTPUT_NAME "opendht")
+    if (OPENDHT_LTO)
+        target_link_libraries(opendht-static -flto -fuse-linker-plugin)
+    endif ()
    if (OPENDHT_ARGON2)
        target_include_directories(opendht-static SYSTEM PRIVATE argon2)
    else ()
+        target_link_libraries(opendht-static ${argon2_LIBRARIES})
        target_include_directories(opendht-static SYSTEM PRIVATE ${argon2_INCLUDE_DIRS})
    endif ()
-    target_link_libraries(opendht-static
-        PRIVATE ${Restbed_LIBRARY} ${argon2_LIBRARIES}
-        PUBLIC ${CMAKE_THREAD_LIBS_INIT} ${GNUTLS_LIBRARIES} ${Nettle_LIBRARIES} ${Jsoncpp_LIBRARIES})
+    target_link_libraries(opendht-static ${CMAKE_THREAD_LIBS_INIT} ${GNUTLS_LIBRARIES} ${Nettle_LIBRARIES} ${Restbed_LIBRARIES} ${Jsoncpp_LIBRARIES})
    install (TARGETS opendht-static DESTINATION ${CMAKE_INSTALL_LIBDIR} EXPORT opendht)
 endif ()

@@ -265,13 +267,16 @@ if (OPENDHT_SHARED)
    set_target_properties (opendht PROPERTIES IMPORT_SUFFIX "_import.lib")
    set_target_properties (opendht PROPERTIES SOVERSION ${opendht_VERSION_MAJOR} VERSION ${opendht_VERSION})
    target_compile_definitions(opendht PRIVATE OPENDHT_BUILD)
+    if (OPENDHT_LTO)
+        target_link_libraries(opendht PRIVATE -flto -fuse-linker-plugin)
+    endif ()
    if (OPENDHT_ARGON2)
        target_include_directories(opendht SYSTEM PRIVATE argon2)
    else ()
        target_link_libraries(opendht PRIVATE ${argon2_LIBRARIES})
        target_include_directories(opendht SYSTEM PRIVATE ${argon2_INCLUDE_DIRS})
    endif ()
-    target_link_libraries(opendht PRIVATE ${CMAKE_THREAD_LIBS_INIT} ${GNUTLS_LIBRARIES} ${Nettle_LIBRARIES} ${Restbed_LIBRARY} ${Jsoncpp_LIBRARIES})
+    target_link_libraries(opendht PRIVATE ${CMAKE_THREAD_LIBS_INIT} ${GNUTLS_LIBRARIES} ${Nettle_LIBRARIES} ${Restbed_LIBRARIES} ${Jsoncpp_LIBRARIES})
    install (TARGETS opendht DESTINATION ${CMAKE_INSTALL_LIBDIR} EXPORT opendht)
 endif ()

@@ -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_LEAN_AND_MEAN;_MBCS;%(PreprocessorDefinitions)</PreprocessorDefinitions>
+      <PreprocessorDefinitions>_CRT_SECURE_NO_WARNINGS;WIN32_NATIVE;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_LEAN_AND_MEAN;_MBCS;%(PreprocessorDefinitions)</PreprocessorDefinitions>
+      <PreprocessorDefinitions>_CRT_SECURE_NO_WARNINGS;WIN32_NATIVE;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_LEAN_AND_MEAN;_MBCS;%(PreprocessorDefinitions)</PreprocessorDefinitions>
+      <PreprocessorDefinitions>_CRT_SECURE_NO_WARNINGS;WIN32_NATIVE;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_LEAN_AND_MEAN;_MBCS;%(PreprocessorDefinitions)</PreprocessorDefinitions>
+      <PreprocessorDefinitions>_CRT_SECURE_NO_WARNINGS;WIN32_NATIVE;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>
@@ -1,16 +1,17 @@
-if(NOT Restbed_FOUND)
-    find_path (Restbed_INCLUDE_DIR restbed
-               HINTS
-               "/usr/include"
-               "/usr/local/include"
-               "/opt/local/include")
-    find_library(Restbed_LIBRARY restbed
-                 HINTS ${Restbed_ROOT_DIR} PATH_SUFFIXES lib)
-    include(FindPackageHandleStandardArgs)
-    find_package_handle_standard_args(Restbed DEFAULT_MSG Restbed_LIBRARY Restbed_INCLUDE_DIR)
-    if (Restbed_INCLUDE_DIR)
-        set(Restbed_FOUND TRUE)
-        set(Restbed_LIBRARIES ${Restbed_LIBRARY})
-        set(Restbed_INCLUDE_DIRS ${Restbed_INCLUDE_DIR})
-    endif()
+find_path (RESTBED_INCLUDE restbed
+           HINTS
+           "/usr/include"
+           "/usr/local/include"
+           "/opt/local/include"
+)
+
+if (RESTBED_INCLUDE)
+    message(STATUS "${green}Found Restbed: ${RESTBED_INCLUDE}")
+else()
+    message(FATAL_ERROR "${red}Failed to locate Restbed.}")
+endif()
+
+if (RESTBED_INCLUDE)
+    set(RESTBED_FOUND TRUE)
+    set(Restbed_LIBRARIES restbed)
 endif()
@@ -1,7 +1,7 @@
 dnl define macros
 m4_define([opendht_major_version], 1)
-m4_define([opendht_minor_version], 8)
-m4_define([opendht_patch_version], 0)
+m4_define([opendht_minor_version], 7)
+m4_define([opendht_patch_version], 2)
 m4_define([opendht_version],
 		  [opendht_major_version.opendht_minor_version.opendht_patch_version])

@@ -130,7 +130,7 @@ PKG_CHECK_MODULES([GnuTLS], [gnutls >= 3.3])
 PKG_CHECK_MODULES([MsgPack], [msgpack >= 1.2])

 AC_ARG_WITH([jsoncpp], AS_HELP_STRING([--without-jsoncpp], [Build without JsonCpp support]))
-AS_IF([test "x$with_jsoncpp" != "xno"],
+AS_IF([test "x$with_jsoncpp" != "xno"], 
      [PKG_CHECK_MODULES([JsonCpp], [jsoncpp >= 1.7.2], [have_jsoncpp=yes], [have_jsoncpp=no])],
      [have_jsoncpp=no])
 AS_IF([test "x$have_jsoncpp" = "xyes"], [
@@ -51,7 +51,6 @@ struct OPENDHT_PUBLIC NodeStats {
             incoming_nodes {0};
    unsigned table_depth {0};
    unsigned getKnownNodes() const { return good_nodes + dubious_nodes; }
-    unsigned long getNetworkSizeEstimation() const { return 8 * std::exp2(table_depth); }
    std::string toString() const;

 #ifdef OPENDHT_JSONCPP
@@ -84,6 +83,16 @@ 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.
 */
@@ -86,16 +86,8 @@ struct OPENDHT_PUBLIC PublicKey

    PublicKey& operator=(PublicKey&& o) noexcept;

-    /**
-     * Get public key fingerprint
-     */
    InfoHash getId() const;
-
-    /**
-     * Get public key long fingerprint
-     */
    PkId getLongId() const;
-
    bool checkSignature(const Blob& data, const Blob& signature) const;
    Blob encrypt(const Blob&) const;

@@ -141,7 +133,6 @@ struct OPENDHT_PUBLIC PrivateKey
    PrivateKey(const Blob& import, const std::string& password = {});
    ~PrivateKey();
    explicit operator bool() const { return key; }
-
    PublicKey getPublicKey() const;
    Blob serialize(const std::string& password = {}) const;

@@ -377,7 +368,6 @@ struct OPENDHT_PUBLIC Certificate {

    /** Same as getPublicKey().getId() */
    InfoHash getId() const;
-    /** Same as getPublicKey().getLongId() */
    PkId getLongId() const;

    /** Read certificate Common Name (CN) */
@@ -414,20 +404,8 @@ struct OPENDHT_PUBLIC Certificate {

    std::string print() const;

-    /**
-     * As a CA, revoke a certificate, adding it to
-     * the attached Certificate Revocation List (CRL)
-     */
    void revoke(const PrivateKey&, const Certificate&);
-
-    /**
-     * Get the list of certificates revoked as as CA.
-     */
    std::vector<std::shared_ptr<RevocationList>> getRevocationLists() const;
-
-    /**
-     * Attach existing revocation list.
-     */
    void addRevocationList(RevocationList&&);
    void addRevocationList(std::shared_ptr<RevocationList>);

@@ -530,13 +508,9 @@ public:
    ~secure_vector() { clean(); }

    static secure_vector<T> getRandom(size_t size) {
-        secure_vector<T> ret(size);
+        secure_vector<T> ret(size/sizeof(T));
        crypto::random_device rdev;
-#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
        std::generate_n((uint8_t*)ret.data_.data(), ret.size()*sizeof(T), std::bind(rand_byte, std::ref(rdev)));
        return ret;
    }
@@ -409,9 +409,7 @@ private:

    // Storage
    void storageAddListener(const InfoHash& id, const Sp<Node>& node, size_t tid, Query&& = {});
-    bool storageStore(const InfoHash& id, const Sp<Value>& value, time_point created, const SockAddr& sa = {}, bool permanent = false);
-    bool storageErase(const InfoHash& id, Value::Id vid);
-    bool storageRefresh(const InfoHash& id, Value::Id vid);
+    bool storageStore(const InfoHash& id, const Sp<Value>& value, time_point created, const SockAddr& sa = {});
    void expireStore();
    void expireStorage(InfoHash h);
    void expireStore(decltype(store)::iterator);
@@ -466,7 +464,7 @@ private:
     * Low-level method that will perform a search on the DHT for the specified
     * infohash (id), using the specified IP version (IPv4 or IPv6).
     */
-    Sp<Search> search(const InfoHash& id, sa_family_t af, GetCallback = {}, QueryCallback = {}, DoneCallback = {}, Value::Filter = {}, const Sp<Query>& q = {});
+    Sp<Search> search(const InfoHash& id, sa_family_t af, GetCallback = {}, QueryCallback = {}, DoneCallback = {}, Value::Filter = {}, Query q = {});

    void announce(const InfoHash& id, sa_family_t af, Sp<Value> value, DoneCallback callback, time_point created=time_point::max(), bool permanent = false);
    size_t listenTo(const InfoHash& id, sa_family_t af, ValueCallback cb, Value::Filter f = Value::AllFilter(), const Sp<Query>& q = {});
@@ -40,7 +40,7 @@ public:
    virtual NodeStatus getStatus() const = 0;

    /**
-     * Get the ID of the DHT node.
+     * Get the ID of the node.
     */
    virtual const InfoHash& getNodeId() const = 0;

@@ -216,7 +216,7 @@ public:
     */
    virtual void setLoggers(LogMethod error = NOLOG, LogMethod warn = NOLOG, LogMethod debug = NOLOG)
    {
-        DHT_LOG.DBG = debug;
+        DHT_LOG.DEBUG = debug;
        DHT_LOG.WARN = warn;
        DHT_LOG.ERR = error;
    }
@@ -34,10 +34,6 @@
 #include <mutex>
 #include <restbed>

-#ifdef OPENDHT_JSONCPP
-#include <json/json.h>
-#endif
-
 namespace Json {
    class Value;
 }
@@ -77,44 +73,15 @@ public:
        size_t pushListenersCount;
        /** Average requests per second */
        double requestRate;
-        /** Node Info **/
-        NodeInfo nodeInfo;

        std::string toString() const {
            std::ostringstream ss;
            ss << "Listens: " << listenCount << " Puts: " << putCount << " PushListeners: " << pushListenersCount << std::endl;
            ss << "Requests: " << requestRate << " per second." << std::endl;
-            auto& ni = nodeInfo;
-            auto& ipv4 = ni.ipv4;
-            if (ipv4.table_depth > 1) {
-                ss << "IPv4 Network estimation: " << ipv4.getNetworkSizeEstimation() << std::endl;;
-            }
-            auto& ipv6 = ni.ipv6;
-            if (ipv6.table_depth > 1) {
-                ss << "IPv6 Network estimation: " << ipv6.getNetworkSizeEstimation() << std::endl;;
-            }
            return ss.str();
        }
-
-#ifdef OPENDHT_JSONCPP
-        /**
-         * Build a json object from a NodeStats
-         */
-        Json::Value toJson() const {
-            Json::Value result;
-            result["listenCount"] = static_cast<Json::UInt64>(listenCount);
-            result["putCount"] = static_cast<Json::UInt64>(putCount);
-            result["pushListenersCount"] = static_cast<Json::UInt64>(pushListenersCount);
-            result["requestRate"] = requestRate;
-            result["nodeInfo"] = nodeInfo.toJson();
-            return result;
-        }
-#endif
    };
-
-    ServerStats stats() const { return stats_; }
-
-    void updateStats() const;
+    ServerStats getStats() const;

    std::shared_ptr<DhtRunner> getNode() const { return dht_; }

@@ -133,14 +100,6 @@ private:
     */
    void getNodeInfo(const std::shared_ptr<restbed::Session>& session) const;

-    /**
-     * Return ServerStats in JSON format
-     * Method: STATS "/"
-     * Result: HTTP 200, body: Node infos in JSON format
-     * @param session
-     */
-    void getStats(const std::shared_ptr<restbed::Session>& session) const;
-
    /**
     * Return Values of an infoHash
     * Method: GET "/{InfoHash: .*}"
@@ -303,8 +262,6 @@ private:

    const std::string pushServer_;

-    mutable ServerStats stats_;
-
 #if OPENDHT_PUSH_NOTIFICATIONS
    struct Listener;
    struct PushListener;
@@ -158,15 +158,6 @@ public:
        },
        getFilterSet<T>());
    }
-    template <class T>
-    std::future<size_t> listen(InfoHash hash, std::function<bool(std::vector<T>&&, bool)> cb)
-    {
-        return listen(hash, [=](const std::vector<std::shared_ptr<Value>>& vals, bool expired) {
-            return cb(unpackVector<T>(vals), expired);
-        },
-        getFilterSet<T>());
-    }
-
    template <typename T>
    std::future<size_t> listen(InfoHash hash, std::function<bool(T&&)> cb, Value::Filter f = Value::AllFilter(), Where w = {})
    {
@@ -183,22 +174,6 @@ public:
        },
        getFilterSet<T>(f), w);
    }
-    template <typename T>
-    std::future<size_t> listen(InfoHash hash, std::function<bool(T&&, bool)> cb, Value::Filter f = Value::AllFilter(), Where w = {})
-    {
-        return listen(hash, [=](const std::vector<std::shared_ptr<Value>>& vals, bool expired) {
-            for (const auto& v : vals) {
-                try {
-                    if (not cb(Value::unpack<T>(*v), expired))
-                        return false;
-                } catch (const std::exception&) {
-                    continue;
-                }
-            }
-            return true;
-        },
-        getFilterSet<T>(f), w);
-    }

    void cancelListen(InfoHash h, size_t token);
    void cancelListen(InfoHash h, std::shared_future<size_t> token);
@@ -273,14 +248,8 @@ public:

    void dumpTables() const;

-    /**
-     * Get the public key fingerprint if an identity is used with this node, 0 otherwise.
-     */
    InfoHash getId() const;

-    /**
-     * Get the ID of the DHT node.
-     */
    InfoHash getNodeId() const;

    /**
@@ -447,7 +416,6 @@ private:
     */
    void tryBootstrapContinuously();

-    void stopNetwork();
    void startNetwork(const SockAddr sin4, const SockAddr sin6);
    time_point loop_();

@@ -518,7 +486,6 @@ private:
               status6 {NodeStatus::Disconnected};
    StatusCallback statusCb {nullptr};

-    int stop_writefd {-1};
    int s4 {-1}, s6 {-1};
    SockAddr bound4 {};
    SockAddr bound6 {};
@@ -85,7 +85,7 @@ private:
 };

 struct Logger {
-    LogMethod DBG = NOLOG;
+    LogMethod DEBUG = NOLOG;
    LogMethod WARN = NOLOG;
    LogMethod ERR = NOLOG;
    void setFilter(const InfoHash& f) {
@@ -114,7 +114,7 @@ struct Logger {
 #if OPENDHT_LOG
        va_list args;
        va_start(args, format);
-        log0(DBG, format, args);
+        log0(DEBUG, format, args);
        va_end(args);
 #endif
    }
@@ -122,7 +122,7 @@ struct Logger {
 #if OPENDHT_LOG
        va_list args;
        va_start(args, format);
-        log1(DBG, f, format, args);
+        log1(DEBUG, f, format, args);
        va_end(args);
 #endif
    }
@@ -130,7 +130,7 @@ struct Logger {
 #if OPENDHT_LOG
        va_list args;
        va_start(args, format);
-        log2(DBG, f1, f2, format, args);
+        log2(DEBUG, f1, f2, format, args);
        va_end(args);
 #endif
    }
@@ -27,6 +27,7 @@
 #include "utils.h"
 #include "rng.h"
 #include "rate_limiter.h"
+#include "callbacks.h"

 #include <vector>
 #include <string>
@@ -408,6 +409,10 @@ 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) {
@@ -457,16 +462,6 @@ 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);

@@ -40,16 +40,12 @@ struct NodeCache {
    ~NodeCache();

 private:
-    class NodeMap : private std::map<InfoHash, std::weak_ptr<Node>> {
+    class NodeMap : public 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);
-        std::vector<Sp<Node>> getCachedNodes(const InfoHash& id, size_t count) const;
        void clearBadNodes();
        void setExpired();
-        void cleanup();
-    private:
-        size_t cleanup_counter {0};
    };

    const NodeMap& cache(sa_family_t af) const { return af == AF_INET ? cache_4 : cache_6; }
@@ -25,7 +25,7 @@
 namespace dht {
 namespace crypto {

-#ifndef _MSC_VER
+#ifndef WIN32_NATIVE
 #ifdef _WIN32

 /**
@@ -310,9 +310,11 @@ public:
        dht_->pushNotificationReceived(notification);
    }

-    void setLoggers(LogMethod error = NOLOG, LogMethod warn = NOLOG, LogMethod debug = NOLOG) override
+    void setLoggers(LogMethod error = NOLOG, LogMethod warn = NOLOG, LogMethod debug = NOLOG)
    {
-        DhtInterface::setLoggers(error, warn, debug);
+        DHT_LOG.DEBUG = debug;
+        DHT_LOG.WARN = warn;
+        DHT_LOG.ERR = error;
        dht_->setLoggers(error, warn, debug);
    }

@@ -67,7 +67,7 @@ public:
     */
    SockAddr(const sockaddr* sa, socklen_t length) {
        if (length > sizeof(sockaddr_storage))
-            throw std::invalid_argument("Socket address length is too large");
+            throw std::runtime_error("Socket address length is too large");
        set(sa, length);
    }
    SockAddr(const sockaddr* sa) {
@@ -78,7 +78,7 @@ public:
            else if(sa->sa_family == AF_INET6)
                len = sizeof(sockaddr_in6);
            else
-                throw std::invalid_argument("Unknown address family");
+                throw std::runtime_error("Unknown address family");
        }
        set(sa, len);
    }
@@ -122,8 +122,6 @@ private:
    std::map<ValueType::Id, ValueType> types {};
 };

-struct CryptoValueCache;
-
 /**
 * A "value" is data potentially stored on the Dht, with some metadata.
 *
@@ -590,16 +588,6 @@ struct OPENDHT_PUBLIC Value
     * Hold encrypted version of the data.
     */
    Blob cypher {};
-
-    bool checkSignature();
-    Sp<Value> decrypt(const crypto::PrivateKey& key);
-
-private:
-    /* Cache for crypto ops */
-    bool signatureChecked {false};
-    bool signatureValid {false};
-    bool decrypted {false};
-    Sp<Value> decryptedValue {};
 };

 using ValuesExport = std::pair<InfoHash, Blob>;
@@ -1,6 +1,6 @@
 lib_LTLIBRARIES = libopendht.la

-libopendht_la_CPPFLAGS = @CPPFLAGS@ -I$(top_srcdir)/include/opendht @Argon2_CFLAGS@ @JsonCpp_CFLAGS@ @MsgPack_CFLAGS@
+libopendht_la_CPPFLAGS = @CPPFLAGS@ -I$(top_srcdir)/include/opendht @Argon2_CFLAGS@ @JsonCpp_CFLAGS@
 libopendht_la_LIBADD   = @Argon2_LIBS@ @JsonCpp_LIBS@ @GnuTLS_LIBS@ @Nettle_LIBS@
 libopendht_la_LDFLAGS  = @LDFLAGS@ @Argon2_LDFLAGS@
 libopendht_la_SOURCES  = \
@@ -65,8 +65,8 @@ libopendht_la_SOURCES += dht_proxy_client.cpp
 nobase_include_HEADERS += ../include/opendht/dht_proxy_client.h ../include/opendht/dht_interface.h
 endif

-libopendht_la_SOURCES += base64.h base64.cpp
 if PROXY_CLIENT_OR_SERVER
+libopendht_la_SOURCES += base64.h base64.cpp
 nobase_include_HEADERS += ../include/opendht/proxy.h
 endif

@@ -62,7 +62,8 @@ NodeStats::toString() const
    ss << "Known nodes: " << good_nodes << " good, " << dubious_nodes << " dubious, " << incoming_nodes << " incoming." << std::endl;
    if (table_depth > 1) {
        ss << "Routing table depth: " << table_depth << std::endl;
-        ss << "Network size estimation: " << getNetworkSizeEstimation() << " nodes" << std::endl;
+        unsigned long tot_nodes = 8 * std::exp2(table_depth);
+        ss << "Network size estimation: " << tot_nodes << " nodes" << std::endl;
    }
    return ss.str();
 }
@@ -80,7 +81,8 @@ NodeStats::toJson() const
    val["incoming"] = static_cast<Json::LargestUInt>(incoming_nodes);
    if (table_depth > 1) {
        val["table_depth"] = static_cast<Json::LargestUInt>(table_depth);
-        val["network_size_estimation"] = static_cast<Json::LargestUInt>(getNetworkSizeEstimation());
+        unsigned long tot_nodes = 8 * std::exp2(table_depth);
+        val["network_size_estimation"] = static_cast<Json::LargestUInt>(tot_nodes);
    }
    return val;
 }
@@ -261,11 +261,11 @@ void Dht::paginate(std::weak_ptr<Search> ws, Sp<Query> query, SearchNode* n) {
    auto select_q = std::make_shared<Query>(Select {}.field(Value::Field::Id), query ? query->where : Where {});
    auto onSelectDone = [this,ws,query](const net::Request& status,
                                        net::RequestAnswer&& answer) mutable {
-        // Retrieve search
+        // retreive search
        auto sr = ws.lock();
        if (not sr) return;
        const auto& id = sr->id;
-        // Retrieve search node
+        // retreive search node
        auto sn = sr->getNode(status.node);
        if (not sn) return;
        // backward compatibility
@@ -381,135 +381,104 @@ void Dht::searchSendAnnounceValue(const Sp<Search>& sr) {
    if (sr->announce.empty())
        return;
    unsigned i = 0;
+    auto probe_query = std::make_shared<Query>(Select {}.field(Value::Field::Id).field(Value::Field::SeqNum));
    std::weak_ptr<Search> ws = sr;
-
-    auto onDone = [this,ws](const net::Request& req, net::RequestAnswer&& answer)
-    { /* when put done */
-        if (auto sr = ws.lock()) {
-            onAnnounceDone(req.node, answer, sr);
-            searchStep(sr);
-        }
-    };
-
-    auto onExpired = [this,ws](const net::Request&, bool over)
-    { /* when put expired */
-        if (over)
-            if (auto sr = ws.lock())
-                scheduler.edit(sr->nextSearchStep, scheduler.time());
-    };
-
-    auto onSelectDone =
-    [this,ws,onDone,onExpired](const net::Request& req, net::RequestAnswer&& answer) mutable
-    { /* on probing done */
-        auto sr = ws.lock();
-        if (not sr) return;
-        const auto& now = scheduler.time();
-        sr->insertNode(req.node, scheduler.time(), answer.ntoken);
-        auto sn = sr->getNode(req.node);
-        if (not sn) return;
-
-        if (not sn->isSynced(now)) {
-            /* Search is now unsynced. Let's call searchStep to sync again. */
-            scheduler.edit(sr->nextSearchStep, now);
-            return;
-        }
-        for (auto& a : sr->announce) {
-            if (sn->getAnnounceTime(a.value->id) > now)
-                continue;
-            bool hasValue {false};
-            uint16_t seq_no = 0;
-            try {
-                const auto& f = std::find_if(answer.fields.cbegin(), answer.fields.cend(),
-                        [&a](const Sp<FieldValueIndex>& i){
-                            return i->index.at(Value::Field::Id).getInt() == a.value->id;
-                        });
-                if (f != answer.fields.cend() and *f) {
-                    hasValue = true;
-                    seq_no = static_cast<uint16_t>((*f)->index.at(Value::Field::SeqNum).getInt());
-                }
-            } catch (std::out_of_range&) { }
-
-            auto next_refresh_time = now + getType(a.value->type).expiration;
-            /* only put the value if the node doesn't already have it */
-            if (not hasValue or seq_no < a.value->seq) {
-                DHT_LOG.w(sr->id, sn->node->id, "[search %s] [node %s] sending 'put' (vid: %d)",
-                        sr->id.toString().c_str(), sn->node->toString().c_str(), a.value->id);
-                sn->acked[a.value->id] = std::make_pair(network_engine.sendAnnounceValue(sn->node,
-                                                sr->id,
-                                                a.value,
-                                                a.permanent ? time_point::max() : a.created,
-                                                sn->token,
-                                                onDone,
-                                                onExpired), next_refresh_time);
-            } else if (hasValue and a.permanent) {
-                DHT_LOG.w(sr->id, sn->node->id, "[search %s] [node %s] sending 'refresh' (vid: %d)",
-                        sr->id.toString().c_str(), sn->node->toString().c_str(), a.value->id);
-                sn->acked[a.value->id] = std::make_pair(network_engine.sendRefreshValue(sn->node,
-                                                sr->id,
-                                                a.value->id,
-                                                sn->token,
-                                                onDone,
-                                                onExpired), next_refresh_time);
-            } else {
-                DHT_LOG.w(sr->id, sn->node->id, "[search %s] [node %s] already has value (vid: %d). Aborting.",
-                        sr->id.toString().c_str(), sn->node->toString().c_str(), a.value->id);
-                auto ack_req = std::make_shared<net::Request>(net::Request::State::COMPLETED);
-                ack_req->reply_time = now;
-                sn->acked[a.value->id] = std::make_pair(std::move(ack_req), next_refresh_time);
-
-                /* step to clear announces */
-                scheduler.edit(sr->nextSearchStep, now);
-            }
-            if (a.permanent) {
-                scheduler.add(next_refresh_time - REANNOUNCE_MARGIN, [this,ws] {
-                    if (auto sr = ws.lock()) {
-                        searchStep(sr);
-                    }
-                });
-            }
-        }
-    };
-
-    Sp<Query> probe_query {};
    const auto& now = scheduler.time();
    for (auto& n : sr->nodes) {
        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()) {
+        if (std::find_if(sr->announce.cbegin(), sr->announce.cend(),
+            [&now,&n](const Announce& a) {
+                return n.getAnnounceTime(a.value->id) <= now;
+            }) == sr->announce.cend())
            continue;
-        }

-        bool sendQuery = false;
-        for (auto& a : sr->announce) {
-            if (n.getAnnounceTime(a.value->id) <= now) {
-                if (a.permanent) {
-                    sendQuery = true;
-                } else {
-                    DHT_LOG.w(sr->id, n.node->id, "[search %s] [node %s] sending 'put' (vid: %d)",
-                            sr->id.toString().c_str(), n.node->toString().c_str(), a.value->id);
-                    n.acked[a.value->id] = {
-                        network_engine.sendAnnounceValue(n.node, sr->id, a.value, a.created, n.token, onDone, onExpired),
-                        now + getType(a.value->type).expiration
-                    };
-                }
+        auto onDone = [this,ws](const net::Request& req, net::RequestAnswer&& answer)
+        { /* when put done */
+            if (auto sr = ws.lock()) {
+                onAnnounceDone(req.node, answer, sr);
+                searchStep(sr);
            }
-        }
+        };
+        auto onExpired = [this,ws](const net::Request&, bool over)
+        { /* when put expired */
+            if (over)
+                if (auto sr = ws.lock())
+                    scheduler.edit(sr->nextSearchStep, scheduler.time());
+        };
+        auto onSelectDone =
+            [this,ws,onDone,onExpired](const net::Request& req, net::RequestAnswer&& answer) mutable
+            { /* on probing done */
+                auto sr = ws.lock();
+                if (not sr) return;
+                const auto& now = scheduler.time();
+                sr->insertNode(req.node, scheduler.time(), answer.ntoken);
+                auto sn = sr->getNode(req.node);
+                if (not sn) return;

-        if (sendQuery) {
-            if (not probe_query)
-                probe_query = std::make_shared<Query>(Select {}.field(Value::Field::Id).field(Value::Field::SeqNum));
-            DHT_LOG.w(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,
-                    *probe_query,
-                    -1,
-                    onSelectDone,
-                    std::bind(&Dht::searchNodeGetExpired, this, _1, _2, ws, probe_query));
-        }
+                if (not sn->isSynced(now)) {
+                    /* Search is now unsynced. Let's call searchStep to sync again. */
+                    scheduler.edit(sr->nextSearchStep, now);
+                    return;
+                }
+                for (auto& a : sr->announce) {
+                    if (sn->getAnnounceTime(a.value->id) > now)
+                        continue;
+                    bool hasValue {false};
+                    uint16_t seq_no = 0;
+                    try {
+                        const auto& f = std::find_if(answer.fields.cbegin(), answer.fields.cend(),
+                                [&a](const Sp<FieldValueIndex>& i){
+                                    return i->index.at(Value::Field::Id).getInt() == a.value->id;
+                                });
+                        if (f != answer.fields.cend() and *f) {
+                            hasValue = true;
+                            seq_no = static_cast<uint16_t>((*f)->index.at(Value::Field::SeqNum).getInt());
+                        }
+                    } catch (std::out_of_range&) { }
+
+                    auto next_refresh_time = now + getType(a.value->type).expiration;
+                    /* only put the value if the node doesn't already have it */
+                    if (not hasValue or seq_no < a.value->seq) {
+                        DHT_LOG.w(sr->id, sn->node->id, "[search %s] [node %s] sending 'put' (vid: %d)",
+                                sr->id.toString().c_str(), sn->node->toString().c_str(), a.value->id);
+                        sn->acked[a.value->id] = std::make_pair(network_engine.sendAnnounceValue(sn->node,
+                                                        sr->id,
+                                                        a.value,
+                                                        a.permanent ? time_point::max() : a.created,
+                                                        sn->token,
+                                                        onDone,
+                                                        onExpired), next_refresh_time);
+                    } else if (hasValue and a.permanent) {
+                        DHT_LOG.w(sr->id, sn->node->id, "[search %s] [node %s] sending 'refresh' (vid: %d)",
+                                sr->id.toString().c_str(), sn->node->toString().c_str(), a.value->id);
+                        sn->acked[a.value->id] = std::make_pair(network_engine.sendRefreshValue(sn->node,
+                                                        sr->id,
+                                                        a.value->id,
+                                                        sn->token,
+                                                        onDone,
+                                                        onExpired), next_refresh_time);
+                    } else {
+                        DHT_LOG.w(sr->id, sn->node->id, "[search %s] [node %s] already has value (vid: %d). Aborting.",
+                                sr->id.toString().c_str(), sn->node->toString().c_str(), a.value->id);
+                        auto ack_req = std::make_shared<net::Request>(net::Request::State::COMPLETED);
+                        ack_req->reply_time = now;
+                        sn->acked[a.value->id] = std::make_pair(std::move(ack_req), next_refresh_time);
+
+                        /* step to clear announces */
+                        scheduler.edit(sr->nextSearchStep, now);
+                    }
+                }
+            };
+        DHT_LOG.w(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,
+                *probe_query,
+                -1,
+                onSelectDone,
+                std::bind(&Dht::searchNodeGetExpired, this, _1, _2, ws, probe_query));
        if (not n.candidate and ++i == TARGET_NODES)
            break;
    }
@@ -635,7 +604,7 @@ Dht::searchStep(Sp<Search> sr)
        // true if this node is part of the target nodes cluter.
        /*bool in = sr->id.xorCmp(myid, sr->nodes.back().node->id) < 0;

-        DHT_LOG_DBG("[search %s IPv%c] synced%s",
+        DHT_LOG_DEBUG("[search %s IPv%c] synced%s",
                sr->id.toString().c_str(), sr->af == AF_INET ? '4' : '6', in ? ", in" : "");*/

        if (not sr->listeners.empty()) {
@@ -665,7 +634,7 @@ Dht::searchStep(Sp<Search> sr)
                i++;
        }
        while (sent and sr->currentlySolicitedNodeCount() < MAX_REQUESTED_SEARCH_NODES);
-        /*DHT_LOG_DBG("[search %s IPv%c] step: sent %u requests (total %u).",
+        /*DHT_LOG_DEBUG("[search %s IPv%c] step: sent %u requests (total %u).",
            sr->id.toString().c_str(), sr->af == AF_INET ? '4' : '6', i, sr->currentlySolicitedNodeCount());*/
    }

@@ -710,7 +679,7 @@ unsigned Dht::refill(Dht::Search& sr) {

 /* Start a search. */
 Sp<Dht::Search>
-Dht::search(const InfoHash& id, sa_family_t af, GetCallback gcb, QueryCallback qcb, DoneCallback dcb, Value::Filter f, const Sp<Query>& q)
+Dht::search(const InfoHash& id, sa_family_t af, GetCallback gcb, QueryCallback qcb, DoneCallback dcb, Value::Filter f, Query q)
 {
    if (!isRunning(af)) {
        DHT_LOG.e(id, "[search %s IPv%c] unsupported protocol", id.toString().c_str(), (af == AF_INET) ? '4' : '6');
@@ -752,14 +721,26 @@ Dht::search(const InfoHash& id, sa_family_t af, GetCallback gcb, QueryCallback q
        sr->expired = false;
        sr->nodes.clear();
        sr->nodes.reserve(SEARCH_NODES+1);
-        sr->nextSearchStep = scheduler.add(time_point::max(), std::bind(&Dht::searchStep, this, sr));
        DHT_LOG.w(id, "[search %s IPv%c] new search", id.toString().c_str(), (af == AF_INET) ? '4' : '6');
        if (search_id == 0)
            search_id++;
    }

-    sr->get(f, q, qcb, gcb, dcb, scheduler);
+    if (gcb or qcb) {
+        auto now = scheduler.time();
+        sr->callbacks.insert(std::make_pair<time_point, Get>(
+            std::move(now),
+            Get { scheduler.time(), f, std::make_shared<Query>(q),
+                qcb ? qcb : QueryCallback {}, gcb ? gcb : GetCallback {}, dcb
+            }
+        ));
+    }
+
    refill(*sr);
+    if (sr->nextSearchStep)
+        scheduler.edit(sr->nextSearchStep, scheduler.time());
+    else
+        sr->nextSearchStep = scheduler.add(scheduler.time(), std::bind(&Dht::searchStep, this, sr));

    return sr;
 }
@@ -772,6 +753,15 @@ Dht::announce(const InfoHash& id,
        time_point created,
        bool permanent)
 {
+    const auto& now = scheduler.time();
+    if (!value) {
+        if (callback)
+            callback(false, {});
+        return;
+    }
+    created = std::min(now, created);
+    storageStore(id, value, created);
+
    auto& srs = searches(af);
    auto srp = srs.find(id);
    auto sr = srp == srs.end() ? search(id, af) : srp->second;
@@ -923,20 +913,13 @@ struct GetStatus : public OpStatus {
 void
 Dht::put(const InfoHash& id, Sp<Value> val, DoneCallback callback, time_point created, bool permanent)
 {
-    if (not val) {
-        if (callback)
-            callback(false, {});
-        return;
-    }
+    scheduler.syncTime();
+
    if (val->id == Value::INVALID_ID) {
        crypto::random_device rdev;
        std::uniform_int_distribution<Value::Id> rand_id {};
        val->id = rand_id(rdev);
    }
-    scheduler.syncTime();
-    const auto& now = scheduler.time();
-    created = std::min(now, created);
-    storageStore(id, val, created, {}, permanent);

    DHT_LOG.d(id, "put: adding %s -> %s", id.toString().c_str(), val->toString().c_str());

@@ -999,10 +982,10 @@ Dht::get(const InfoHash& id, GetCallback getcb, DoneCallback donecb, Value::Filt
 {
    scheduler.syncTime();

-    auto q = std::make_shared<Query>(Select {}, std::move(where));
+    Query q {{}, where};
    auto op = std::make_shared<GetStatus<Value>>();
-    auto f = filter.chain(q->where.getFilter());

+    auto f = filter.chain(q.where.getFilter());
    auto add_values = [op,f](const std::vector<Sp<Value>>& values) {
        std::vector<Sp<Value>> newvals {};
        for (const auto& v : values) {
@@ -1068,17 +1051,16 @@ void Dht::query(const InfoHash& id, QueryCallback cb, DoneCallback done_cb, Quer
    /* Try to answer this search locally. */
    qcb(local_fields);

-    auto sq = std::make_shared<Query>(std::move(q));
    Dht::search(id, AF_INET, {}, qcb, [=](bool ok, const std::vector<Sp<Node>>& nodes) {
        //DHT_LOG_WARN("DHT done IPv4");
        op->status4 = {true, ok};
        doneCallbackWrapper(done_cb, nodes, *op);
-    }, f, sq);
+    }, f, q);
    Dht::search(id, AF_INET6, {}, qcb, [=](bool ok, const std::vector<Sp<Node>>& nodes) {
        //DHT_LOG_WARN("DHT done IPv6");
        op->status6 = {true, ok};
        doneCallbackWrapper(done_cb, nodes, *op);
-    }, f, sq);
+    }, f, q);
 }

 std::vector<Sp<Value>>
@@ -1141,8 +1123,6 @@ bool
 Dht::cancelPut(const InfoHash& id, const Value::Id& vid)
 {
    bool canceled {false};
-    if (storageErase(id, vid))
-        canceled = true;
    auto sr_cancel_put = [&](std::map<InfoHash, Sp<Search>> srs) {
        auto srp = srs.find(id);
        if (srp == srs.end())
@@ -1211,11 +1191,11 @@ Dht::storageChanged(const InfoHash& id, Storage& st, ValueStorage& v, bool newVa
 }

 bool
-Dht::storageStore(const InfoHash& id, const Sp<Value>& value, time_point created, const SockAddr& sa, bool permanent)
+Dht::storageStore(const InfoHash& id, const Sp<Value>& value, time_point created, const SockAddr& sa)
 {
    const auto& now = scheduler.time();
    created = std::min(created, now);
-    auto expiration = permanent ? time_point::max() : created + getType(value->type).expiration;
+    auto expiration = created + getType(value->type).expiration;
    if (expiration < now)
        return false;

@@ -1237,9 +1217,7 @@ Dht::storageStore(const InfoHash& id, const Sp<Value>& value, time_point created
    if (auto vs = store.first) {
        total_store_size += store.second.size_diff;
        total_values += store.second.values_diff;
-        if (not permanent) {
-            scheduler.add(expiration, std::bind(&Dht::expireStorage, this, id));
-        }
+        scheduler.add(expiration, std::bind(&Dht::expireStorage, this, id));
        if (total_store_size > max_store_size) {
            expireStore();
        }
@@ -1249,18 +1227,6 @@ Dht::storageStore(const InfoHash& id, const Sp<Value>& value, time_point created
    return std::get<0>(store);
 }

-bool
-Dht::storageErase(const InfoHash& id, Value::Id vid)
-{
-    auto st = store.find(id);
-    if (st == store.end())
-        return false;
-    auto ret = st->second.remove(id, vid);
-    total_store_size += ret.size_diff;
-    total_values += ret.values_diff;
-    return ret.values_diff;
-}
-
 void
 Dht::storageAddListener(const InfoHash& id, const Sp<Node>& node, size_t socket_id, Query&& query)
 {
@@ -1619,6 +1585,11 @@ 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)
@@ -1627,12 +1598,6 @@ 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();
 }

@@ -1776,7 +1741,7 @@ Dht::Dht(const int& s, const int& s6, Config config)
 bool
 Dht::neighbourhoodMaintenance(RoutingTable& list)
 {
-    //DHT_LOG_DBG("neighbourhoodMaintenance");
+    //DHT_LOG_DEBUG("neighbourhoodMaintenance");
    auto b = list.findBucket(myid);
    if (b == list.end())
        return false;
@@ -2344,8 +2309,7 @@ Dht::onAnnounce(Sp<Node> n,
        Sp<Value> vc = v;
        if (lv) {
            if (*lv == *vc) {
-                storageRefresh(hash, v->id);
-                DHT_LOG.d(hash, node.id, "[store %s] [node %s] refreshed value %s", hash.toString().c_str(), node.toString().c_str(), std::to_string(v->id).c_str());
+                DHT_LOG.d(hash, node.id, "[store %s] nothing to do for %s", hash.toString().c_str(), lv->toString().c_str());
            } else {
                const auto& type = getType(lv->type);
                if (type.editPolicy(hash, lv, vc, node.id, node.getAddr())) {
@@ -2377,11 +2341,14 @@ Dht::onRefresh(Sp<Node> node, const InfoHash& hash, const Blob& token, const Val
 {
    using namespace net;

+    const auto& now = scheduler.time();
    if (not tokenMatch(token, node->getAddr())) {
        DHT_LOG.w(hash, node->id, "[node %s] incorrect token %s for 'put'", node->toString().c_str(), hash.toString().c_str());
        throw DhtProtocolException {DhtProtocolException::UNAUTHORIZED, DhtProtocolException::PUT_WRONG_TOKEN};
    }
-    if (storageRefresh(hash, vid)) {
+
+    auto s = store.find(hash);
+    if (s != store.end() and s->second.refresh(now, vid)) {
        DHT_LOG.d(hash, node->id, "[store %s] [node %s] refreshed value %s", hash.toString().c_str(), node->toString().c_str(), std::to_string(vid).c_str());
    } else {
        DHT_LOG.d(hash, node->id, "[store %s] [node %s] got refresh for unknown value",
@@ -2391,20 +2358,6 @@ Dht::onRefresh(Sp<Node> node, const InfoHash& hash, const Blob& token, const Val
    return {};
 }

-bool
-Dht::storageRefresh(const InfoHash& id, Value::Id vid)
-{
-    const auto& now = scheduler.time();
-    auto s = store.find(id);
-    if (s != store.end()) {
-        auto expiration = s->second.refresh(now, vid, types);
-        if (expiration != time_point::max())
-            scheduler.add(expiration, std::bind(&Dht::expireStorage, this, id));
-        return true;
-    }
-    return false;
-}
-
 void
 Dht::onAnnounceDone(const Sp<Node>& node, net::RequestAnswer& answer, Sp<Search>& sr)
 {
@@ -254,8 +254,7 @@ DhtProxyClient::get(const InfoHash& key, GetCallback cb, DoneCallback donecb, Va
    o.thread = std::thread([=](){
        // Try to contact the proxy and set the status to connected when done.
        // will change the connectivity status
-        struct GetState{ std::atomic_bool ok {true}; std::atomic_bool stop {false}; };
-        auto state = std::make_shared<GetState>();
+        auto ok = std::make_shared<std::atomic_bool>(true);
        restbed::Http::async(req,
            [=](const std::shared_ptr<restbed::Request>& req,
                const std::shared_ptr<restbed::Response>& reply) {
@@ -263,9 +262,9 @@ DhtProxyClient::get(const InfoHash& key, GetCallback cb, DoneCallback donecb, Va

            if (code == 200) {
                try {
-                    while (restbed::Http::is_open(req) and not *finished and not state->stop) {
+                    while (restbed::Http::is_open(req) and not *finished) {
                        restbed::Http::fetch("\n", reply);
-                        if (*finished or state->stop)
+                        if (*finished)
                            break;
                        std::string body;
                        reply->get_body(body);
@@ -280,30 +279,29 @@ DhtProxyClient::get(const InfoHash& key, GetCallback cb, DoneCallback donecb, Va
                            auto value = std::make_shared<Value>(json);
                            if ((not filter or filter(*value)) and cb) {
                                std::lock_guard<std::mutex> lock(lockCallbacks);
-                                callbacks_.emplace_back([cb, value, state]() {
-                                    if (not state->stop and not cb({value}))
-                                        state->stop = true;
+                                callbacks_.emplace_back([cb, value, finished]() {
+                                    if (not *finished and not cb({value}))
+                                        *finished = true;
                                });
                                loopSignal_();
                            }
                        } else {
-                            state->ok = false;
+                            *ok = false;
                        }
                    }
                } catch (std::runtime_error& e) { }
            } else {
-                state->ok = false;
+                *ok = false;
            }
        }).wait();
        if (donecb) {
            std::lock_guard<std::mutex> lock(lockCallbacks);
            callbacks_.emplace_back([=](){
-                donecb(state->ok, {});
-                state->stop = true;
+                donecb(*ok, {});
            });
            loopSignal_();
        }
-        if (!state->ok) {
+        if (!ok) {
            // Connection failed, update connectivity
            opFailed();
        }
@@ -344,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);
                }
@@ -708,12 +706,11 @@ DhtProxyClient::sendListen(const std::shared_ptr<restbed::Request>& req, const V
                    Json::CharReaderBuilder rbuilder;
                    auto reader = std::unique_ptr<Json::CharReader>(rbuilder.newCharReader());
                    if (reader->parse(body.data(), body.data() + body.size(), &json, &err)) {
-                        auto expired = json.get("expired", Json::Value(false)).asBool();
                        auto value = std::make_shared<Value>(json);
                        if ((not filter or filter(*value)) and cb)  {
                            std::lock_guard<std::mutex> lock(lockCallbacks);
-                            callbacks_.emplace_back([cb, value, state, expired]() {
-                                if (not state->cancel and not cb({value}, expired))
+                            callbacks_.emplace_back([cb, value, state]() {
+                                if (not state->cancel and not cb({value}, false))
                                    state->cancel = true;
                            });
                            loopSignal_();
@@ -954,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);
            }
@@ -70,7 +70,6 @@ DhtProxyServer::DhtProxyServer(std::shared_ptr<DhtRunner> dht, in_port_t port ,
        auto resource = std::make_shared<restbed::Resource>();
        resource->set_path("/");
        resource->set_method_handler("GET", std::bind(&DhtProxyServer::getNodeInfo, this, _1));
-        resource->set_method_handler("STATS", std::bind(&DhtProxyServer::getStats, this, _1));
        service_->publish(resource);
        resource = std::make_shared<restbed::Resource>();
        resource->set_path("/{hash: .*}");
@@ -138,7 +137,7 @@ DhtProxyServer::DhtProxyServer(std::shared_ptr<DhtRunner> dht, in_port_t port ,
    printStatsJob_ = scheduler_.add(scheduler_.time() + PRINT_STATS_PERIOD, [this] {
        if (stopListeners) return;
        if (service_->is_up())
-            updateStats();
+            std::cout << getStats().toString() << std::endl;
        // Refresh stats cache
        auto newInfo = dht_->getNodeInfo();
        {
@@ -179,30 +178,31 @@ DhtProxyServer::stop()
        server_thread.join();
 }

-void
-DhtProxyServer::updateStats() const
+DhtProxyServer::ServerStats
+DhtProxyServer::getStats() const
 {
+    ServerStats ret {};
    auto now = clock::now();
    auto last = lastStatsReset_.exchange(now);
    auto count = requestNum_.exchange(0);
    auto dt = std::chrono::duration<double>(now - last);
-    stats_.requestRate = count / dt.count();
+    ret.requestRate = count / dt.count();
 #if OPENDHT_PUSH_NOTIFICATIONS
-    stats_.pushListenersCount = pushListeners_.size();
+    ret.pushListenersCount = pushListeners_.size();
 #endif
-    stats_.putCount = puts_.size();
-    stats_.listenCount = currentListeners_.size();
-    stats_.nodeInfo = nodeInfo_;
+    ret.putCount = puts_.size();
+    ret.listenCount = currentListeners_.size();
+    return ret;
 }

 void
-DhtProxyServer::getNodeInfo(const Sp<restbed::Session>& session) const
+DhtProxyServer::getNodeInfo(const std::shared_ptr<restbed::Session>& session) const
 {
    requestNum_++;
    const auto request = session->get_request();
    int content_length = std::stoi(request->get_header("Content-Length", "0"));
    session->fetch(content_length,
-        [this](const Sp<restbed::Session>& s, const restbed::Bytes& /*b*/) mutable
+        [this](const std::shared_ptr<restbed::Session> s, const restbed::Bytes& /*b*/) mutable
        {
            try {
                if (dht_) {
@@ -235,44 +235,14 @@ DhtProxyServer::getNodeInfo(const Sp<restbed::Session>& session) const
 }

 void
-DhtProxyServer::getStats(const Sp<restbed::Session>& session) const
-{
-    requestNum_++;
-    const auto request = session->get_request();
-    int content_length = std::stoi(request->get_header("Content-Length", "0"));
-    session->fetch(content_length,
-        [this](const Sp<restbed::Session>& s, const restbed::Bytes& /*b*/) mutable
-        {
-            try {
-                if (dht_) {
-#ifdef OPENDHT_JSONCPP
-                    Json::StreamWriterBuilder wbuilder;
-                    wbuilder["commentStyle"] = "None";
-                    wbuilder["indentation"] = "";
-                    auto output = Json::writeString(wbuilder, stats_.toJson()) + "\n";
-                    s->close(restbed::OK, output);
-#else
-                    s->close(restbed::NotFound, "{\"err\":\"JSON not enabled on this instance\"}");
-#endif
-                }
-                else
-                    s->close(restbed::SERVICE_UNAVAILABLE, "{\"err\":\"Incorrect DhtRunner\"}");
-            } catch (...) {
-                s->close(restbed::INTERNAL_SERVER_ERROR, "{\"err\":\"Internal server error\"}");
-            }
-        }
-    );
-}
-
-void
-DhtProxyServer::get(const Sp<restbed::Session>& session) const
+DhtProxyServer::get(const std::shared_ptr<restbed::Session>& session) const
 {
    requestNum_++;
    const auto request = session->get_request();
    int content_length = std::stoi(request->get_header("Content-Length", "0"));
    auto hash = request->get_path_parameter("hash");
    session->fetch(content_length,
-        [=](const Sp<restbed::Session>& s, const restbed::Bytes& /*b* */)
+        [=](const std::shared_ptr<restbed::Session> s, const restbed::Bytes& /*b* */)
        {
            try {
                if (dht_) {
@@ -280,21 +250,23 @@ DhtProxyServer::get(const Sp<restbed::Session>& session) const
                    if (!infoHash) {
                        infoHash = InfoHash::get(hash);
                    }
-                    s->yield(restbed::OK, "", [=](const Sp<restbed::Session>&) {});
-                    dht_->get(infoHash, [s](const Sp<Value>& value) {
-                        if (s->is_closed()) return false;
-                        // Send values as soon as we get them
-                        Json::StreamWriterBuilder wbuilder;
-                        wbuilder["commentStyle"] = "None";
-                        wbuilder["indentation"] = "";
-                        auto output = Json::writeString(wbuilder, value->toJson()) + "\n";
-                        s->yield(output, [](const Sp<restbed::Session>& /*session*/){ });
-                        return true;
-                    }, [s](bool /*ok* */) {
-                        // Communication is finished
-                        if (not s->is_closed()) {
+                    s->yield(restbed::OK, "", [=]( const std::shared_ptr< restbed::Session > s) {
+                        auto cacheSession = std::weak_ptr<restbed::Session>(s);
+
+                        dht_->get(infoHash, [cacheSession](std::shared_ptr<Value> value) {
+                            auto s = cacheSession.lock();
+                            if (!s) return false;
+                            // Send values as soon as we get them
+                            Json::StreamWriterBuilder wbuilder;
+                            wbuilder["commentStyle"] = "None";
+                            wbuilder["indentation"] = "";
+                            auto output = Json::writeString(wbuilder, value->toJson()) + "\n";
+                            s->yield(output, [](const std::shared_ptr<restbed::Session> /*session*/){ });
+                            return true;
+                        }, [s](bool /*ok* */) {
+                            // Communication is finished
                            s->close();
-                        }
+                        });
                    });
                } else {
                    s->close(restbed::SERVICE_UNAVAILABLE, "{\"err\":\"Incorrect DhtRunner\"}");
@@ -307,7 +279,7 @@ DhtProxyServer::get(const Sp<restbed::Session>& session) const
 }

 void
-DhtProxyServer::listen(const Sp<restbed::Session>& session)
+DhtProxyServer::listen(const std::shared_ptr<restbed::Session>& session)
 {
    requestNum_++;
    const auto request = session->get_request();
@@ -317,7 +289,7 @@ DhtProxyServer::listen(const Sp<restbed::Session>& session)
    if (!infoHash)
        infoHash = InfoHash::get(hash);
    session->fetch(content_length,
-        [=](const Sp<restbed::Session>& s, const restbed::Bytes& /*b* */)
+        [=](const std::shared_ptr<restbed::Session> s, const restbed::Bytes& /*b* */)
        {
            try {
                if (dht_) {
@@ -335,7 +307,7 @@ DhtProxyServer::listen(const Sp<restbed::Session>& session)
                    // cache the session to avoid an incrementation of the shared_ptr's counter
                    // else, the session->close() will not close the socket.
                    auto cacheSession = std::weak_ptr<restbed::Session>(s);
-                    listener.token = dht_->listen(infoHash, [cacheSession](const std::vector<Sp<Value>>& values, bool expired) {
+                    listener.token = dht_->listen(infoHash, [cacheSession](std::shared_ptr<Value> value) {
                        auto s = cacheSession.lock();
                        if (!s) return false;
                        // Send values as soon as we get them
@@ -343,13 +315,8 @@ DhtProxyServer::listen(const Sp<restbed::Session>& session)
                            Json::StreamWriterBuilder wbuilder;
                            wbuilder["commentStyle"] = "None";
                            wbuilder["indentation"] = "";
-                            for (const auto& value : values) {
-                                auto val = value->toJson();
-                                if (expired)
-                                    val["expired"] = true;
-                                auto output = Json::writeString(wbuilder, val) + "\n";
-                                s->yield(output, [](const Sp<restbed::Session>&){ });
-                            }
+                            auto output = Json::writeString(wbuilder, value->toJson()) + "\n";
+                            s->yield(output, [](const std::shared_ptr<restbed::Session>){ });
                        }
                        return !s->is_closed();
                    });
@@ -485,6 +452,7 @@ DhtProxyServer::subscribe(const std::shared_ptr<restbed::Session>& session)
 void
 DhtProxyServer::unsubscribe(const std::shared_ptr<restbed::Session>& session)
 {
+
    requestNum_++;
    const auto request = session->get_request();
    int content_length = std::stoi(request->get_header("Content-Length", "0"));
@@ -37,7 +37,6 @@
 #include <winsock2.h>
 #include <ws2tcpip.h>
 #define close(x) closesocket(x)
-#define write(...) _write(__VA_ARGS__)
 #endif

 namespace dht {
@@ -165,7 +164,7 @@ DhtRunner::shutdown(ShutdownCallback cb) {
 void
 DhtRunner::join()
 {
-    stopNetwork();
+    running_network = false;
    running = false;
    cv.notify_all();
    bootstrap_cv.notify_all();
@@ -411,15 +410,25 @@ 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 delay = clock::now() - pck.received;
-            if (delay > std::chrono::milliseconds(500))
+            auto now = clock::now();
+            auto delay = now - pck.received;
+            if (delay > std::chrono::seconds(1))
                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);
    }
@@ -459,10 +468,8 @@ int bindSocket(const SockAddr& addr, SockAddr& bound)
    if (is_ipv6)
        setsockopt(sock, IPPROTO_IPV6, IPV6_V6ONLY, (const char*)&set, sizeof(set));
    int rc = bind(sock, addr.get(), addr.getLength());
-    if(rc < 0) {
-        close(sock);
+    if(rc < 0)
        throw DhtException("Can't bind socket on " + addr.toString() + " " + strerror(rc));
-    }
    sockaddr_storage ss;
    socklen_t ss_len = sizeof(ss);
    getsockname(sock, (sockaddr*)&ss, &ss_len);
@@ -470,64 +477,12 @@ int bindSocket(const SockAddr& addr, SockAddr& bound)
    return sock;
 }

-#ifdef _WIN32
-inline void udpPipe(int fds[2])
-{
-    int lst = socket(AF_INET, SOCK_DGRAM, 0);
-    if (lst < 0)
-        throw DhtException(std::string("Can't open socket: ") + strerror(lst));
-    sockaddr_in inaddr;
-    sockaddr addr;
-    memset(&inaddr, 0, sizeof(inaddr));
-    memset(&addr, 0, sizeof(addr));
-    inaddr.sin_family = AF_INET;
-    inaddr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
-    inaddr.sin_port = 0;
-    int yes=1;
-    setsockopt(lst, SOL_SOCKET, SO_REUSEADDR, (char*)&yes, sizeof(yes));
-    int rc = bind(lst, (sockaddr*)&inaddr, sizeof(inaddr));
-    if (rc < 0) {
-        close(lst);
-        throw DhtException("Can't bind socket on " + print_addr((sockaddr*)&addr, sizeof(inaddr)) + " " + strerror(rc));
-    }
-    socklen_t len = sizeof(inaddr);
-    getsockname(lst, &addr, &len);
-    fds[0] = lst;
-    fds[1] = socket(AF_INET, SOCK_DGRAM, 0);
-    connect(fds[1], &addr, len);
-}
-#endif
-
-void
-DhtRunner::stopNetwork()
-{
-    running_network = false;
-    if (stop_writefd != -1) {
-        if (write(stop_writefd, "\0", 1) == -1) {
-            perror("write");
-        }
-    }
-}
-
 void
 DhtRunner::startNetwork(const SockAddr sin4, const SockAddr sin6)
 {
-    stopNetwork();
+    running_network = false;
    if (rcv_thread.joinable())
        rcv_thread.join();
-
-    int stopfds[2];
-#ifndef _WIN32
-    auto status = pipe(stopfds);
-    if (status == -1) {
-        throw DhtException("Can't open pipe");
-    }
-#else
-    udpPipe(stopfds);
-#endif
-    int stop_readfd = stopfds[0];
-    stop_writefd = stopfds[1];
-
    s4 = -1;
    s6 = -1;

@@ -542,19 +497,19 @@ DhtRunner::startNetwork(const SockAddr sin4, const SockAddr sin6)
 #endif

    running_network = true;
-    rcv_thread = std::thread([this, stop_readfd]() {
+    rcv_thread = std::thread([this]() {
        try {
            while (running_network) {
+                struct timeval tv {/*.tv_sec = */0, /*.tv_usec = */250000};
                fd_set readfds;

                FD_ZERO(&readfds);
-                FD_SET(stop_readfd, &readfds);
                if(s4 >= 0)
                    FD_SET(s4, &readfds);
                if(s6 >= 0)
                    FD_SET(s6, &readfds);

-                int rc = select(s4 > s6 ? s4 + 1 : s6 + 1, &readfds, nullptr, nullptr, nullptr);
+                int rc = select(s4 > s6 ? s4 + 1 : s6 + 1, &readfds, nullptr, nullptr, &tv);
                if(rc < 0) {
                    if(errno != EINTR) {
                        perror("select");
@@ -575,13 +530,13 @@ DhtRunner::startNetwork(const SockAddr sin4, const SockAddr sin6)
                    else if(s6 >= 0 && FD_ISSET(s6, &readfds))
                        rc = recvfrom(s6, (char*)buf.data(), buf.size(), 0, (sockaddr*)&from, &from_len);
                    else
-                        continue;
+                        break;
                    if (rc > 0) {
                        {
                            std::lock_guard<std::mutex> lck(sock_mtx);
                            if (rcv.size() >= RX_QUEUE_MAX_SIZE) {
                                std::cerr << "Dropping packet: queue is full!" << std::endl;
-                                rcv.pop();
+                                continue;
                            }
                            rcv.emplace(ReceivedPacket {Blob {buf.begin(), buf.begin()+rc+1}, SockAddr(from, from_len), clock::now()});
                        }
@@ -600,11 +555,6 @@ DhtRunner::startNetwork(const SockAddr sin4, const SockAddr sin6)
        s6 = -1;
        bound4 = {};
        bound6 = {};
-        if (stop_readfd != -1)
-            close(stop_readfd);
-        if (stop_writefd != -1)
-            close(stop_writefd);
-        stop_writefd = -1;
    });
 }

@@ -643,7 +593,6 @@ 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;
@@ -651,20 +600,21 @@ 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();
@@ -958,8 +908,8 @@ DhtRunner::findCertificate(InfoHash hash, std::function<void(const std::shared_p
 void
 DhtRunner::resetDht()
 {
-#if OPENDHT_PROXY_CLIENT
    listeners_.clear();
+#if OPENDHT_PROXY_CLIENT
    dht_via_proxy_.reset();
 #endif // OPENDHT_PROXY_CLIENT
    dht_.reset();
@@ -185,7 +185,7 @@ NetworkEngine::tellListener(Sp<Node> node, Tid socket_id, const InfoHash& hash,
 }

 void
-NetworkEngine::tellListenerRefreshed(Sp<Node> n, Tid socket_id, const InfoHash&, const Blob& token, const std::vector<Value::Id>& values)
+NetworkEngine::tellListenerRefreshed(Sp<Node> n, Tid socket_id, const InfoHash& hash, const Blob& token, const std::vector<Value::Id>& values)
 {
    msgpack::sbuffer buffer;
    msgpack::packer<msgpack::sbuffer> pk(&buffer);
@@ -215,7 +215,7 @@ NetworkEngine::tellListenerRefreshed(Sp<Node> n, Tid socket_id, const InfoHash&,
 }

 void
-NetworkEngine::tellListenerExpired(Sp<Node> n, Tid socket_id, const InfoHash&, const Blob& token, const std::vector<Value::Id>& values)
+NetworkEngine::tellListenerExpired(Sp<Node> n, Tid socket_id, const InfoHash& hash, const Blob& token, const std::vector<Value::Id>& values)
 {
    msgpack::sbuffer buffer;
    msgpack::packer<msgpack::sbuffer> pk(&buffer);
@@ -403,6 +403,7 @@ 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;
@@ -419,7 +420,7 @@ NetworkEngine::processMessage(const uint8_t *buf, size_t buflen, const SockAddr&
        msg->msgpack_unpack(msg_res.get());
    } catch (const std::exception& e) {
        DHT_LOG.w("Can't parse message of size %lu: %s", buflen, e.what());
-        //DHT_LOG.DBG.logPrintable(buf, buflen);
+        DHT_LOG.DEBUG.logPrintable(buf, buflen);
        return;
    }

@@ -551,7 +552,7 @@ NetworkEngine::process(std::unique_ptr<ParsedMessage>&& msg, const SockAddr& fro
        }
        case MessageType::Reply:
            if (req) { /* request reply */
-                if (req->getType() == MessageType::AnnounceValue or req->getType() == MessageType::Listen)
+                if (msg->type == MessageType::AnnounceValue or msg->type == MessageType::Listen)
                    req->node->authSuccess();
                req->reply_time = scheduler.time();

@@ -663,6 +664,8 @@ 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));
@@ -977,7 +980,7 @@ NetworkEngine::sendNodesValues(const SockAddr& addr, Tid tid, const Blob& nodes,
            pk.pack(std::string("v")); pk.pack_array(st.size()*fields.size());
            for (const auto& v : st)
                v->msgpack_pack_fields(fields, pk);
-            //DHT_LOG_DBG("sending closest nodes (%d+%d nodes.), %u value headers containing %u fields",
+            //DHT_LOG_DEBUG("sending closest nodes (%d+%d nodes.), %u value headers containing %u fields",
            //        nodes.size(), nodes6.size(), st.size(), fields.size());
        }
    }
@@ -20,9 +20,6 @@

 namespace dht {

-constexpr size_t CLEANUP_MAX_NODES {1024};
-constexpr size_t CLEANUP_FREQ {1024};
-
 NodeCache::~NodeCache()
 {
    cache_4.setExpired();
@@ -44,33 +41,29 @@ NodeCache::getNode(const InfoHash& id, const SockAddr& addr, time_point now, boo
 std::vector<Sp<Node>>
 NodeCache::getCachedNodes(const InfoHash& id, sa_family_t sa_f, size_t count) const
 {
-    return cache(sa_f).getCachedNodes(id, count);
-}
+    const auto& c = cache(sa_f);

-std::vector<Sp<Node>>
-NodeCache::NodeMap::getCachedNodes(const InfoHash& id, size_t count) const
-{
    std::vector<Sp<Node>> nodes;
-    nodes.reserve(std::min(size(), count));
-    const_iterator it;
-    auto dec_it = [this](const_iterator& it) {
+    nodes.reserve(std::min(c.size(), count));
+    NodeMap::const_iterator it;
+    auto dec_it = [&c](NodeMap::const_iterator& it) {
        auto ret = it;
-        it = (it == cbegin()) ? cend() : std::prev(it);
+        it = (it == c.cbegin()) ? c.cend() : std::prev(it);
        return ret;
    };

-    auto it_p = lower_bound(id),
+    auto it_p = c.lower_bound(id),
        it_n = it_p;
-    if (not empty())
+    if (not c.empty())
        dec_it(it_p); /* Create 2 separate iterator if we could */

-    while (nodes.size() < count and (it_n != cend() or it_p != cend())) {
+    while (nodes.size() < count and (it_n != c.cend() or it_p != c.cend())) {
        /* If one of the iterator is at the end, then take the other one
           If they are both in middle of somewhere comapre both and take
           the closest to the id. */
-        if (it_p == cend())      it = it_n++;
-        else if (it_n == cend()) it = dec_it(it_p);
-        else                     it = id.xorCmp(it_p->first, it_n->first) < 0 ? dec_it(it_p) : it_n++;
+        if (it_p == c.cend())       it = it_n++;
+        else if (it_n == c.cend())  it = dec_it(it_p);
+        else                        it = id.xorCmp(it_p->first, it_n->first) < 0 ? dec_it(it_p) : it_n++;

        if (auto n = it->second.lock())
            if ( not n->isExpired() and not n->isClient() )
@@ -111,10 +104,6 @@ NodeCache::NodeMap::getNode(const InfoHash& id, const SockAddr& addr, time_point
    if (not node) {
        node = std::make_shared<Node>(id, addr, client);
        it.first->second = node;
-        if (cleanup_counter++ == CLEANUP_FREQ) {
-            cleanup();
-            cleanup_counter = 0;
-        }
    } else if (confirm or node->isOld(now)) {
        node->update(addr);
    }
@@ -131,7 +120,6 @@ NodeCache::NodeMap::clearBadNodes() {
            erase(it++);
        }
    }
-    cleanup_counter = 0;
 }

 void
@@ -140,21 +128,6 @@ NodeCache::NodeMap::setExpired() {
        if (auto n = wn.second.lock())
            n->setExpired();
    clear();
-    cleanup_counter = 0;
-}
-
-void
-NodeCache::NodeMap::cleanup()
-{
-    auto it = lower_bound(InfoHash::getRandom());
-    for (size_t n = 0, maxNodes = std::min(size(), CLEANUP_MAX_NODES); n != maxNodes; n++) {
-        if (it == end())
-            it = begin();
-        if (it->second.expired())
-            erase(it++);
-        else
-            ++it;
-    }
 }

 }
@@ -320,21 +320,22 @@ struct Dht::SearchNode {
    }

    /**
-     * Assuming the node is synced, should a "put" request be sent to this node now ?
+     * Assumng the node is synced, should a "put" request be sent to this node now ?
     */
    time_point getAnnounceTime(Value::Id vid) const {
        const auto& ack = acked.find(vid);
-        if (ack == acked.cend() or not ack->second.first) {
+        const auto& gs = probe_query ? getStatus.find(probe_query) : getStatus.cend();
+        if ((ack == acked.cend() or not ack->second.first) and (gs == getStatus.cend()
+                                                          or not gs->second or not gs->second->pending()))
            return time_point::min();
-        }
-        if (ack->second.first->completed()) {
-            return ack->second.second - REANNOUNCE_MARGIN;
-        }
-        return ack->second.first->pending() ? time_point::max() : time_point::min();
+        return ((gs != getStatus.cend() and gs->second and gs->second->pending())
+                or ack == acked.cend() or not ack->second.first or ack->second.first->pending())
+                ? time_point::max()
+                : (ack->second.first->completed() ? ack->second.second - REANNOUNCE_MARGIN : time_point::min());
    }

    /**
-     * Assuming the node is synced, should the "listen" request with Query q be
+     * Assumng the node is synced, should the "listen" request with Query q be
     * sent to this node now ?
     */
    time_point getListenTime(const Sp<Query>& q) const {
@@ -475,17 +476,6 @@ struct Dht::Search {
    bool isAnnounced(Value::Id id) const;
    bool isListening(time_point now) const;

-    void get(Value::Filter f, const Sp<Query>& q, const QueryCallback& qcb, const GetCallback& gcb, const DoneCallback& dcb, Scheduler& scheduler) {
-        if (gcb or qcb) {
-            const auto& now = scheduler.time();
-            callbacks.emplace(now, Get { now, f, q, qcb, gcb, dcb });
-            auto values = cache.get(f);
-            if (not values.empty())
-                gcb(values);
-            scheduler.edit(nextSearchStep, now);
-        }
-    }
-
    size_t listen(ValueCallback cb, Value::Filter f, const Sp<Query>& q, Scheduler& scheduler) {
        //DHT_LOG.e(id, "[search %s IPv%c] listen", id.toString().c_str(), (af == AF_INET) ? '4' : '6');
        return cache.listen(cb, q, f, [&](const Sp<Query>& q, ValueCallback vcb){
@@ -56,7 +56,7 @@ SecureDht::SecureDht(std::unique_ptr<DhtInterface> dht, SecureDht::Config conf)
            1
        }, [this](bool ok) {
            if (ok)
-                DHT_LOG.DBG("SecureDht: public key announced successfully");
+                DHT_LOG.DEBUG("SecureDht: public key announced successfully");
        }, {}, true);
    }
 }
@@ -68,19 +68,27 @@ ValueType
 SecureDht::secureType(ValueType&& type)
 {
    type.storePolicy = [this,type](InfoHash id, Sp<Value>& v, const InfoHash& nid, const SockAddr& a) {
-        if (v->isSigned())
-            return v->checkSignature();
+        if (v->isSigned()) {
+            if (!v->owner or !v->owner->checkSignature(v->getToSign(), v->signature)) {
+                DHT_LOG.WARN("Signature verification failed");
+                return false;
+            }
+            else
+                DHT_LOG.WARN("Signature verification succeeded");
+        }
        return type.storePolicy(id, v, nid, a);
    };
    type.editPolicy = [this,type](InfoHash id, const Sp<Value>& o, Sp<Value>& n, const InfoHash& nid, const SockAddr& a) {
-        if (not o->isSigned())
+        if (!o->isSigned())
            return type.editPolicy(id, o, n, nid, a);
-        if (o->owner != n->owner or not n->isSigned()) {
+        if (o->owner != n->owner) {
            DHT_LOG.WARN("Edition forbidden: owner changed.");
            return false;
        }
-        if (not n->checkSignature())
+        if (!o->owner or !o->owner->checkSignature(n->getToSign(), n->signature)) {
+            DHT_LOG.WARN("Edition forbidden: signature verification failed.");
            return false;
+        }
        if (o->seq == n->seq) {
            // If the data is exactly the same,
            // it can be reannounced, possibly by someone else.
@@ -131,7 +139,7 @@ SecureDht::registerCertificate(const InfoHash& node, const Blob& data)
    }
    InfoHash h = crt->getPublicKey().getId();
    if (node == h) {
-        DHT_LOG.DBG("Registering certificate for %s", h.toString().c_str());
+        DHT_LOG.DEBUG("Registering certificate for %s", h.toString().c_str());
        auto it = nodesCertificates_.find(h);
        if (it == nodesCertificates_.end())
            std::tie(it, std::ignore) = nodesCertificates_.emplace(h, std::move(crt));
@@ -139,7 +147,7 @@ SecureDht::registerCertificate(const InfoHash& node, const Blob& data)
            it->second = std::move(crt);
        return it->second;
    } else {
-        DHT_LOG.WARN("Certificate %s for node %s does not match node id !", h.toString().c_str(), node.toString().c_str());
+        DHT_LOG.DEBUG("Certificate %s for node %s does not match node id !", h.toString().c_str(), node.toString().c_str());
        return nullptr;
    }
 }
@@ -156,7 +164,7 @@ SecureDht::findCertificate(const InfoHash& node, std::function<void(const Sp<cry
 {
    Sp<crypto::Certificate> b = getCertificate(node);
    if (b && *b) {
-        DHT_LOG.DBG("Using certificate from cache for %s", node.toString().c_str());
+        DHT_LOG.DEBUG("Using certificate from cache for %s", node.toString().c_str());
        if (cb)
            cb(b);
        return;
@@ -164,7 +172,7 @@ SecureDht::findCertificate(const InfoHash& node, std::function<void(const Sp<cry
    if (localQueryMethod_) {
        auto res = localQueryMethod_(node);
        if (not res.empty()) {
-            DHT_LOG.DBG("Registering certificate from local store for %s", node.toString().c_str());
+            DHT_LOG.DEBUG("Registering certificate from local store for %s", node.toString().c_str());
            nodesCertificates_.emplace(node, res.front());
            if (cb)
                cb(res.front());
@@ -179,7 +187,7 @@ SecureDht::findCertificate(const InfoHash& node, std::function<void(const Sp<cry
        for (const auto& v : vals) {
            if (auto cert = registerCertificate(node, v->data)) {
                *found = true;
-                DHT_LOG.DBG("Found certificate for %s", node.toString().c_str());
+                DHT_LOG.DEBUG("Found certificate for %s", node.toString().c_str());
                if (cb)
                    cb(cert);
                return false;
@@ -197,7 +205,7 @@ SecureDht::findPublicKey(const InfoHash& node, std::function<void(const Sp<const
 {
    auto pk = getPublicKey(node);
    if (pk && *pk) {
-        DHT_LOG.DBG("Found public key from cache for %s", node.toString().c_str());
+        DHT_LOG.DEBUG("Found public key from cache for %s", node.toString().c_str());
        if (cb)
            cb(pk);
        return;
@@ -228,22 +236,24 @@ SecureDht::checkValue(const Sp<Value>& v)
            return {};
        }
        try {
-            if (auto decrypted_val = v->decrypt(*key_)) {
-                if (decrypted_val->owner)
-                    nodesPubKeys_[decrypted_val->owner->getId()] = decrypted_val->owner;
-                return decrypted_val;
+            Value decrypted_val (decrypt(*v));
+            if (decrypted_val.recipient == getId()) {
+                if (decrypted_val.owner)
+                    nodesPubKeys_[decrypted_val.owner->getId()] = decrypted_val.owner;
+                return std::make_shared<Value>(std::move(decrypted_val));
            }
+            // Ignore values belonging to other people
        } catch (const std::exception& e) {
            DHT_LOG.WARN("Could not decrypt value %s : %s", v->toString().c_str(), e.what());
        }
    }
    // Check signed values
    else if (v->isSigned()) {
-        if (v->checkSignature()) {
-            if (v->owner)
-                nodesPubKeys_[v->owner->getId()] = v->owner;
+        if (v->owner and v->owner->checkSignature(v->getToSign(), v->signature)) {
+            nodesPubKeys_[v->owner->getId()] = v->owner;
            return v;
-        } else
+        }
+        else
            DHT_LOG.WARN("Signature verification failed for %s", v->toString().c_str());
    }
    // Forward normal values
@@ -317,13 +327,14 @@ SecureDht::putSigned(const InfoHash& hash, Sp<Value> val, DoneCallback callback,
    // Check if we are already announcing a value
    auto p = dht_->getPut(hash, val->id);
    if (p && val->seq <= p->seq) {
+        DHT_LOG.DEBUG("Found previous value being announced.");
        val->seq = p->seq + 1;
    }

    // Check if data already exists on the dht
    get(hash,
        [val,this] (const std::vector<Sp<Value>>& vals) {
-            DHT_LOG.DBG("Found online previous value being announced.");
+            DHT_LOG.DEBUG("Found online previous value being announced.");
            for (const auto& v : vals) {
                if (!v->isSigned())
                    DHT_LOG.ERR("Existing non-signed value seems to exists at this location.");
@@ -345,7 +356,7 @@ 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)
 {
-    findPublicKey(to, [=](const Sp<const crypto::PublicKey>& pk) {
+    findPublicKey(to, [=](const Sp<const crypto::PublicKey> pk) {
        if(!pk || !*pk) {
            if (callback)
                callback(false, {});
@@ -154,16 +154,15 @@ struct Storage {
     *
     * @param now  The reference to now
     * @param vid  The value id
-     * @return time of the next expiration, time_point::max() if no expiration
+     * @return true if a value storage was updated, false otherwise
     */
-    time_point refresh(const time_point& now, const Value::Id& vid, const TypeStore& types) {
+    bool refresh(const time_point& now, const Value::Id& vid) {
        for (auto& vs : values)
            if (vs.data->id == vid) {
                vs.created = now;
-                vs.expiration = std::max(vs.expiration, now + types.getType(vs.data->type).expiration);
-                return vs.expiration;
+                return true;
            }
-        return time_point::max();
+        return false;
    }

    StoreDiff remove(const InfoHash& id, Value::Id);
@@ -233,41 +233,6 @@ Value::toJson() const
    return val;
 }

-bool
-Value::checkSignature()
-{
-    if (!signatureChecked) {
-        signatureChecked = true;
-        if (isSigned()) {
-            signatureValid = owner and owner->checkSignature(getToSign(), signature);
-        } else {
-            signatureValid = true;
-        }
-    }
-    return signatureValid;
-}
-
-Sp<Value>
-Value::decrypt(const crypto::PrivateKey& key)
-{
-    if (not decrypted) {
-        decrypted = true;
-        if (isEncrypted()) {
-            auto decryptedBlob = key.decrypt(cypher);
-            std::unique_ptr<Value> v {new Value(id)};
-            auto msg = msgpack::unpack((const char*)decryptedBlob.data(), decryptedBlob.size());
-            v->msgpack_unpack_body(msg.get());
-            if (v->recipient != key.getPublicKey().getId())
-                throw crypto::DecryptError("Recipient mismatch");
-            // Ignore values belonging to other people
-            if (not v->owner or not v->owner->checkSignature(v->getToSign(), v->signature))
-                throw crypto::DecryptError("Signature mismatch");
-            decryptedValue = std::move(v);
-        }
-    }
-    return decryptedValue;
-}
-
 uint64_t
 unpackId(const Json::Value& json, const std::string& key) {
    uint64_t ret = 0;
@@ -66,7 +66,7 @@ public:
    CallbackQueue expireValues(const time_point& now, time_point& next) {
        std::vector<Sp<Value>> expired_values;
        for (auto it = values.begin(); it != values.end();) {
-            if (it->second.expiration <= now) {
+            if (it->second.expiration < now) {
                expired_values.emplace_back(std::move(it->second.data));
                it = values.erase(it);
            } else {
@@ -74,7 +74,7 @@ public:
                ++it;
            }
        }
-        while (values.size() > MAX_VALUES) {
+        while (values.size() >= MAX_VALUES) {
            // too many values, remove oldest values
            time_point oldest_creation = time_point::max();
            auto oldest_value = values.end();
@@ -127,7 +127,7 @@ private:
    ValueCache& operator=(ValueCache&&) = delete;

    /* The maximum number of values we store in the cache. */
-    static constexpr unsigned MAX_VALUES {4096};
+    static constexpr unsigned MAX_VALUES {1024};

    struct CacheValueStorage {
        Sp<Value> data {};
@@ -46,43 +46,6 @@ CryptoTester::testSignatureEncryption() {
    CPPUNIT_ASSERT(data == decrypted);
 }

-void
-CryptoTester::testCertificateRevocation()
-{
-    auto ca1 = dht::crypto::generateIdentity("ca1");
-    auto account1 = dht::crypto::generateIdentity("acc1", ca1, 4096, true);
-    auto device11 = dht::crypto::generateIdentity("dev11", account1);
-    auto device12 = dht::crypto::generateIdentity("dev12", account1);
-
-
-    dht::crypto::TrustList list;
-    list.add(*ca1.second);
-    auto v = list.verify(*account1.second);
-    CPPUNIT_ASSERT_MESSAGE(v.toString(), v);
-
-    list.add(*account1.second);
-    v = list.verify(*device11.second);
-    CPPUNIT_ASSERT_MESSAGE(v.toString(), v);
-    v = list.verify(*device12.second);
-    CPPUNIT_ASSERT_MESSAGE(v.toString(), v);
-
-    auto ca2 = dht::crypto::generateIdentity("ca2");
-    auto account2 = dht::crypto::generateIdentity("acc2", ca2, 4096, true);
-    auto device2 = dht::crypto::generateIdentity("dev2", account2);
-
-    v = list.verify(*device2.second);
-    CPPUNIT_ASSERT_MESSAGE(v.toString(), !v);
-    
-    account1.second->revoke(*account1.first, *device11.second);
-    dht::crypto::TrustList list2;
-    list2.add(*account1.second);
-
-    v = list2.verify(*device11.second);
-    CPPUNIT_ASSERT_MESSAGE(v.toString(), !v);
-    v = list2.verify(*device12.second);
-    CPPUNIT_ASSERT_MESSAGE(v.toString(), v);
-}
-
 void
 CryptoTester::tearDown() {

@@ -28,7 +28,6 @@ namespace test {
 class CryptoTester : public CppUnit::TestFixture {
    CPPUNIT_TEST_SUITE(CryptoTester);
    CPPUNIT_TEST(testSignatureEncryption);
-    CPPUNIT_TEST(testCertificateRevocation);
    CPPUNIT_TEST_SUITE_END();

 public:
@@ -41,13 +40,9 @@ class CryptoTester : public CppUnit::TestFixture {
     */
    void tearDown();
    /**
-     * Test data signature, encryption and decryption
+     * Test the differents behaviors of constructors
     */
    void testSignatureEncryption();
-    /**
-     * Test certificate generation, validation and revocation
-     */
-    void testCertificateRevocation();
 };

 }  // namespace test
@@ -116,7 +116,10 @@ DhtProxyTester::testListen() {
    // All values
    dht::Value secondVal {"You're a monster"};
    auto secondVal_data = secondVal.data;
-    nodePeer.put(key, std::move(secondVal));
+    nodePeer.put(key, std::move(secondVal), [&](bool) {
+        done = true;
+        cv.notify_all();
+    });
    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);
@@ -1,7 +1,7 @@
 bin_PROGRAMS = dhtnode dhtchat dhtscanner
 noinst_HEADERS = tools_common.h

-AM_CPPFLAGS = -I../include @JsonCpp_CFLAGS@ @MsgPack_CFLAGS@
+AM_CPPFLAGS = -I../include

 dhtnode_SOURCES = dhtnode.cpp
 dhtnode_LDFLAGS = -lopendht -lreadline -L@top_builddir@/src/.libs @Argon2_LDFLAGS@ @GnuTLS_LIBS@
@@ -144,7 +144,7 @@ void cmd_loop(std::shared_ptr<DhtRunner>& dht, dht_params& params
 #if OPENDHT_PROXY_SERVER
            for (const auto& proxy : proxies) {
                std::cout << "Stats for proxy on port " << proxy.first << std::endl;
-                std::cout << "  " << proxy.second->stats().toString() << std::endl;
+                std::cout << "  " << proxy.second->getStats().toString() << std::endl;
            }
 #endif
            continue;
@@ -63,15 +63,11 @@ $(function() {
                        if (!element || element.length == 0)
                            return;
                        const o = JSON.parse(element);
-                        if (o.expired) {
-                            $('#value'+o.id).slideUp(100, (e) => $(e.target).remove());
-                        } else {
-                            const d = window.atob(o.data);
-                            const delay = Math.max(0, start-now);
-                            $('<li class="list-group-item" id="value'+o.id+'"/>').append(valueGetElement(o)).appendTo(group).hide().delay(delay).slideDown(100);
-                            lastAppended = i+1;
-                            start = Math.max(start, now)+25;
-                        }
+                        const d = window.atob(o.data);
+                        const delay = Math.max(0, start-now);
+                        $('<li class="list-group-item"/>').append(valueGetElement(o)).appendTo(group).hide().delay(delay).slideDown(100);
+                        lastAppended = i+1;
+                        start = Math.max(start, now)+25;
                    }
                } else if (this.status !== 0) {
                    group.empty().append($('<li class="list-group-item list-group-item-danger"/>').text("Error loading content: " + this.statusText));
Author	SHA1	Message	Date
Adrien Béraud	5fbbcb2f45	Revert "search: use pointer to SearchNode" This reverts commit `3a41f2ce56`.	2018-06-20 14:40:41 -04:00
Adrien Béraud	b10e8bbc51	search: improve canGet performance	2018-06-20 14:01:48 -04:00
Adrien Béraud	0e05b12422	network engine: make MessageStats public, add packet counter	2018-06-20 14:01:48 -04:00
Adrien Béraud	9980fc89b3	dhtrunner: add stats	2018-06-20 14:01:48 -04:00