make move operators noexcept

Server thread pool

CMakeLists.txt

@@ -184,40 +184,34 @@ if (OPENDHT_INDEX)
 endif()
 
 if (OPENDHT_PROXY_SERVER)
-  add_definitions(-DOPENDHT_PROXY_SERVER=true)
+  add_definitions(-DOPENDHT_PROXY_SERVER)
   if (OPENDHT_PROXY_SERVER_IDENTITY)
-    add_definitions(-DOPENDHT_PROXY_SERVER_IDENTITY=true)
-  else ()
-    add_definitions(-DOPENDHT_PROXY_SERVER_IDENTITY=false)
+    add_definitions(-DOPENDHT_PROXY_SERVER_IDENTITY)
   endif()
   list (APPEND opendht_HEADERS
     include/opendht/dht_proxy_server.h
   )
   list (APPEND opendht_SOURCES
+    src/thread_pool.h
+    src/thread_pool.cpp
     src/dht_proxy_server.cpp
   )
-else ()
-    add_definitions(-DENABLE_PROXY_SERVER=false)
 endif ()
 
 if (OPENDHT_PROXY_CLIENT)
-  add_definitions(-DOPENDHT_PROXY_CLIENT=true)
+  add_definitions(-DOPENDHT_PROXY_CLIENT)
   list (APPEND opendht_HEADERS
     include/opendht/dht_proxy_client.h
   )
   list (APPEND opendht_SOURCES
     src/dht_proxy_client.cpp
   )
-else ()
-  add_definitions(-DOPENDHT_PROXY_CLIENT=false)
 endif ()
 
 if (OPENDHT_PROXY_SERVER OR OPENDHT_PROXY_CLIENT)
   if (OPENDHT_PUSH_NOTIFICATIONS)
     message("Using push notification")
-    add_definitions(-DOPENDHT_PUSH_NOTIFICATIONS=true)
-  else ()
-    add_definitions(-DOPENDHT_PUSH_NOTIFICATIONS=false)
+    add_definitions(-DOPENDHT_PUSH_NOTIFICATIONS)
   endif ()
   list (APPEND opendht_HEADERS
     include/opendht/proxy.h

configure.ac

@@ -177,21 +177,10 @@ AM_COND_IF([ENABLE_TOOLS], [
   ])
 ])
 
-AM_COND_IF(ENABLE_PROXY_SERVER,
-    [CPPFLAGS+=" -DOPENDHT_PROXY_SERVER=true"],
-    [CPPFLAGS+=" -DOPENDHT_PROXY_SERVER=false"])
-
-AM_COND_IF(ENABLE_PROXY_CLIENT,
-    [CPPFLAGS+=" -DOPENDHT_PROXY_CLIENT=true"],
-    [CPPFLAGS+=" -DOPENDHT_PROXY_CLIENT=false"])
-
-AM_COND_IF(ENABLE_PUSH_NOTIFICATIONS,
-    [CPPFLAGS+=" -DOPENDHT_PUSH_NOTIFICATIONS=true"],
-    [CPPFLAGS+=" -DOPENDHT_PUSH_NOTIFICATIONS=false"])
-
-AM_COND_IF(ENABLE_PROXY_SERVER_IDENTITY,
-    [CPPFLAGS+=" -DOPENDHT_PROXY_SERVER_IDENTITY=true"],
-    [CPPFLAGS+=" -DOPENDHT_PROXY_SERVER_IDENTITY=false"])
+AM_COND_IF(ENABLE_PROXY_SERVER, [CPPFLAGS+=" -DOPENDHT_PROXY_SERVER"], [])
+AM_COND_IF(ENABLE_PROXY_CLIENT, [CPPFLAGS+=" -DOPENDHT_PROXY_CLIENT"], [])
+AM_COND_IF(ENABLE_PUSH_NOTIFICATIONS, [CPPFLAGS+=" -DOPENDHT_PUSH_NOTIFICATIONS"], [])
+AM_COND_IF(ENABLE_PROXY_SERVER_IDENTITY, [CPPFLAGS+=" -DOPENDHT_PROXY_SERVER_IDENTITY"], [])
 
 AM_COND_IF([HAVE_DOXYGEN], [
   AC_CONFIG_FILES([doc/Doxyfile doc/Makefile])

include/opendht.h

@@ -19,7 +19,7 @@
 #pragma once
 
 #include "opendht/dhtrunner.h"
-#if OPENDHT_PROXY_SERVER
+#ifdef OPENDHT_PROXY_SERVER
 #include "opendht/dht_proxy_server.h"
 #endif
 #include "opendht/log.h"

include/opendht/callbacks.h

@@ -103,6 +103,9 @@ struct OPENDHT_PUBLIC Config {
 
     /** Makes the DHT responsible to maintain its stored values. Consumes more ressources. */
     bool maintain_storage;
+
+    /** If set, the dht will load its state from this file on start and save its state in this file on shutdown */
+    std::string persist_path;
 };
 
 /**

include/opendht/crypto.h

@@ -186,7 +186,7 @@ class OPENDHT_PUBLIC RevocationList
 public:
     RevocationList();
     RevocationList(const Blob& b);
-    RevocationList(RevocationList&& o) : crl(o.crl) { o.crl = nullptr; }
+    RevocationList(RevocationList&& o) noexcept : crl(o.crl) { o.crl = nullptr; }
     ~RevocationList();
 
     RevocationList& operator=(RevocationList&& o) { crl = o.crl; o.crl = nullptr; return *this; }
@@ -500,7 +500,7 @@ struct OPENDHT_PUBLIC TrustList
     };
 
     TrustList();
-    TrustList(TrustList&& o) : trust(std::move(o.trust)) {
+    TrustList(TrustList&& o) noexcept : trust(std::move(o.trust)) {
         o.trust = nullptr;
     }
     TrustList& operator=(TrustList&& o);

include/opendht/dht.h

@@ -261,11 +261,14 @@ public:
      * Get the list of good nodes for local storage saving purposes
      * The list is ordered to minimize the back-to-work delay.
      */
-    std::vector<NodeExport> exportNodes();
+    std::vector<NodeExport> exportNodes() const;
 
     std::vector<ValuesExport> exportValues() const;
     void importValues(const std::vector<ValuesExport>&);
 
+    void saveState(const std::string& path) const;
+    void loadState(const std::string& path);
+
     NodeStats getNodesStats(sa_family_t af) const;
 
     std::string getStorageLog() const;
@@ -359,13 +362,6 @@ private:
     // registred types
     TypeStore types;
 
-    // are we a bootstrap node ?
-    // note: Any running node can be used as a bootstrap node.
-    //       Only nodes running only as bootstrap nodes should
-    //       be put in bootstrap mode.
-    const bool is_bootstrap {false};
-    const bool maintain_storage {false};
-
     // the stuff
     RoutingTable buckets4 {};
     RoutingTable buckets6 {};
@@ -400,6 +396,15 @@ private:
 
     std::mt19937_64 rd {crypto::getSeededRandomEngine<std::mt19937_64>()};
 
+    std::string persistPath;
+
+    // are we a bootstrap node ?
+    // note: Any running node can be used as a bootstrap node.
+    //       Only nodes running only as bootstrap nodes should
+    //       be put in bootstrap mode.
+    const bool is_bootstrap {false};
+    const bool maintain_storage {false};
+
     void rotateSecrets();
 
     Blob makeToken(const SockAddr&, bool old) const;

include/opendht/dht_interface.h

@@ -181,7 +181,7 @@ public:
      * Get the list of good nodes for local storage saving purposes
      * The list is ordered to minimize the back-to-work delay.
      */
-    virtual std::vector<NodeExport> exportNodes() = 0;
+    virtual std::vector<NodeExport> exportNodes() const = 0;
 
     virtual std::vector<ValuesExport> exportValues() const = 0;
     virtual void importValues(const std::vector<ValuesExport>&) = 0;

include/opendht/dht_proxy_client.h

@@ -17,8 +17,6 @@
  *  along with this program. If not, see <https://www.gnu.org/licenses/>.
  */
 
-#if OPENDHT_PROXY_CLIENT
-
 #pragma once
 
 #include <functional>
@@ -49,7 +47,7 @@ public:
     explicit DhtProxyClient(std::function<void()> loopSignal, const std::string& serverHost, const std::string& pushClientId = "");
 
     virtual void setPushNotificationToken(const std::string& token) {
-#if OPENDHT_PUSH_NOTIFICATIONS
+#ifdef OPENDHT_PUSH_NOTIFICATIONS
         deviceKey_ = token;
 #endif
     }
@@ -237,7 +235,7 @@ public:
     void insertNode(const InfoHash&, const sockaddr*, socklen_t) { }
     void insertNode(const NodeExport&) { }
     std::pair<size_t, size_t> getStoreSize() const { return {}; }
-    std::vector<NodeExport> exportNodes() { return {}; }
+    std::vector<NodeExport> exportNodes() const { return {}; }
     std::vector<ValuesExport> exportValues() const { return {}; }
     void importValues(const std::vector<ValuesExport>&) {}
     std::string getStorageLog() const { return {}; }
@@ -376,7 +374,7 @@ private:
 
     const std::function<void()> loopSignal_;
 
-#if OPENDHT_PUSH_NOTIFICATIONS
+#ifdef OPENDHT_PUSH_NOTIFICATIONS
     void fillBody(std::shared_ptr<restbed::Request> request, bool resubscribe);
     void getPushRequest(Json::Value&) const;
 #endif // OPENDHT_PUSH_NOTIFICATIONS
@@ -385,5 +383,3 @@ private:
 };
 
 }
