Revert "search: use pointer to SearchNode"

This reverts commit 3a41f2ce56.

CMakeLists.txt

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

MSVC/opendht.vcxproj

@@ -146,7 +146,7 @@
       <Optimization>Disabled</Optimization>
       <SDLCheck>true</SDLCheck>
       <AdditionalIncludeDirectories>..\..\include;..\include;..\include\opendht;..\..\msgpack-c\include;</AdditionalIncludeDirectories>
-      <PreprocessorDefinitions>_CRT_SECURE_NO_WARNINGS;WIN32_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>

configure.ac

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

include/opendht/callbacks.h

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

include/opendht/dht.h

@@ -409,8 +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 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);

include/opendht/dht_proxy_server.h

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

include/opendht/dhtrunner.h

@@ -416,7 +416,6 @@ private:
      */
     void tryBootstrapContinuously();
 
-    void stopNetwork();
     void startNetwork(const SockAddr sin4, const SockAddr sin6);
     time_point loop_();
 
@@ -487,7 +486,6 @@ private:
                status6 {NodeStatus::Disconnected};
     StatusCallback statusCb {nullptr};
 
-    int stop_writefd {-1};
     int s4 {-1}, s6 {-1};
     SockAddr bound4 {};
     SockAddr bound6 {};

include/opendht/network_engine.h

@@ -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);
 

include/opendht/rng.h

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

src/callbacks.cpp

@@ -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;
 }

src/dht.cpp

@@ -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
@@ -383,94 +383,6 @@ void Dht::searchSendAnnounceValue(const Sp<Search>& sr) {
     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);
-                    }
-                });
-            }
-        }
-    };
-
     const auto& now = scheduler.time();
     for (auto& n : sr->nodes) {
         if (not n.isSynced(now))
@@ -481,6 +393,83 @@ void Dht::searchSendAnnounceValue(const Sp<Search>& sr) {
             }) == sr->announce.cend())
             continue;
 
+        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);
+                    }
+                }
+            };
         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;
@@ -764,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;
@@ -915,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, {}, true);
 
     DHT_LOG.d(id, "put: adding %s -> %s", id.toString().c_str(), val->toString().c_str());
 
@@ -1132,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())
@@ -1202,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;
 
@@ -1228,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();
         }
@@ -1240,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)
 {
@@ -1610,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)
@@ -1618,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();
 }
 
@@ -2374,10 +2348,7 @@ Dht::onRefresh(Sp<Node> node, const InfoHash& hash, const Blob& token, const Val
     }
 
     auto s = store.find(hash);
-    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, 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",

src/dht_proxy_client.cpp

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

src/dht_proxy_server.cpp

@@ -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,20 +178,21 @@ 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
@@ -234,36 +234,6 @@ DhtProxyServer::getNodeInfo(const std::shared_ptr<restbed::Session>& session) co
     );
 }
 
-void
-DhtProxyServer::getStats(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 std::shared_ptr<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 std::shared_ptr<restbed::Session>& session) const
 {

src/dhtrunner.cpp

@@ -164,7 +164,7 @@ DhtRunner::shutdown(ShutdownCallback cb) {
 void
 DhtRunner::join()
 {
-    stopNetwork();
+    running_network = false;
     running = false;
     cv.notify_all();
     bootstrap_cv.notify_all();
@@ -410,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);
     }
@@ -458,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);
@@ -469,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;
 
@@ -541,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");
@@ -574,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()});
                         }
@@ -599,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;
     });
 }
 
@@ -642,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;
@@ -650,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();
@@ -957,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();

src/network_engine.cpp

@@ -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;
@@ -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));

src/search.h

@@ -320,7 +320,7 @@ 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);
@@ -335,7 +335,7 @@ struct Dht::SearchNode {
     }
 
     /**
-     * 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 {

src/storage.h

@@ -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);

src/value_cache.h

@@ -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 {};

tests/cryptotester.cpp

@@ -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() {
 

tests/cryptotester.h

@@ -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

tests/dhtproxytester.cpp

@@ -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);

tools/dhtnode.cpp

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