-
-#endif // OPENDHT_PROXY_CLIENT

include/opendht/dht_proxy_server.h

@@ -17,8 +17,6 @@
  *  along with this program. If not, see <https://www.gnu.org/licenses/>.
  */
 
-#if OPENDHT_PROXY_SERVER
-
 #pragma once
 
 #include "callbacks.h"
@@ -45,6 +43,7 @@ namespace Json {
 namespace dht {
 
 class DhtRunner;
+class ThreadPool;
 
 /**
  * Describes the REST API
@@ -178,7 +177,7 @@ private:
 
     void cancelPut(const InfoHash& key, Value::Id vid);
 
-#if OPENDHT_PROXY_SERVER_IDENTITY
+#ifdef OPENDHT_PROXY_SERVER_IDENTITY
     /**
      * Put a value to sign by the proxy on the DHT
      * Method: SIGN "/{InfoHash: .*}"
@@ -229,7 +228,7 @@ private:
      */
     void removeClosedListeners(bool testSession = true);
 
-#if OPENDHT_PUSH_NOTIFICATIONS
+#ifdef OPENDHT_PUSH_NOTIFICATIONS
     /**
      * Subscribe to push notifications for an iOS or Android device.
      * Method: SUBSCRIBE "/{InfoHash: .*}"
@@ -253,7 +252,7 @@ private:
      * @param key of the device
      * @param json, the content to send
      */
-    void sendPushNotification(const std::string& key, const Json::Value& json, bool isAndroid) const;
+    void sendPushNotification(const std::string& key, Json::Value&& json, bool isAndroid) const;
 
     /**
      * Remove a push listener between a client and a hash
@@ -272,11 +271,13 @@ private:
     std::thread server_thread {};
     std::unique_ptr<restbed::Service> service_;
     std::shared_ptr<DhtRunner> dht_;
+    Json::StreamWriterBuilder jsonBuilder_;
 
     std::mutex schedulerLock_;
     std::condition_variable schedulerCv_;
     Scheduler scheduler_;
     std::thread schedulerThread_;
+    std::unique_ptr<ThreadPool> threadPool_;
 
     Sp<Scheduler::Job> printStatsJob_;
     mutable std::mutex statsMutex_;
@@ -305,7 +306,7 @@ private:
 
     mutable ServerStats stats_;
 
-#if OPENDHT_PUSH_NOTIFICATIONS
+#ifdef OPENDHT_PUSH_NOTIFICATIONS
     struct Listener;
     struct PushListener;
     std::mutex lockPushListeners_;
@@ -315,5 +316,3 @@ private:
 };
 
 }
-
-#endif //OPENDHT_PROXY_SERVER

include/opendht/dhtrunner.h

@@ -354,7 +354,8 @@ public:
                     /*.node_id = */{},
                     /*.network = */network,
                     /*.is_bootstrap = */false,
-                    /*.maintain_storage*/false
+                    /*.maintain_storage = */false,
+                    /*.persist_path = */{}
                 },
                 /*.id = */identity
             },

include/opendht/securedht.h

@@ -168,7 +168,7 @@ public:
     void setStorageLimit(size_t limit = DEFAULT_STORAGE_LIMIT) override {
         dht_->setStorageLimit(limit);
     }
-    std::vector<NodeExport> exportNodes() override {
+    std::vector<NodeExport> exportNodes() const override {
         return dht_->exportNodes();
     }
     std::vector<ValuesExport> exportValues() const override {

include/opendht/sockaddr.h

@@ -58,7 +58,7 @@ public:
     SockAddr(const SockAddr& o) {
         set(o.get(), o.getLength());
     }
-    SockAddr(SockAddr&& o) : len(o.len), addr(std::move(o.addr)) {
+    SockAddr(SockAddr&& o) noexcept : len(o.len), addr(std::move(o.addr)) {
         o.len = 0;
     }
 

src/Makefile.am

@@ -56,7 +56,7 @@ nobase_include_HEADERS = \
         ../include/opendht/rng.h
 
 if ENABLE_PROXY_SERVER
-libopendht_la_SOURCES += dht_proxy_server.cpp
+libopendht_la_SOURCES += dht_proxy_server.cpp thread_pool.h thread_pool.cpp
 nobase_include_HEADERS += ../include/opendht/dht_proxy_server.h
 endif
 

src/dht.cpp

@@ -29,6 +29,7 @@
 #include <algorithm>
 #include <random>
 #include <sstream>
+#include <fstream>
 
 namespace dht {
 
@@ -54,6 +55,9 @@ Dht::getStatus(sa_family_t af) const
 void
 Dht::shutdown(ShutdownCallback cb)
 {
+    if (not persistPath.empty())
+        saveState(persistPath);
+
     if (not maintain_storage) {
         if (cb) cb();
         return;
@@ -1722,9 +1726,7 @@ Dht::~Dht()
 Dht::Dht() : store(), network_engine(DHT_LOG, scheduler) {}
 
 Dht::Dht(const int& s, const int& s6, Config config)
-    : myid(config.node_id ? config.node_id : InfoHash::getRandom()),
-    is_bootstrap(config.is_bootstrap),
-    maintain_storage(config.maintain_storage), store(), store_quota(),
+    : myid(config.node_id ? config.node_id : InfoHash::getRandom()), store(), store_quota(),
     network_engine(myid, config.network, s, s6, DHT_LOG, scheduler,
             std::bind(&Dht::onError, this, _1, _2),
             std::bind(&Dht::onNewNode, this, _1, _2),
@@ -1734,7 +1736,10 @@ Dht::Dht(const int& s, const int& s6, Config config)
             std::bind(&Dht::onGetValues, this, _1, _2, _3, _4),
             std::bind(&Dht::onListen, this, _1, _2, _3, _4, _5),
             std::bind(&Dht::onAnnounce, this, _1, _2, _3, _4, _5),
-            std::bind(&Dht::onRefresh, this, _1, _2, _3, _4))
+            std::bind(&Dht::onRefresh, this, _1, _2, _3, _4)),
+    persistPath(config.persist_path),
+    is_bootstrap(config.is_bootstrap),
+    maintain_storage(config.maintain_storage)
 {
     scheduler.syncTime();
     if (s < 0 && s6 < 0)
@@ -1765,8 +1770,10 @@ Dht::Dht(const int& s, const int& s6, Config config)
     expire();
 
     DHT_LOG.d("DHT initialised with node ID %s", myid.toString().c_str());
-}
 
+    if (not persistPath.empty())
+        loadState(persistPath);
+}
 
 bool
 Dht::neighbourhoodMaintenance(RoutingTable& list)
@@ -2056,7 +2063,7 @@ Dht::importValues(const std::vector<ValuesExport>& import)
 
 
 std::vector<NodeExport>
-Dht::exportNodes()
+Dht::exportNodes() const
 {
     const auto& now = scheduler.time();
     std::vector<NodeExport> nodes;
@@ -2283,7 +2290,7 @@ Dht::onListen(Sp<Node> node, const InfoHash& hash, const Blob& token, size_t soc
 }
 
 void
-Dht::onListenDone(const Sp<Node>& node, net::RequestAnswer& answer, Sp<Search>& sr)
+Dht::onListenDone(const Sp<Node>& /* node */, net::RequestAnswer& /* answer */, Sp<Search>& sr)
 {
     // DHT_LOG.d(sr->id, node->id, "[search %s] [node %s] got listen confirmation",
     //            sr->id.toString().c_str(), node->toString().c_str(), answer.values.size());
@@ -2424,4 +2431,52 @@ Dht::onAnnounceDone(const Sp<Node>& node, net::RequestAnswer& answer, Sp<Search>
     sr->checkAnnounced(answer.vid);
 }
 
+
+void
+Dht::saveState(const std::string& path) const
+{
+    std::ofstream file(path);
+    msgpack::pack(file, exportNodes());
+    msgpack::pack(file, exportValues());
+}
+
+void
+Dht::loadState(const std::string& path)
+{
+    DHT_LOG.d("Importing state from %s", path.c_str());
+    try {
+        // Import nodes from binary file
+        msgpack::unpacker pac;
+        {
+            // Read whole file
+            std::ifstream file(path, std::ios::binary|std::ios::ate);
+            if (!file.is_open()) {
+                return;
+            }
+            auto size = file.tellg();
+            file.seekg (0, std::ios::beg);
+            pac.reserve_buffer(size);
+            file.read (pac.buffer(), size);
+            pac.buffer_consumed(size);
+        }
+        // Import nodes
+        msgpack::object_handle oh;
+        if (pac.next(oh)) {
+            {
+                auto imported_nodes = oh.get().as<std::vector<NodeExport>>();
+                DHT_LOG.d("Importing %zu nodes", imported_nodes.size());
+                for (const auto& node : imported_nodes)
+                    insertNode(node);
+            }
+            if (pac.next(oh)) {
+                auto imported_values = oh.get().as<std::vector<ValuesExport>>();
+                DHT_LOG.d("Importing %zu values", imported_values.size());
+                importValues(imported_values);
+            }
+        }
+    } catch (const std::exception& e) {
+        DHT_LOG.w("Error importing state from %s: %s", path.c_str(), e.what());
+    }
+}
+
 }

src/dht_proxy_client.cpp

@@ -17,8 +17,6 @@
  *  along with this program. If not, see <https://www.gnu.org/licenses/>.
  */
 
-#if OPENDHT_PROXY_CLIENT
-
 #include "dht_proxy_client.h"
 
 #include "dhtrunner.h"
@@ -45,8 +43,7 @@ struct DhtProxyClient::ListenState {
 
 struct DhtProxyClient::Listener
 {
-    ValueCache cache;
-    Sp<Scheduler::Job> cacheExpirationJob {};
+    OpValueCache cache;
     ValueCallback cb;
     Value::Filter filter;
     Sp<restbed::Request> req;
@@ -54,23 +51,21 @@ struct DhtProxyClient::Listener
     unsigned callbackId;
     Sp<ListenState> state;
     Sp<Scheduler::Job> refreshJob;
-    Listener(ValueCache&& c, Sp<Scheduler::Job>&& j, const Sp<restbed::Request>& r, Value::Filter&& f)
-        : cache(std::move(c)),
-          cacheExpirationJob(std::move(j)),
-          filter(std::move(f)),
-          req(r)
-    {}
+    Listener(OpValueCache&& c, const Sp<restbed::Request>& r, Value::Filter&& f)
+        : cache(std::move(c)), filter(std::move(f)),req(r) {}
 };
 
 struct PermanentPut {
     Sp<Value> value;
     Sp<Scheduler::Job> refreshJob;
     Sp<std::atomic_bool> ok;
+    PermanentPut(const Sp<Value>& v, Sp<Scheduler::Job>&& j, const Sp<std::atomic_bool>& o)
+        : value(v), refreshJob(std::move(j)), ok(o) {}
 };
 
 struct DhtProxyClient::ProxySearch {
     SearchCache ops {};
-    Sp<Scheduler::Job> opExpirationJob;
+    Sp<Scheduler::Job> opExpirationJob {};
     std::map<size_t, Listener> listeners {};
     std::map<Value::Id, PermanentPut> puts {};
 };
@@ -338,24 +333,26 @@ DhtProxyClient::put(const InfoHash& key, Sp<Value> val, DoneCallback cb, time_po
     if (permanent) {
         std::lock_guard<std::mutex> lock(searchLock_);
         auto id = val->id;
-        auto search = searches_.emplace(key, ProxySearch{}).first;
+        auto& search = searches_[key];
         auto nextRefresh = scheduler.time() + proxy::OP_TIMEOUT - proxy::OP_MARGIN;
         auto ok = std::make_shared<std::atomic_bool>(false);
-        search->second.puts.erase(id);
-        search->second.puts.emplace(id, PermanentPut {val, scheduler.add(nextRefresh, [this, key, id, ok]{
-            std::lock_guard<std::mutex> lock(searchLock_);
-            auto s = searches_.find(key);
-            if (s != searches_.end()) {
-                auto p = s->second.puts.find(id);
-                if (p != s->second.puts.end()) {
-                    doPut(key, p->second.value,
-                    [ok](bool result, const std::vector<std::shared_ptr<dht::Node> >&){
-                        *ok = result;
-                    }, time_point::max(), true);
-                    scheduler.edit(p->second.refreshJob, scheduler.time() + proxy::OP_TIMEOUT - proxy::OP_MARGIN);
+        search.puts.erase(id);
+        search.puts.emplace(std::piecewise_construct,
+            std::forward_as_tuple(id),
+            std::forward_as_tuple(val, scheduler.add(nextRefresh, [this, key, id, ok]{
+                std::lock_guard<std::mutex> lock(searchLock_);
+                auto s = searches_.find(key);
+                if (s != searches_.end()) {
+                    auto p = s->second.puts.find(id);
+                    if (p != s->second.puts.end()) {
+                        doPut(key, p->second.value,
+                        [ok](bool result, const std::vector<std::shared_ptr<dht::Node> >&){
+                            *ok = result;
+                        }, time_point::max(), true);
+                        scheduler.edit(p->second.refreshJob, scheduler.time() + proxy::OP_TIMEOUT - proxy::OP_MARGIN);
+                    }
                 }
-            }
-        }), ok});
+            }), ok));
     }
     doPut(key, val, std::move(cb), created, permanent);
 }
@@ -373,7 +370,7 @@ DhtProxyClient::doPut(const InfoHash& key, Sp<Value> val, DoneCallback cb, time_
         if (deviceKey_.empty()) {
             json["permanent"] = true;
         } else {
-#if OPENDHT_PUSH_NOTIFICATIONS
+#ifdef OPENDHT_PUSH_NOTIFICATIONS
             Json::Value refresh;
             getPushRequest(refresh);
             json["permanent"] = refresh;
@@ -663,12 +660,9 @@ DhtProxyClient::getPublicAddress(sa_family_t family)
 size_t
 DhtProxyClient::listen(const InfoHash& key, ValueCallback cb, Value::Filter filter, Where where) {
     DHT_LOG.d(key, "[search %s]: listen", key.to_c_str());
-    auto it = searches_.find(key);
-    if (it == searches_.end()) {
-        it = searches_.emplace(key, ProxySearch{}).first;
-    }
+    auto& search = searches_[key];
     auto query = std::make_shared<Query>(Select{}, where);
-    auto token = it->second.ops.listen(cb, query, filter, [this, key, filter](Sp<Query> /*q*/, ValueCallback cb, SyncCallback scb) -> size_t {
+    auto token = search.ops.listen(cb, query, filter, [this, key, filter](Sp<Query> /*q*/, ValueCallback cb, SyncCallback /*scb*/) -> size_t {
         scheduler.syncTime();
         restbed::Uri uri(serverHost_ + "/" + key.toString());
         std::lock_guard<std::mutex> lock(searchLock_);
@@ -684,21 +678,9 @@ DhtProxyClient::listen(const InfoHash& key, ValueCallback cb, Value::Filter filt
         auto l = search->second.listeners.find(token);
         if (l == search->second.listeners.end()) {
             auto f = filter;
-            l = search->second.listeners.emplace(token, Listener {
-                ValueCache(cb, std::move(scb)), scheduler.add(time_point::max(), [this, key, token]{
-                    std::lock_guard<std::mutex> lock(searchLock_);
-                    auto s = searches_.find(key);
-                    if (s == searches_.end()) {
-                        return;
-                    }
-                    auto l = s->second.listeners.find(token);
-                    if (l == s->second.listeners.end()) {
-                        return;
-                    }
-                    auto next = l->second.cache.expireValues(scheduler.time());
-                    scheduler.edit(l->second.cacheExpirationJob, next);
-                }), req, std::move(f)
-            }).first;
+            l = search->second.listeners.emplace(std::piecewise_construct,
+                    std::forward_as_tuple(token),
+                    std::forward_as_tuple(std::move(cb), req, std::move(f))).first;
         } else {
             if (l->second.state)
                 l->second.state->cancel = true;
@@ -718,17 +700,7 @@ DhtProxyClient::listen(const InfoHash& key, ValueCallback cb, Value::Filter filt
             if (l == s->second.listeners.end()) {
                 return false;
             }
-            const std::vector<Sp<Value>> new_values_empty;
-            std::vector<Value::Id> expired_ids;
-            if (expired) {
-                expired_ids.reserve(values.size());
-                for (const auto& v : values)
-                    expired_ids.emplace_back(v->id);
-            }
-            auto next = l->second.cache.onValues(expired ? new_values_empty : values, std::vector<Value::Id>{}, expired_ids, types, scheduler.time());
-            scheduler.edit(l->second.cacheExpirationJob, next);
-            loopSignal_();
-            return true;
+            return l->second.cache.onValue(values, expired);
         };
 
         auto vcb = l->second.cb;
@@ -751,8 +723,7 @@ DhtProxyClient::listen(const InfoHash& key, ValueCallback cb, Value::Filter filt
         }
         l->second.thread = std::thread([this, req, vcb, filter, state]() {
             sendListen(req, vcb, filter, state,
-                    deviceKey_.empty() ? ListenMethod::LISTEN
-                                    : ListenMethod::SUBSCRIBE);
+                    deviceKey_.empty() ? ListenMethod::LISTEN : ListenMethod::SUBSCRIBE);
         });
         return token;
     });
@@ -791,23 +762,23 @@ void DhtProxyClient::sendListen(const std::shared_ptr<restbed::Request> &req,
                                 const Value::Filter &filter,
                                 const Sp<ListenState> &state,
                                 ListenMethod method) {
-  auto settings = std::make_shared<restbed::Settings>();
-  if (method != ListenMethod::LISTEN) {
-    req->set_method("SUBSCRIBE");
-  } else {
-    std::chrono::milliseconds timeout(std::numeric_limits<int>::max());
-    settings->set_connection_timeout(
-        timeout); // Avoid the client to close the socket after 5 seconds.
-    req->set_method("LISTEN");
-  }
-  try {
-#if OPENDHT_PUSH_NOTIFICATIONS
-    if (method != ListenMethod::LISTEN)
-      fillBody(req, method == ListenMethod::RESUBSCRIBE);
-#endif
+    auto settings = std::make_shared<restbed::Settings>();
+    if (method != ListenMethod::LISTEN) {
+        req->set_method("SUBSCRIBE");
+    } else {
+        std::chrono::milliseconds timeout(std::numeric_limits<int>::max());
+        settings->set_connection_timeout(timeout); // Avoid the client to close the socket after 5 seconds.
+        req->set_method("LISTEN");
+    }
+    try {
+#ifdef OPENDHT_PUSH_NOTIFICATIONS
+        if (method != ListenMethod::LISTEN)
+        fillBody(req, method == ListenMethod::RESUBSCRIBE);
+    #endif
         restbed::Http::async(req,
-            [this, filter, cb, state](const std::shared_ptr<restbed::Request>& req,
-                                    const std::shared_ptr<restbed::Response>& reply) {
+              [this, filter, cb, state](const std::shared_ptr<restbed::Request>& req,
+                                        const std::shared_ptr<restbed::Response>& reply)
+        {
             auto code = reply->get_status_code();
             if (code == 200) {
                 try {
@@ -994,7 +965,7 @@ DhtProxyClient::restartListeners()
 void
 DhtProxyClient::pushNotificationReceived(const std::map<std::string, std::string>& notification)
 {
-#if OPENDHT_PUSH_NOTIFICATIONS
+#ifdef OPENDHT_PUSH_NOTIFICATIONS
     scheduler.syncTime();
     {
         // If a push notification is received, the proxy is up and running
@@ -1021,20 +992,45 @@ DhtProxyClient::pushNotificationReceived(const std::map<std::string, std::string
                     resubscribe(key, list.second);
             }
         } else {
-            auto keyidx = notification.find("key");
-            InfoHash key(keyidx->second);
-
-            for (auto& search: searches_) {
-                for (auto& list : search.second.listeners) {
-                    if (search.first != key || list.second.state->cancel)
-                        continue;
-                    DHT_LOG.d(search.first, "[search %s] handling push notification", search.first.to_c_str());
+            auto key = InfoHash(notification.at("key"));
+            auto& search = searches_.at(key);
+            for (auto& list : search.listeners) {
+                if (list.second.state->cancel)
+                    continue;
+                DHT_LOG.d(key, "[search %s] handling push notification", key.to_c_str());
+                auto expired = notification.find("exp");
+                auto token = list.first;
+                auto state = list.second.state;
+                if (expired == notification.end()) {
                     auto cb = list.second.cb;
                     auto filter = list.second.filter;
-                    get(search.first, [cb](const std::vector<Sp<Value>>& vals) {
-                        cb(vals, false);
-                        return true;
-                    }, DoneCallbackSimple{}, std::move(filter));
+                    auto oldValues = list.second.cache.getValues();
+                    get(key, [cb](const std::vector<Sp<Value>>& vals) {
+                        return cb(vals, false);
+                    }, [cb, oldValues](bool /*ok*/) {
+                        // Decrement old values refcount to expire values not present in the new list
+                        cb(oldValues, true);
+                    }, std::move(filter));
+                } else {
+                    std::stringstream ss(expired->second);
+                    std::vector<Value::Id> ids;
+                    while(ss.good()){
+                        std::string substr;
+                        getline(ss, substr, ',');
+                        ids.emplace_back(std::stoull(substr));
+                    }
+                    std::lock_guard<std::mutex> lock(lockCallbacks);
+                    callbacks_.emplace_back([this, key, token, state, ids]() {
+                        if (state->cancel) return;
+                        std::lock_guard<std::mutex> lock(searchLock_);
+                        auto s = searches_.find(key);
+                        if (s == searches_.end()) return;
+                        auto l = s->second.listeners.find(token);
+                        if (l == s->second.listeners.end()) return;
+                        if (not state->cancel and not l->second.cache.onValuesExpired(ids))
+                            state->cancel = true;
+                    });
+                    loopSignal_();
                 }
             }
         }
@@ -1047,20 +1043,22 @@ DhtProxyClient::pushNotificationReceived(const std::map<std::string, std::string
 void
 DhtProxyClient::resubscribe(const InfoHash& key, Listener& listener)
 {
-#if OPENDHT_PUSH_NOTIFICATIONS
+#ifdef OPENDHT_PUSH_NOTIFICATIONS
     if (deviceKey_.empty()) return;
     scheduler.syncTime();
     DHT_LOG.d(key, "[search %s] resubscribe push listener", key.to_c_str());
     // Subscribe
-    restbed::Uri uri(serverHost_ + "/" + key.toString());
-    auto req = std::make_shared<restbed::Request>(uri);
-    req->set_method("SUBSCRIBE");
-
     auto state = listener.state;
-    if (listener.thread.joinable())
+    if (listener.thread.joinable()) {
+        state->cancel = true;
+        if (listener.req)
+            restbed::Http::close(listener.req);
         listener.thread.join();
+    }
     state->cancel = false;
     state->ok = true;
+    auto req = std::make_shared<restbed::Request>(restbed::Uri {serverHost_ + "/" + key.toString()});
+    req->set_method("SUBSCRIBE");
     listener.req = req;
     scheduler.edit(listener.refreshJob, scheduler.time() + proxy::OP_TIMEOUT - proxy::OP_MARGIN);
     auto vcb = listener.cb;
@@ -1071,7 +1069,7 @@ DhtProxyClient::resubscribe(const InfoHash& key, Listener& listener)
 #endif
 }
 
-#if OPENDHT_PUSH_NOTIFICATIONS
+#ifdef OPENDHT_PUSH_NOTIFICATIONS
 void
 DhtProxyClient::getPushRequest(Json::Value& body) const
 {
@@ -1094,9 +1092,9 @@ DhtProxyClient::fillBody(std::shared_ptr<restbed::Request> req, bool resubscribe
     // }
     Json::Value body;
     getPushRequest(body);
-    if (!resubscribe) {
+    if (resubscribe) {
         // This is the first listen, we want to retrieve previous values.
-        body["previous_values"] = true;
+        body["refresh"] = true;
     }
     Json::StreamWriterBuilder wbuilder;
     wbuilder["commentStyle"] = "None";
@@ -1109,5 +1107,3 @@ DhtProxyClient::fillBody(std::shared_ptr<restbed::Request> req, bool resubscribe
 #endif // OPENDHT_PUSH_NOTIFICATIONS
 
 } // namespace dht
-
-#endif // OPENDHT_PROXY_CLIENT

src/dht_proxy_server.cpp

@@ -17,9 +17,9 @@
  *  along with this program. If not, see <https://www.gnu.org/licenses/>.
  */
 
-#if OPENDHT_PROXY_SERVER
 #include "dht_proxy_server.h"
 
+#include "thread_pool.h"
 #include "default_types.h"
 #include "dhtrunner.h"
 
@@ -47,9 +47,11 @@ struct DhtProxyServer::SearchPuts {
 };
 
 constexpr const std::chrono::minutes PRINT_STATS_PERIOD {2};
+constexpr const size_t IO_THREADS_MAX {64};
+
 
 DhtProxyServer::DhtProxyServer(std::shared_ptr<DhtRunner> dht, in_port_t port , const std::string& pushServer)
-: dht_(dht) , pushServer_(pushServer)
+: dht_(dht), threadPool_(new ThreadPool(IO_THREADS_MAX)), pushServer_(pushServer)
 {
     if (not dht_)
         throw std::invalid_argument("A DHT instance must be provided");
@@ -58,13 +60,16 @@ DhtProxyServer::DhtProxyServer(std::shared_ptr<DhtRunner> dht, in_port_t port ,
 
     std::cout << "Running DHT proxy server on port " << port << std::endl;
     if (not pushServer.empty()) {
-#if !OPENDHT_PUSH_NOTIFICATIONS
-        std::cerr << "Push server defined but built OpenDHT built without push notification support" << std::endl;
-#else
+#ifdef OPENDHT_PUSH_NOTIFICATIONS
         std::cout << "Using push notification server: " << pushServer << std::endl;
+#else
+        std::cerr << "Push server defined but built OpenDHT built without push notification support" << std::endl;
 #endif
     }
 
+    jsonBuilder_["commentStyle"] = "None";
+    jsonBuilder_["indentation"] = "";
+
     server_thread = std::thread([this, port]() {
         // Create endpoints
         auto resource = std::make_shared<restbed::Resource>();
@@ -76,12 +81,12 @@ DhtProxyServer::DhtProxyServer(std::shared_ptr<DhtRunner> dht, in_port_t port ,
         resource->set_path("/{hash: .*}");
         resource->set_method_handler("GET", std::bind(&DhtProxyServer::get, this, _1));
         resource->set_method_handler("LISTEN", [this](const Sp<restbed::Session>& session) mutable { listen(session); } );
-#if OPENDHT_PUSH_NOTIFICATIONS
+#ifdef OPENDHT_PUSH_NOTIFICATIONS
         resource->set_method_handler("SUBSCRIBE", [this](const Sp<restbed::Session>& session) mutable { subscribe(session); } );
         resource->set_method_handler("UNSUBSCRIBE", [this](const Sp<restbed::Session>& session) mutable { unsubscribe(session); } );
 #endif //OPENDHT_PUSH_NOTIFICATIONS
         resource->set_method_handler("POST", [this](const Sp<restbed::Session>& session) mutable { put(session); });
-#if OPENDHT_PROXY_SERVER_IDENTITY
+#ifdef OPENDHT_PROXY_SERVER_IDENTITY
         resource->set_method_handler("SIGN", std::bind(&DhtProxyServer::putSigned, this, _1));
         resource->set_method_handler("ENCRYPT", std::bind(&DhtProxyServer::putEncrypted, this, _1));
 #endif // OPENDHT_PROXY_SERVER_IDENTITY
@@ -177,6 +182,7 @@ DhtProxyServer::stop()
         schedulerThread_.join();
     if (server_thread.joinable())
         server_thread.join();
+    threadPool_->stop();
 }
 
 void
@@ -187,7 +193,7 @@ DhtProxyServer::updateStats() const
     auto count = requestNum_.exchange(0);
     auto dt = std::chrono::duration<double>(now - last);
     stats_.requestRate = count / dt.count();
-#if OPENDHT_PUSH_NOTIFICATIONS
+#ifdef OPENDHT_PUSH_NOTIFICATIONS
     stats_.pushListenersCount = pushListeners_.size();
 #endif
     stats_.putCount = puts_.size();
@@ -219,10 +225,7 @@ DhtProxyServer::getNodeInfo(const Sp<restbed::Session>& session) const
                         result = nodeInfo_.toJson();
                     }
                     result["public_ip"] = s->get_origin(); // [ipv6:ipv4]:port or ipv4:port
-                    Json::StreamWriterBuilder wbuilder;
-                    wbuilder["commentStyle"] = "None";
-                    wbuilder["indentation"] = "";
-                    auto output = Json::writeString(wbuilder, result) + "\n";
+                    auto output = Json::writeString(jsonBuilder_, result) + "\n";
                     s->close(restbed::OK, output);
                 }
                 else
@@ -246,10 +249,7 @@ DhtProxyServer::getStats(const Sp<restbed::Session>& session) const
             try {
                 if (dht_) {
 #ifdef OPENDHT_JSONCPP
-                    Json::StreamWriterBuilder wbuilder;
-                    wbuilder["commentStyle"] = "None";
-                    wbuilder["indentation"] = "";
-                    auto output = Json::writeString(wbuilder, stats_.toJson()) + "\n";
+                    auto output = Json::writeString(jsonBuilder_, stats_.toJson()) + "\n";
                     s->close(restbed::OK, output);
 #else
                     s->close(restbed::NotFound, "{\"err\":\"JSON not enabled on this instance\"}");
@@ -281,13 +281,10 @@ DhtProxyServer::get(const Sp<restbed::Session>& session) const
                         infoHash = InfoHash::get(hash);
                     }
                     s->yield(restbed::OK, "", [=](const Sp<restbed::Session>&) {});
-                    dht_->get(infoHash, [s](const Sp<Value>& value) {
+                    dht_->get(infoHash, [this,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";
+                        auto output = Json::writeString(jsonBuilder_, value->toJson()) + "\n";
                         s->yield(output, [](const Sp<restbed::Session>& /*session*/){ });
                         return true;
                     }, [s](bool /*ok* */) {
@@ -335,19 +332,16 @@ 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, [this,cacheSession](const std::vector<Sp<Value>>& values, bool expired) {
                         auto s = cacheSession.lock();
                         if (!s) return false;
                         // Send values as soon as we get them
                         if (!s->is_closed()) {
-                            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";
+                                auto output = Json::writeString(jsonBuilder_, val) + "\n";
                                 s->yield(output, [](const Sp<restbed::Session>&){ });
                             }
                         }
@@ -367,7 +361,7 @@ DhtProxyServer::listen(const Sp<restbed::Session>& session)
     );
 }
 
-#if OPENDHT_PUSH_NOTIFICATIONS
+#ifdef OPENDHT_PUSH_NOTIFICATIONS
 
 struct DhtProxyServer::Listener {
     std::string clientId;
@@ -430,20 +424,14 @@ DhtProxyServer::subscribe(const std::shared_ptr<restbed::Session>& session)
                             scheduler_.edit(listener.expireNotifyJob, timeout - proxy::OP_MARGIN);
                             s->yield(restbed::OK);
 
-                            if (root.isMember("previous_values") &&
-                                root["previous_values"].asBool()) {
+                            if (!root.isMember("refresh") or !root["refresh"].asBool()) {
                                 dht_->get(
                                     infoHash,
                                     [this, 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";
+                                        auto output = Json::writeString(jsonBuilder_, value->toJson()) + "\n";
                                         s->yield(output, [](const Sp<restbed::Session>
                                                                 & /*session*/) {});
                                         return true;
@@ -473,12 +461,22 @@ DhtProxyServer::subscribe(const std::shared_ptr<restbed::Session>& session)
 
                     // The listener is not found, so add it.
                     listener.internalToken = dht_->listen(infoHash,
-                        [this, infoHash, pushToken, isAndroid, clientId](std::vector<std::shared_ptr<Value>> /*value*/) {
-                            // Build message content.
-                            Json::Value json;
-                            json["key"] = infoHash.toString();
-                            json["to"] = clientId;
-                            sendPushNotification(pushToken, json, isAndroid);
+                        [this, infoHash, pushToken, isAndroid, clientId](const std::vector<std::shared_ptr<Value>>& values, bool expired) {
+                            threadPool_->run([this, infoHash, pushToken, isAndroid, clientId, values, expired]() {
+                                // Build message content
+                                Json::Value json;
+                                json["key"] = infoHash.toString();
+                                json["to"] = clientId;
+                                if (expired and values.size() < 3) {
+                                    std::stringstream ss;
+                                    for(size_t i = 0; i < values.size(); ++i) {
+                                        if(i != 0) ss << ",";
+                                        ss << values[i]->id;
+                                    }
+                                    json["exp"] = ss.str();
+                                }
+                                sendPushNotification(pushToken, std::move(json), isAndroid);
+                            });
                             return true;
                         }
                     );
@@ -493,7 +491,7 @@ DhtProxyServer::subscribe(const std::shared_ptr<restbed::Session>& session)
                             Json::Value json;
                             json["timeout"] = infoHash.toString();
                             json["to"] = clientId;
-                            sendPushNotification(pushToken, json, isAndroid);
+                            sendPushNotification(pushToken, std::move(json), isAndroid);
                         }
                     );
                 }
@@ -571,7 +569,7 @@ DhtProxyServer::cancelPushListen(const std::string& pushToken, const dht::InfoHa
 }
 
 void
-DhtProxyServer::sendPushNotification(const std::string& token, const Json::Value& json, bool isAndroid) const
+DhtProxyServer::sendPushNotification(const std::string& token, Json::Value&& json, bool isAndroid) const
 {
     if (pushServer_.empty())
         return;
@@ -583,9 +581,9 @@ DhtProxyServer::sendPushNotification(const std::string& token, const Json::Value
     Json::Value notification(Json::objectValue);
     Json::Value tokens(Json::arrayValue);
     tokens[0] = token;
-    notification["tokens"] = tokens;
+    notification["tokens"] = std::move(tokens);
     notification["platform"] = isAndroid ? 2 : 1;
-    notification["data"] = json;
+    notification["data"] = std::move(json);
     notification["priority"] = "high";
     notification["time_to_live"] = 600;
 
@@ -593,7 +591,7 @@ DhtProxyServer::sendPushNotification(const std::string& token, const Json::Value
     notifications[0] = notification;
 
     Json::Value content;
-    content["notifications"] = notifications;
+    content["notifications"] = std::move(notifications);
 
     Json::StreamWriterBuilder wbuilder;
     wbuilder["commentStyle"] = "None";
@@ -684,7 +682,7 @@ DhtProxyServer::put(const std::shared_ptr<restbed::Session>& session)
                                         std::cout << "Permanent put expired: " << infoHash << " " << vid << std::endl;
                                         cancelPut(infoHash, vid);
                                     });
-#if OPENDHT_PUSH_NOTIFICATIONS
+#ifdef OPENDHT_PUSH_NOTIFICATIONS
                                     if (not pushToken.empty()) {
                                         pput.expireNotifyJob = scheduler_.add(timeout - proxy::OP_MARGIN,
                                             [this, infoHash, vid, pushToken, clientId, isAndroid]
@@ -694,7 +692,7 @@ DhtProxyServer::put(const std::shared_ptr<restbed::Session>& session)
                                             json["timeout"] = infoHash.toString();
                                             json["to"] = clientId;
                                             json["vid"] = std::to_string(vid);
-                                            sendPushNotification(pushToken, json, isAndroid);
+                                            sendPushNotification(pushToken, std::move(json), isAndroid);
                                         });
                                     }
 #endif
@@ -734,7 +732,7 @@ DhtProxyServer::put(const std::shared_ptr<restbed::Session>& session)
     );
 }
 
-#if OPENDHT_PROXY_SERVER_IDENTITY
+#ifdef OPENDHT_PROXY_SERVER_IDENTITY
 void
 DhtProxyServer::putSigned(const std::shared_ptr<restbed::Session>& session) const
 {
@@ -840,7 +838,7 @@ void
 DhtProxyServer::handleOptionsMethod(const std::shared_ptr<restbed::Session>& session) const
 {
     requestNum_++;
-#if OPENDHT_PROXY_SERVER_IDENTITY
+#ifdef OPENDHT_PROXY_SERVER_IDENTITY
     const auto allowed = "OPTIONS, GET, POST, LISTEN, SIGN, ENCRYPT";
 #else
     const auto allowed = "OPTIONS, GET, POST, LISTEN";
@@ -910,4 +908,3 @@ DhtProxyServer::removeClosedListeners(bool testSession)
 }
 
 }
-#endif //OPENDHT_PROXY_SERVER

src/dhtrunner.cpp

@@ -21,7 +21,7 @@
 #include "dhtrunner.h"
 #include "securedht.h"
 
-#if OPENDHT_PROXY_CLIENT
+#ifdef OPENDHT_PROXY_CLIENT
 #include "dht_proxy_client.h"
 #endif
 
@@ -55,7 +55,7 @@ struct DhtRunner::Listener {
 };
 
 DhtRunner::DhtRunner() : dht_()
-#if OPENDHT_PROXY_CLIENT
+#ifdef OPENDHT_PROXY_CLIENT
 , dht_via_proxy_()
 #endif //OPENDHT_PROXY_CLIENT
 {
@@ -105,7 +105,7 @@ DhtRunner::run(const SockAddr& local4, const SockAddr& local6, DhtRunner::Config
     auto dht = std::unique_ptr<DhtInterface>(new Dht(s4, s6, SecureDht::getConfig(config.dht_config)));
     dht_ = std::unique_ptr<SecureDht>(new SecureDht(std::move(dht), config.dht_config));
 
-#if OPENDHT_PROXY_CLIENT
+#ifdef OPENDHT_PROXY_CLIENT
     config_ = config;
 #endif
     enableProxy(not config.proxy_server.empty());
@@ -151,7 +151,7 @@ DhtRunner::run(const SockAddr& local4, const SockAddr& local6, DhtRunner::Config
 
 void
 DhtRunner::shutdown(ShutdownCallback cb) {
-#if OPENDHT_PROXY_CLIENT
+#ifdef OPENDHT_PROXY_CLIENT
     if (dht_via_proxy_)
         dht_via_proxy_->shutdown(cb);
 #endif
@@ -250,7 +250,7 @@ DhtRunner::setLoggers(LogMethod error, LogMethod warn, LogMethod debug) {
     std::lock_guard<std::mutex> lck(dht_mtx);
     if (dht_)
         dht_->setLoggers(error, warn, debug);
-#if OPENDHT_PROXY_CLIENT
+#ifdef OPENDHT_PROXY_CLIENT
     if (dht_via_proxy_)
         dht_via_proxy_->setLoggers(error, warn, debug);
 #endif
@@ -262,7 +262,7 @@ DhtRunner::setLogFilter(const InfoHash& f) {
     activeDht()->setLogFilter(f);
     if (dht_)
         dht_->setLogFilter(f);
-#if OPENDHT_PROXY_CLIENT
+#ifdef OPENDHT_PROXY_CLIENT
     if (dht_via_proxy_)
         dht_via_proxy_->setLogFilter(f);
 #endif
@@ -376,7 +376,7 @@ DhtRunner::registerCertificate(std::shared_ptr<crypto::Certificate> cert) {
 void
 DhtRunner::setLocalCertificateStore(CertificateStoreQuery&& query_method) {
     std::lock_guard<std::mutex> lck(dht_mtx);
-#if OPENDHT_PROXY_CLIENT
+#ifdef OPENDHT_PROXY_CLIENT
     if (dht_via_proxy_)
         dht_via_proxy_->setLocalCertificateStore(std::forward<CertificateStoreQuery>(query_method));
 #endif
@@ -658,7 +658,7 @@ DhtRunner::listen(InfoHash hash, ValueCallback vcb, Value::Filter f, Where w)
     {
         std::lock_guard<std::mutex> lck(storage_mtx);
         pending_ops.emplace([=](SecureDht& dht) mutable {
-#if OPENDHT_PROXY_CLIENT
+#ifdef OPENDHT_PROXY_CLIENT
             auto tokenbGlobal = listener_token_++;
             Listener listener {};
             listener.hash = hash;
@@ -697,7 +697,7 @@ DhtRunner::cancelListen(InfoHash h, size_t token)
 {
     {
         std::lock_guard<std::mutex> lck(storage_mtx);
-#if OPENDHT_PROXY_CLIENT
+#ifdef OPENDHT_PROXY_CLIENT
         pending_ops.emplace([=](SecureDht&) {
             auto it = listeners_.find(token);
             if (it == listeners_.end()) return;
@@ -721,7 +721,7 @@ DhtRunner::cancelListen(InfoHash h, std::shared_future<size_t> ftoken)
 {
     {
         std::lock_guard<std::mutex> lck(storage_mtx);
-#if OPENDHT_PROXY_CLIENT
+#ifdef OPENDHT_PROXY_CLIENT
         pending_ops.emplace([=](SecureDht&) {
             auto it = listeners_.find(ftoken.get());
             if (it == listeners_.end()) return;
@@ -973,7 +973,7 @@ DhtRunner::findCertificate(InfoHash hash, std::function<void(const std::shared_p
 void
 DhtRunner::resetDht()
 {
-#if OPENDHT_PROXY_CLIENT
+#ifdef OPENDHT_PROXY_CLIENT
     listeners_.clear();
     dht_via_proxy_.reset();
 #endif // OPENDHT_PROXY_CLIENT
@@ -983,7 +983,7 @@ DhtRunner::resetDht()
 SecureDht*
 DhtRunner::activeDht() const
 {
-#if OPENDHT_PROXY_CLIENT
+#ifdef OPENDHT_PROXY_CLIENT
     return use_proxy? dht_via_proxy_.get() : dht_.get();
 #else
     return dht_.get();
@@ -993,7 +993,7 @@ DhtRunner::activeDht() const
 void
 DhtRunner::setProxyServer(const std::string& proxy, const std::string& pushNodeId)
 {
-#if OPENDHT_PROXY_CLIENT
+#ifdef OPENDHT_PROXY_CLIENT
     if (config_.proxy_server == proxy and config_.push_node_id == pushNodeId)
         return;
     config_.proxy_server = proxy;
@@ -1008,7 +1008,7 @@ DhtRunner::setProxyServer(const std::string& proxy, const std::string& pushNodeI
 void
 DhtRunner::enableProxy(bool proxify)
 {
-#if OPENDHT_PROXY_CLIENT
+#ifdef OPENDHT_PROXY_CLIENT
     if (dht_via_proxy_) {
         dht_via_proxy_->shutdown({});
     }
@@ -1026,7 +1026,7 @@ DhtRunner::enableProxy(bool proxify)
             }, config_.proxy_server, config_.push_node_id)
         );
         dht_via_proxy_ = std::unique_ptr<SecureDht>(new SecureDht(std::move(dht_via_proxy), config_.dht_config));
-#if OPENDHT_PUSH_NOTIFICATIONS
+#ifdef OPENDHT_PUSH_NOTIFICATIONS
         if (not pushToken_.empty())
             dht_via_proxy_->setPushNotificationToken(pushToken_);
 #endif
@@ -1059,8 +1059,8 @@ DhtRunner::enableProxy(bool proxify)
 void
 DhtRunner::forwardAllMessages(bool forward)
 {
-#if OPENDHT_PROXY_SERVER
-#if OPENDHT_PROXY_CLIENT
+#ifdef OPENDHT_PROXY_SERVER
+#ifdef OPENDHT_PROXY_CLIENT
     if (dht_via_proxy_)
         dht_via_proxy_->forwardAllMessages(forward);
 #endif // OPENDHT_PROXY_CLIENT
@@ -1074,7 +1074,7 @@ DhtRunner::forwardAllMessages(bool forward)
  */
 void
 DhtRunner::setPushNotificationToken(const std::string& token) {
-#if OPENDHT_PROXY_CLIENT && OPENDHT_PUSH_NOTIFICATIONS
+#if defined(OPENDHT_PROXY_CLIENT) && defined(OPENDHT_PUSH_NOTIFICATIONS)
     pushToken_ = token;
     if (dht_via_proxy_)
         dht_via_proxy_->setPushNotificationToken(token);
@@ -1084,7 +1084,7 @@ DhtRunner::setPushNotificationToken(const std::string& token) {
 void
 DhtRunner::pushNotificationReceived(const std::map<std::string, std::string>& data)
 {
-#if OPENDHT_PROXY_CLIENT && OPENDHT_PUSH_NOTIFICATIONS
+#if defined(OPENDHT_PROXY_CLIENT) && defined(OPENDHT_PUSH_NOTIFICATIONS)
     {
         std::lock_guard<std::mutex> lck(storage_mtx);
         pending_ops_prio.emplace([=](SecureDht&) {

src/op_cache.cpp

@@ -34,6 +34,7 @@ OpValueCache::onValuesAdded(const std::vector<Sp<Value>>& vals) {
     }
     return newValues.empty() ? true : callback(newValues, false);
 }
+
 bool
 OpValueCache::onValuesExpired(const std::vector<Sp<Value>>& vals) {
     std::vector<Sp<Value>> expiredValues;
@@ -49,6 +50,24 @@ OpValueCache::onValuesExpired(const std::vector<Sp<Value>>& vals) {
     }
     return expiredValues.empty() ? true : callback(expiredValues, true);
 }
+
+bool
+OpValueCache::onValuesExpired(const std::vector<Value::Id>& vids)
+{
+    std::vector<Sp<Value>> expiredValues;
+    for (const auto& vid : vids) {
+        auto vit = values.find(vid);
+        if (vit != values.end()) {
+            vit->second.refCount--;
+            if (not vit->second.refCount) {
+                expiredValues.emplace_back(std::move(vit->second.data));
+                values.erase(vit);
+            }
+        }
+    }
+    return expiredValues.empty() ? true : callback(expiredValues, true);
+}
+
 std::vector<Sp<Value>>
 OpValueCache::get(const Value::Filter& filter) const {
     std::vector<Sp<Value>> ret;

src/op_cache.h

@@ -32,8 +32,8 @@ struct OpCacheValueStorage
 
 class OpValueCache {
 public:
-    OpValueCache(ValueCallback&& cb) : callback(std::forward<ValueCallback>(cb)) {}
-    OpValueCache(OpValueCache&& o) : values(std::move(o.values)), callback(std::move(o.callback)) {
+    OpValueCache(ValueCallback&& cb) noexcept : callback(std::forward<ValueCallback>(cb)) {}
+    OpValueCache(OpValueCache&& o) noexcept : values(std::move(o.values)), callback(std::move(o.callback)) {
         o.callback = {};
     }
 
@@ -56,6 +56,7 @@ public:
 
     bool onValuesAdded(const std::vector<Sp<Value>>& vals);
     bool onValuesExpired(const std::vector<Sp<Value>>& vals);
+    bool onValuesExpired(const std::vector<Value::Id>& vals);
 
     void onNodeChanged(ListenSyncStatus status) {
         switch (status) {

src/securedht.cpp

@@ -235,7 +235,7 @@ SecureDht::checkValue(const Sp<Value>& v)
     // Decrypt encrypted values
     if (v->isEncrypted()) {
         if (not key_) {
-#if OPENDHT_PROXY_SERVER
+#ifdef OPENDHT_PROXY_SERVER
             if (forward_all_) // We are currently a proxy, send messages to clients.
                 return v;
 #endif

src/storage.h

@@ -90,20 +90,7 @@ struct Storage {
 
     Storage() {}
     Storage(time_point t) : maintenance_time(t) {}
-
-#if defined(__GNUC__) && __GNUC__ == 4 && __GNUC_MINOR__ <= 9 || defined(_WIN32)
-    // GCC-bug: remove me when support of GCC < 4.9.2 is abandoned
-    Storage(Storage&& o) noexcept
-        : maintenance_time(std::move(o.maintenance_time))
-        , listeners(std::move(o.listeners))
-        , local_listeners(std::move(o.local_listeners))
-        , listener_token(std::move(o.listener_token))
-        , values(std::move(o.values))
-        , total_size(std::move(o.total_size)) {}
-#else
     Storage(Storage&& o) noexcept = default;
-#endif
-
     Storage& operator=(Storage&& o) = default;
 
     bool empty() const {

src/thread_pool.cpp

@@ -0,0 +1,134 @@
+/*
+ *  Copyright (C) 2016-2019 Savoir-faire Linux Inc.
+ *
+ *  Author: Adrien Béraud <adrien.beraud@savoirfairelinux.com>
+ *
+ *  This program is free software; you can redistribute it and/or modify
+ *  it under the terms of the GNU General Public License as published by
+ *  the Free Software Foundation; either version 3 of the License, or
+ *  (at your option) any later version.
+ *
+ *  This program is distributed in the hope that it will be useful,
+ *  but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *  GNU General Public License for more details.
+ *
+ *  You should have received a copy of the GNU General Public License
+ *  along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#include "thread_pool.h"
+
+#include <atomic>
+#include <thread>
+
+#include <ciso646> // fix windows compiler bug
+
+namespace dht {
+
+constexpr const size_t IO_THREADS_MAX {64};
+
+struct ThreadPool::ThreadState
+{
+    std::thread thread {};
+    std::atomic_bool run {true};
+};
+
+ThreadPool&
+ThreadPool::computation()
+{
+    static ThreadPool pool;
+    return pool;
+}
+
+ThreadPool&
+ThreadPool::io()
+{
+    static ThreadPool pool(IO_THREADS_MAX);
+    return pool;
+}
+
+
+ThreadPool::ThreadPool(size_t maxThreads) : maxThreads_(maxThreads)
+{
+    threads_.reserve(maxThreads_);
+}
+
+ThreadPool::ThreadPool()
+ : ThreadPool(std::max<size_t>(std::thread::hardware_concurrency(), 4))
+{}
+
+ThreadPool::~ThreadPool()
+{
+    join();
+}
+
+void
+ThreadPool::run(std::function<void()>&& cb)
+{
+    std::unique_lock<std::mutex> l(lock_);
+    if (not running_) return;
+
+    // launch new thread if necessary
+    if (not readyThreads_ && threads_.size() < maxThreads_) {
+        threads_.emplace_back(new ThreadState());
+        auto& t = *threads_.back();
+        t.thread = std::thread([&]() {
+            while (t.run) {
+                std::function<void()> task;
+
+                // pick task from queue
+                {
+                    std::unique_lock<std::mutex> l(lock_);
+                    readyThreads_++;
+                    cv_.wait(l, [&](){
+                        return not t.run or not tasks_.empty();
+                    });
+                    readyThreads_--;
+                    if (not t.run)
+                        break;
+                    task = std::move(tasks_.front());
+                    tasks_.pop();
+                }
+
+                // run task
+                try {
+                    if (task)
+                        task();
+                } catch (const std::exception& e) {
+                    // LOG_ERR("Exception running task: %s", e.what());
+                }
+            }
+        });
+    }
+
+    // push task to queue
+    tasks_.emplace(std::move(cb));
+
+    // notify thread
+    l.unlock();
+    cv_.notify_one();
+}
+
+void
+ThreadPool::stop()
+{
+    {
+        std::lock_guard<std::mutex> l(lock_);
+        running_ = false;
+    }
+    for (auto& t : threads_)
+        t->run = false;
+    cv_.notify_all();
+}
+
+void
+ThreadPool::join()
+{
+    stop();
+    for (auto& t : threads_)
+        t->thread.join();
+    threads_.clear();
+}
+
+}

src/thread_pool.h

@@ -0,0 +1,69 @@
+/*
+ *  Copyright (C) 2016-2019 Savoir-faire Linux Inc.
+ *
+ *  Author: Adrien Béraud <adrien.beraud@savoirfairelinux.com>
+ *
+ *  This program is free software; you can redistribute it and/or modify
+ *  it under the terms of the GNU General Public License as published by
+ *  the Free Software Foundation; either version 3 of the License, or
+ *  (at your option) any later version.
+ *
+ *  This program is distributed in the hope that it will be useful,
+ *  but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *  GNU General Public License for more details.
+ *
+ *  You should have received a copy of the GNU General Public License
+ *  along with this program. If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#pragma once
+
+#include <condition_variable>
+#include <vector>
+#include <queue>
+#include <future>
+#include <functional>
+
+namespace dht {
+
+class ThreadPool {
+public:
+    static ThreadPool& computation();
+    static ThreadPool& io();
+
+    ThreadPool();
+    ThreadPool(size_t maxThreads);
+    ~ThreadPool();
+
+    void run(std::function<void()>&& cb);
+
+    template<class T>
+    std::future<T> get(std::function<T()>&& cb) {
+        auto ret = std::make_shared<std::promise<T>>();
+        run(std::bind([=](std::function<T()>& mcb) mutable {
+                ret->set_value(mcb());
+            }, std::move(cb)));
+        return ret->get_future();
+    }
+    template<class T>
+    std::shared_ptr<std::future<T>> getShared(std::function<T()>&& cb) {
+        return std::make_shared<std::future<T>>(get(std::move(cb)));
+    }
+
+    void stop();
+    void join();
+
+private:
+    struct ThreadState;
+    std::queue<std::function<void()>> tasks_ {};
+    std::vector<std::unique_ptr<ThreadState>> threads_;
+    unsigned readyThreads_ {0};
+    std::mutex lock_ {};
+    std::condition_variable cv_ {};
+
+    const unsigned maxThreads_;
+    bool running_ {true};
+};
+
+}

src/value_cache.h

@@ -34,7 +34,7 @@ public:
         if (syncCallback)
             syncCallback(ListenSyncStatus::ADDED);
     }
-    ValueCache(ValueCache&& o) : values(std::move(o.values)), callback(std::move(o.callback)), syncCallback(std::move(o.syncCallback)) {
+    ValueCache(ValueCache&& o) noexcept : values(std::move(o.values)), callback(std::move(o.callback)), syncCallback(std::move(o.syncCallback)) {
         o.callback = {};
         o.syncCallback = {};
     }

tools/dhtchat.cpp

@@ -76,7 +76,7 @@ main(int argc, char **argv)
         if (not params.bootstrap.first.empty())
             dht.bootstrap(params.bootstrap.first.c_str(), params.bootstrap.second.c_str());
 
-#if OPENDHT_PROXY_CLIENT
+#ifdef OPENDHT_PROXY_CLIENT
     if (!params.proxyclient.empty()) {
         dht.setProxyServer(params.proxyclient);
         dht.enableProxy(true);

tools/dhtnode.cpp

@@ -58,9 +58,9 @@ void print_help() {
               << "  ld [key]   Print basic information about currenty stored values on this node (or key)." << std::endl
               << "  lr         Print the full current routing table of this node." << std::endl;
 
-#if OPENDHT_PROXY_SERVER
+#ifdef OPENDHT_PROXY_SERVER
     std::cout << std::endl << "Operations with the proxy:" << std::endl
-#if OPENDHT_PUSH_NOTIFICATIONS
+#ifdef OPENDHT_PUSH_NOTIFICATIONS
               << "  pst [port] <pushServer> Start the proxy interface on port." << std::endl
 #else
               << "  pst [port]              Start the proxy interface on port." << std::endl
@@ -68,9 +68,9 @@ void print_help() {
               << "  psp [port]              Stop the proxy interface on port." << std::endl;
 #endif //OPENDHT_PROXY_SERVER
 
-#if OPENDHT_PROXY_CLIENT
+#ifdef OPENDHT_PROXY_CLIENT
     std::cout << std::endl << "Operations with the proxy:" << std::endl
-#if OPENDHT_PUSH_NOTIFICATIONS
+#ifdef OPENDHT_PUSH_NOTIFICATIONS
               << "  stt [server_address] <device_key> Start the proxy client." << std::endl
               << "  rs  [token]                       Resubscribe to opendht." << std::endl
               << "  rp  [token]                       Inject a push notification in Opendht." << std::endl
@@ -102,7 +102,7 @@ void print_help() {
 }
 
 void cmd_loop(std::shared_ptr<DhtRunner>& dht, dht_params& params
-#if OPENDHT_PROXY_SERVER
+#ifdef OPENDHT_PROXY_SERVER
     , std::map<in_port_t, std::unique_ptr<DhtProxyServer>>& proxies
 #endif
 )
@@ -141,7 +141,7 @@ void cmd_loop(std::shared_ptr<DhtRunner>& dht, dht_params& params
             std::cout << dht->getNodesStats(AF_INET).toString() << std::endl;
             std::cout << "IPv6 stats:" << std::endl;
             std::cout << dht->getNodesStats(AF_INET6).toString() << std::endl;
-#if OPENDHT_PROXY_SERVER
+#ifdef 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;
@@ -201,16 +201,16 @@ void cmd_loop(std::shared_ptr<DhtRunner>& dht, dht_params& params
             dht->connectivityChanged();
             continue;
         }
-#if OPENDHT_PROXY_SERVER
+#ifdef OPENDHT_PROXY_SERVER
         else if (op == "pst") {
-#if OPENDHT_PUSH_NOTIFICATIONS
+#ifdef OPENDHT_PUSH_NOTIFICATIONS
                 iss >> idstr >> pushServer;
 #else
                 iss >> idstr;
 #endif // OPENDHT_PUSH_NOTIFICATIONS
             try {
                 unsigned int port = std::stoi(idstr);
-#if OPENDHT_PUSH_NOTIFICATIONS
+#ifdef OPENDHT_PUSH_NOTIFICATIONS
                 proxies.emplace(port, std::unique_ptr<DhtProxyServer>(new DhtProxyServer(dht, port, pushServer)));
 #else
                 proxies.emplace(port, std::unique_ptr<DhtProxyServer>(new DhtProxyServer(dht, port)));
@@ -227,7 +227,7 @@ void cmd_loop(std::shared_ptr<DhtRunner>& dht, dht_params& params
             continue;
         }
 #endif //OPENDHT_PROXY_SERVER
-#if OPENDHT_PROXY_CLIENT
+#ifdef OPENDHT_PROXY_CLIENT
         else if (op == "stt") {
             dht->enableProxy(true);
             continue;
@@ -235,7 +235,7 @@ void cmd_loop(std::shared_ptr<DhtRunner>& dht, dht_params& params
             dht->enableProxy(false);
             continue;
         }
-#if OPENDHT_PUSH_NOTIFICATIONS
+#ifdef OPENDHT_PUSH_NOTIFICATIONS
         else if (op == "rp") {
             iss >> value;
             dht->pushNotificationReceived({{"to", "dhtnode"}, {"token", value}});
@@ -500,6 +500,7 @@ main(int argc, char **argv)
         dht::DhtRunner::Config config {};
         config.dht_config.node_config.network = params.network;
         config.dht_config.node_config.maintain_storage = false;
+        config.dht_config.node_config.persist_path = params.persist_path;
         config.dht_config.id = crt;
         config.threaded = true;
         config.proxy_server = params.proxyclient;
@@ -523,11 +524,11 @@ main(int argc, char **argv)
             dht->bootstrap(params.bootstrap.first.c_str(), params.bootstrap.second.c_str());
         }
 
-#if OPENDHT_PROXY_SERVER
+#ifdef OPENDHT_PROXY_SERVER
         std::map<in_port_t, std::unique_ptr<DhtProxyServer>> proxies;
 #endif
         if (params.proxyserver != 0) {
-#if OPENDHT_PROXY_SERVER
+#ifdef OPENDHT_PROXY_SERVER
             proxies.emplace(params.proxyserver, std::unique_ptr<DhtProxyServer>(new DhtProxyServer(dht, params.proxyserver, params.pushserver)));
 #else
             std::cerr << "DHT proxy server requested but OpenDHT built without proxy server support." << std::endl;
@@ -539,7 +540,7 @@ main(int argc, char **argv)
             while (runner.wait());
         else
             cmd_loop(dht, params
-#if OPENDHT_PROXY_SERVER
+#ifdef OPENDHT_PROXY_SERVER
                 , proxies
 #endif
             );

tools/tools_common.h

@@ -103,6 +103,7 @@ struct dht_params {
     std::string proxyclient {};
     std::string pushserver {};
     std::string devicekey {};
+    std::string persist_path {};
 };
 
 static const constexpr struct option long_options[] = {
@@ -114,6 +115,7 @@ static const constexpr struct option long_options[] = {
    {"verbose",    no_argument      , nullptr, 'v'},
    {"daemonize",  no_argument      , nullptr, 'd'},
    {"service",    no_argument      , nullptr, 's'},
+   {"persist",    required_argument, nullptr, 'f'},
    {"logfile",    required_argument, nullptr, 'l'},
    {"syslog",     no_argument      , nullptr, 'L'},
    {"proxyserver",required_argument, nullptr, 'S'},
@@ -127,7 +129,7 @@ dht_params
 parseArgs(int argc, char **argv) {
     dht_params params;
     int opt;
-    while ((opt = getopt_long(argc, argv, "hidsvp:n:b:l:", long_options, nullptr)) != -1) {
+    while ((opt = getopt_long(argc, argv, "hidsvp:n:b:f:l:", long_options, nullptr)) != -1) {
         switch (opt) {
         case 'p': {
                 int port_arg = atoi(optarg);
@@ -154,6 +156,9 @@ parseArgs(int argc, char **argv) {
         case 'D':
             params.devicekey = optarg;
             break;
+        case 'f':
+            params.persist_path = optarg;
+            break;
         case 'n':
             params.network = strtoul(optarg, nullptr, 0);
             break;