dht: move socket to SearchNode

.travis.yml

@@ -16,13 +16,12 @@ env:
 
 before_install:
   - |
-    # non llvm builds
-    if [[ "$OPENDHT_TEST_JOB" != *"opendht.llvm"* ]]; then
+    if [[ "$OPENDHT_TEST_JOB" == *"opendht.classic"* ]]; then
+      docker pull aberaud/opendht-deps-bionic;
+    fi
+    if [[ "$OPENDHT_TEST_JOB" == *"opendht.proxyclient"* ]] || [[ "$OPENDHT_TEST_JOB" == *"opendht.proxyserver"* ]] || [[ "$OPENDHT_TEST_JOB" == *"opendht.push"* ]]; then
       docker pull aberaud/opendht-deps;
     fi
-
-  - |
-    # classic build
     if [[ "$OPENDHT_TEST_JOB" == *"opendht.llvm"* ]]; then
       docker pull aberaud/opendht-deps-llvm
     fi

CMakeLists.txt

@@ -4,7 +4,7 @@ include(CMakeDependentOption)
 
 project (opendht)
 set (opendht_VERSION_MAJOR 2)
-set (opendht_VERSION_MINOR 1.0)
+set (opendht_VERSION_MINOR 1.3)
 set (opendht_VERSION ${opendht_VERSION_MAJOR}.${opendht_VERSION_MINOR})
 set (PACKAGE_VERSION ${opendht_VERSION})
 set (VERSION "${opendht_VERSION}")

configure.ac

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

docker/DockerfileDeps

@@ -1,33 +1,20 @@
-FROM ubuntu:16.04
+FROM ubuntu:20.04
 MAINTAINER Adrien Béraud <adrien.beraud@savoirfairelinux.com>
+
 RUN apt-get update && apt-get install -y \
-        build-essential cmake git wget libncurses5-dev libreadline-dev nettle-dev \
-        libgnutls28-dev libuv1-dev cython3 python3-dev python3-setuptools libcppunit-dev libjsoncpp-dev \
-        autotools-dev autoconf libssl-dev \
+        dialog apt-utils \
+    && apt-get clean \
+    && echo 'debconf debconf/frontend select Noninteractive' | debconf-set-selections
+
+RUN apt-get update && apt-get install -y \
+        build-essential pkg-config cmake git wget \
+        autotools-dev autoconf \
+        cython3 python3-dev python3-setuptools \
+        libncurses5-dev libreadline-dev nettle-dev libcppunit-dev \
+        libgnutls28-dev libuv1-dev libjsoncpp-dev libargon2-dev \
+        libssl-dev libfmt-dev libhttp-parser-dev libasio-dev libmsgpack-dev \
     && apt-get clean
 
-RUN echo "*** Installing libfmt ***" \
-    && wget https://github.com/fmtlib/fmt/archive/5.3.0.tar.gz \
-    && tar -xzf 5.3.0.tar.gz && cd fmt-5.3.0/ \
-    && cmake -DCMAKE_INSTALL_PREFIX=/usr -DBUILD_SHARED_LIBS=On -DFMT_TEST=Off -DFMT_DOC=Off . \
-    && make -j8 && make install \
-    && cd ../ && rm -rf fmt*
-
-# libhttp-parser-dev is 2.1-2 which is too old
-RUN echo "*** Building nodejs/http_parser dependency ***" \
-    && wget https://github.com/nodejs/http-parser/archive/v2.9.2.tar.gz \
-    && tar xvf v2.9.2.tar.gz && cd http-parser-2.9.2/ \
-    && PREFIX=/usr make -j8 install \
-    && cd ../ && rm -rf http-parser*
-
-# libasio-dev (1.10.6-3) is too old
-RUN echo "** Building a recent version of asio ***" \
-    && wget https://github.com/aberaud/asio/archive/a7d66ef4017d8f1b7f2cef1bb4ba8e23b0961571.tar.gz \
-    && tar -xvf a7d66ef4017d8f1b7f2cef1bb4ba8e23b0961571.tar.gz && cd asio-a7d66ef4017d8f1b7f2cef1bb4ba8e23b0961571/asio \
-    && ./autogen.sh && ./configure --prefix=/usr --without-boost --disable-examples --disable-tests  \
-    && make install \
-    && cd ../../ && rm -rf asio*
-
 RUN echo "*** Downloading RESTinio ***" \
     && mkdir restinio && cd restinio \
     && wget https://github.com/aberaud/restinio/archive/a7a10e419d9089c5b8ee63f5e3098c892f22fae4.tar.gz \
@@ -38,19 +25,3 @@ RUN echo "*** Downloading RESTinio ***" \
              -DRESTINIO_FIND_DEPS=ON -DRESTINIO_ALLOW_SOBJECTIZER=Off -DRESTINIO_USE_BOOST_ASIO=none . \
     && make -j8 && make install \
     && cd ../../ && rm -rf restinio*
-
-
-# build restbed from sources
-RUN git clone --recursive https://github.com/corvusoft/restbed.git \
-	&& cd restbed && mkdir build && cd build \
-	&& cmake -DBUILD_TESTS=NO -DBUILD_EXAMPLES=NO -DBUILD_SSL=NO -DBUILD_SHARED=YES -DCMAKE_INSTALL_PREFIX=/usr -DCMAKE_INSTALL_LIBDIR=lib .. \
-	&& make -j8 install \
-	&& cd .. && rm -rf restbed
-
-#build msgpack from source
-RUN wget https://github.com/msgpack/msgpack-c/releases/download/cpp-2.1.5/msgpack-2.1.5.tar.gz \
-    && tar -xzf msgpack-2.1.5.tar.gz \
-    && cd msgpack-2.1.5 && mkdir build && cd build \
-    && cmake -DMSGPACK_CXX11=ON -DMSGPACK_BUILD_EXAMPLES=OFF -DCMAKE_INSTALL_PREFIX=/usr .. \
-    && make -j8 && make install \
-    && cd ../.. && rm -rf msgpack-2.1.5 msgpack-2.1.5.tar.gz

docker/DockerfileDepsBionic

@@ -3,11 +3,13 @@ MAINTAINER Adrien Béraud <adrien.beraud@savoirfairelinux.com>
 
 RUN echo "APT::Acquire::Retries \"3\";" > /etc/apt/apt.conf.d/80-retries
 RUN apt-get update && apt-get install -y \
-        apt-transport-https build-essential cmake git wget libncurses5-dev libreadline-dev nettle-dev \
+        apt-transport-https build-essential pkg-config git wget libncurses5-dev libreadline-dev nettle-dev \
         libgnutls28-dev libuv1-dev cython3 python3-dev python3-setuptools libcppunit-dev libjsoncpp-dev \
         autotools-dev autoconf libfmt-dev libhttp-parser-dev libmsgpack-dev libssl-dev \
     && apt-get clean
 
+RUN apt-get update && apt-get install -y python3-pip && pip3 install --upgrade cmake
+
 # libasio-dev (1.10) is too old
 RUN echo "** Building a recent version of asio ***" \
     && wget https://github.com/aberaud/asio/archive/a7d66ef4017d8f1b7f2cef1bb4ba8e23b0961571.tar.gz \

docker/DockerfileDepsLlvm

@@ -1,40 +1,21 @@
-FROM ubuntu:16.04
+FROM ubuntu:20.04
 MAINTAINER Adrien Béraud <adrien.beraud@savoirfairelinux.com>
+RUN apt-get update && apt-get install -y \
+        dialog apt-utils \
+    && apt-get clean \
+    && echo 'debconf debconf/frontend select Noninteractive' | debconf-set-selections
+
 RUN apt-get update \
-    && apt-get install -y llvm llvm-dev clang make cmake git wget libncurses5-dev libreadline-dev \
+    && apt-get install -y llvm llvm-dev clang make cmake pkg-config git wget libncurses5-dev libreadline-dev \
        nettle-dev libgnutls28-dev libuv1-dev libmsgpack-dev libjsoncpp-dev cython3 python3-dev \
        python3-setuptools libcppunit-dev python3-pip \
-       autotools-dev autoconf libssl-dev \
+       autotools-dev autoconf libssl-dev libargon2-dev \
+       libfmt-dev libhttp-parser-dev libasio-dev \
     && apt-get remove -y gcc g++ && apt-get autoremove -y && apt-get clean
 
 ENV CC cc
 ENV CXX c++
 
-RUN echo "*** Installing libfmt ***" \
-    && wget https://github.com/fmtlib/fmt/archive/5.3.0.tar.gz \
-    && tar -xzf 5.3.0.tar.gz && cd fmt-5.3.0/ \
-    && cmake -DCMAKE_INSTALL_PREFIX=/usr -DBUILD_SHARED_LIBS=On -DFMT_TEST=Off -DFMT_DOC=Off . \
-    && make -j8 && make install \
-    && cd ../ && rm -rf fmt*
-
-# libhttp-parser-dev is 2.1-2 which is too old
-RUN echo "*** Building nodejs/http_parser dependency ***" \
-    && wget https://github.com/nodejs/http-parser/archive/v2.9.2.tar.gz \
-    && tar xvf v2.9.2.tar.gz && cd http-parser-2.9.2/ \
-    && PREFIX=/usr make -j8 install \
-    && cd ../ && rm -rf http-parser*
-
-# libasio-dev (1.10.6-3) is too old
-RUN echo "** Building a recent version of asio ***" \
-    && wget https://github.com/aberaud/asio/archive/a7d66ef4017d8f1b7f2cef1bb4ba8e23b0961571.tar.gz \
-    && tar -xvf a7d66ef4017d8f1b7f2cef1bb4ba8e23b0961571.tar.gz && cd asio-a7d66ef4017d8f1b7f2cef1bb4ba8e23b0961571/asio \
-    && ./autogen.sh && ./configure --prefix=/usr --without-boost --disable-examples --disable-tests  \
-    && make install \
-    && cd ../../ && rm -rf asio*
-
-#patch for https://github.com/Stiffstream/restinio-conan-example/issues/2
-RUN apt-get update && apt-get install -y python3-pip && pip3 install --upgrade cmake
-
 RUN echo "*** Downloading RESTinio ***" \
     && mkdir restinio && cd restinio \
     && wget https://github.com/aberaud/restinio/archive/a7a10e419d9089c5b8ee63f5e3098c892f22fae4.tar.gz \
@@ -45,18 +26,3 @@ RUN echo "*** Downloading RESTinio ***" \
              -DRESTINIO_FIND_DEPS=ON -DRESTINIO_ALLOW_SOBJECTIZER=Off -DRESTINIO_USE_BOOST_ASIO=none . \
     && make -j8 && make install \
     && cd ../../ && rm -rf restinio*
-
-# build restbed from sources
-RUN git clone --recursive https://github.com/corvusoft/restbed.git \
-	&& cd restbed && mkdir build && cd build \
-	&& cmake -DBUILD_TESTS=NO -DBUILD_EXAMPLES=NO -DBUILD_SSL=NO -DBUILD_SHARED=YES -DCMAKE_INSTALL_PREFIX=/usr -DCMAKE_INSTALL_LIBDIR=lib .. \
-	&& make -j8 install \
-	&& cd .. && rm -rf restbed
-
-#build msgpack from source
-RUN wget https://github.com/msgpack/msgpack-c/releases/download/cpp-2.1.5/msgpack-2.1.5.tar.gz \
-    && tar -xzf msgpack-2.1.5.tar.gz \
-    && cd msgpack-2.1.5 && mkdir build && cd build \
-    && cmake -DMSGPACK_CXX11=ON -DMSGPACK_BUILD_EXAMPLES=OFF -DCMAKE_INSTALL_PREFIX=/usr .. \
-    && make -j8 && make install \
-    && cd ../.. && rm -rf msgpack-2.1.5 msgpack-2.1.5.tar.gz

docker/DockerfileLlvm

@@ -0,0 +1,6 @@
+FROM aberaud/opendht-deps-llvm
+MAINTAINER Adrien Béraud <adrien.beraud@savoirfairelinux.com>
+RUN git clone https://github.com/savoirfairelinux/opendht.git \
+	&& cd opendht && mkdir build && cd build \
+	&& cmake .. -DCMAKE_INSTALL_PREFIX=/usr -DOPENDHT_PYTHON=On -DOPENDHT_LTO=On && make -j8 && make install \
+	&& cd ../.. && rm -rf opendht

docker/DockerfileTravis

@@ -1,6 +1,8 @@
-FROM aberaud/opendht-deps
+FROM aberaud/opendht-deps-bionic
 MAINTAINER Adrien Béraud <adrien.beraud@savoirfairelinux.com>
 
+RUN apt-get update && apt-get install -y python3-pip && pip3 install --upgrade cmake
+
 COPY . /root/opendht
 RUN cd /root/opendht && mkdir build && cd build \
 	&& cmake -DCMAKE_INSTALL_PREFIX=/usr -DOPENDHT_PYTHON=On -DOPENDHT_C=On -DOPENDHT_LTO=On -DOPENDHT_TESTS=ON ..  \

include/opendht/dht.h

@@ -371,6 +371,8 @@ private:
     Dht(const Dht&) = delete;
     Dht& operator=(const Dht&) = delete;
 
+    std::mt19937_64 rd {crypto::getSeededRandomEngine<std::mt19937_64>()};
+
     InfoHash myid {};
 
     uint64_t secret {};
@@ -412,7 +414,6 @@ private:
     std::map<size_t, std::tuple<size_t, size_t, size_t>> listeners {};
     size_t listener_token {1};
 
-    std::mt19937_64 rd {crypto::getSeededRandomEngine<std::mt19937_64>()};
 
     // timing
     Scheduler scheduler;

include/opendht/dht_proxy_server.h

@@ -98,8 +98,10 @@ public:
     struct ServerStats {
         /** Current number of listen operations */
         size_t listenCount {0};
-        /** Current number of permanent put operations */
+        /** Current number of permanent put operations (hash used) */
         size_t putCount {0};
+        /** Current number of permanent put values */
+        size_t totalPermanentPuts {0};
         /** Current number of push tokens with at least one listen operation */
         size_t pushListenersCount {0};
         /** Average requests per second */
@@ -129,6 +131,7 @@ public:
             Json::Value result;
             result["listenCount"] = static_cast<Json::UInt64>(listenCount);
             result["putCount"] = static_cast<Json::UInt64>(putCount);
+            result["totalPermanentPuts"] = static_cast<Json::UInt64>(totalPermanentPuts);
             result["pushListenersCount"] = static_cast<Json::UInt64>(pushListenersCount);
             result["requestRate"] = requestRate;
             if (nodeInfo)

include/opendht/dhtrunner.h

@@ -389,7 +389,7 @@ public:
      * @param threaded: If false, loop() must be called periodically. Otherwise a thread is launched.
      * @param cb: Optional callback to receive general state information.
      */
-    void run(const SockAddr& local4, const SockAddr& local6, const Config& config, Context&& context = {});
+    void run(SockAddr& local4, SockAddr& local6, const Config& config, Context&& context = {});
 
     /**
      * Same as @run(sockaddr_in, sockaddr_in6, Identity, bool, StatusCallback), but with string IP addresses and service (port).

include/opendht/infohash.h

@@ -237,6 +237,9 @@ public:
 
     static Hash getRandom();
 
+    template <typename Rd>
+    static Hash getRandom(Rd&);
+
     template <size_t M>
     OPENDHT_PUBLIC friend std::ostream& operator<< (std::ostream& s, const Hash<M>& h);
 
@@ -324,6 +327,19 @@ Hash<N>::getRandom()
     return h;
 }
 
+template <size_t N>
+template <typename Rd>
+Hash<N>
+Hash<N>::getRandom(Rd& rdev)
+{
+    Hash h;
+    std::uniform_int_distribution<uint32_t> rand_int;
+    auto a = reinterpret_cast<uint32_t*>(h.data());
+    auto b = reinterpret_cast<uint32_t*>(h.data() + h.size());
+    std::generate(a, b, std::bind(rand_int, std::ref(rdev)));
+    return h;
+}
+
 struct HexMap : public std::array<std::array<char, 2>, 256> {
     HexMap() {
         for (size_t i=0; i<size(); i++) {

include/opendht/network_engine.h

@@ -211,6 +211,7 @@ private:
 
 public:
     using RequestCb = std::function<void(const Request&, RequestAnswer&&)>;
+    using RequestErrorCb = std::function<bool(const Request&, DhtProtocolException&&)>;
     using RequestExpiredCb = std::function<void(const Request&, bool)>;
 
     NetworkEngine(const Sp<Logger>& log, std::mt19937_64& rd, Scheduler& scheduler, std::unique_ptr<DatagramSocket>&& sock);
@@ -361,10 +362,9 @@ public:
                            const InfoHash& hash,
                            const Query& query,
                            const Blob& token,
-                           Sp<Request> previous,
+                           Tid socketId,
                            RequestCb&& on_done,
-                           RequestExpiredCb&& on_expired,
-                           SocketCb&& socket_cb);
+                           RequestExpiredCb&& on_expired);
     /**
      * Send a "announce" request to a given node.
      *
@@ -403,6 +403,7 @@ public:
                                  const Value::Id& vid,
                                  const Blob& token,
                                  RequestCb&& on_done,
+                                 RequestErrorCb&& on_error,
                                  RequestExpiredCb&& on_expired);
     /**
      * Send a "update" request to a given node. Used for Listen operations
@@ -434,8 +435,8 @@ public:
      */
     void processMessage(const uint8_t *buf, size_t buflen, SockAddr addr);
 
-    Sp<Node> insertNode(const InfoHash& myid, const SockAddr& addr) {
-        auto n = cache.getNode(myid, addr, scheduler.time(), 0);
+    Sp<Node> insertNode(const InfoHash& id, const SockAddr& addr) {
+        auto n = cache.getNode(id, addr, scheduler.time(), 0);
         onNewNode(n, 0);
         return n;
     }

include/opendht/rng.h

@@ -21,6 +21,8 @@
 #include <random>
 #include <algorithm>
 #include <functional>
+#include <thread>
+#include <stdexcept>
 
 namespace dht {
 namespace crypto {
@@ -111,11 +113,18 @@ using random_device = std::random_device;
 template<class T = std::mt19937, std::size_t N = T::state_size>
 auto getSeededRandomEngine () -> typename std::enable_if<!!N, T>::type {
     typename T::result_type random_data[N];
-    random_device source;
-    std::generate(std::begin(random_data), std::end(random_data), std::ref(source));
-    std::seed_seq seeds(std::begin(random_data), std::end(random_data));
-    T seededEngine (seeds);
-    return seededEngine;
+    for (unsigned i=0; i<256; i++) {
+        try {
+            random_device source;
+            std::generate(std::begin(random_data), std::end(random_data), std::ref(source));
+            std::seed_seq seeds(std::begin(random_data), std::end(random_data));
+            T seededEngine (seeds);
+            return seededEngine;
+        } catch (...) {
+            std::this_thread::sleep_for(std::chrono::milliseconds(1));
+        }
+    }
+    throw std::runtime_error("Can't seed random engine");
 }
 
 }} // dht::crypto

src/dht.cpp

@@ -490,7 +490,29 @@ void Dht::searchSendAnnounceValue(const Sp<Search>& sr) {
                     logger_->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] = {
-                    network_engine.sendRefreshValue(sn->node, sr->id, a.value->id, sn->token, onDone, onExpired),
+                    network_engine.sendRefreshValue(sn->node, sr->id, a.value->id, sn->token, onDone,
+                    [this, ws, node=sn->node, v=a.value, 
+                     onDone, 
+                     onExpired, 
+                     created = a.permanent ? time_point::max() : a.created,
+                     next_refresh_time
+                    ](const net::Request& req, net::DhtProtocolException&& e){
+                        if (e.getCode() == net::DhtProtocolException::NOT_FOUND) {
+                            if (logger_)
+                                logger_->e(node->id, "[node %s] returned error 404: storage not found", node->toString().c_str());
+                            if (auto sr = ws.lock()) {
+                                if (auto sn = sr->getNode(node)) {
+                                    sn->acked[v->id] = {
+                                        network_engine.sendAnnounceValue(sn->node, sr->id, v, created, sn->token, onDone, onExpired),
+                                        next_refresh_time
+                                    };
+                                    scheduler.edit(sr->nextSearchStep, scheduler.time());
+                                    return true;
+                                }
+                            }
+                        }
+                        return false;
+                    }, onExpired),
                     next_refresh_time
                 };
             } else {
@@ -569,33 +591,30 @@ Dht::searchSynchedNodeListen(const Sp<Search>& sr, SearchNode& n)
     std::weak_ptr<Search> ws = sr;
     for (const auto& l : sr->listeners) {
         const auto& query = l.second.query;
-        auto list_token = l.first;
-        if (n.getListenTime(query, listenExp) > scheduler.time())
+        
+        auto r = n.listenStatus.find(query);
+        if (n.getListenTime(r, listenExp) > scheduler.time())
             continue;
         // if (logger_)
         //     logger_->d(sr->id, n.node->id, "[search %s] [node %s] sending 'listen'",
         //        sr->id.toString().c_str(), n.node->toString().c_str());
 
-        auto r = n.listenStatus.find(query);
         if (r == n.listenStatus.end()) {
             r = n.listenStatus.emplace(std::piecewise_construct,
                 std::forward_as_tuple(query),
                 std::forward_as_tuple(
-                [ws,list_token](const std::vector<Sp<Value>>& values, bool expired){
-                    if (auto sr = ws.lock()) {
-                        auto l = sr->listeners.find(list_token);
-                        if (l != sr->listeners.end()) {
-                            l->second.get_cb(values, expired);
+                    l.second.get_cb,
+                    l.second.sync_cb,
+                    n.node->openSocket([this,ws,query](const Sp<Node>& node, net::RequestAnswer&& answer) mutable {
+                        /* on new values */
+                        if (auto sr = ws.lock()) {
+                            scheduler.edit(sr->nextSearchStep, scheduler.time());
+                            sr->insertNode(node, scheduler.time(), answer.ntoken);
+                            if (auto sn = sr->getNode(node)) {
+                                sn->onValues(query, std::move(answer), types, scheduler);
+                            }
                         }
-                    }
-                }, [ws,list_token] (ListenSyncStatus status) {
-                    if (auto sr = ws.lock()) {
-                        auto l = sr->listeners.find(list_token);
-                        if (l != sr->listeners.end()) {
-                            l->second.sync_cb(status);
-                        }
-                    }
-                })).first;
+                    }))).first;
             r->second.cacheExpirationJob = scheduler.add(time_point::max(), [this,ws,query,node=n.node]{
                 if (auto sr = ws.lock()) {
                     if (auto sn = sr->getNode(node)) {
@@ -604,8 +623,7 @@ Dht::searchSynchedNodeListen(const Sp<Search>& sr, SearchNode& n)
                 }
             });
         }
-        auto prev_req = r != n.listenStatus.end() ? r->second.req : nullptr;
-        auto new_req = network_engine.sendListen(n.node, sr->id, *query, n.token, prev_req,
+        auto new_req = network_engine.sendListen(n.node, sr->id, *query, n.token, r->second.socketId,
             [this,ws,query](const net::Request& req, net::RequestAnswer&& answer) mutable
             { /* on done */
                 if (auto sr = ws.lock()) {
@@ -625,16 +643,6 @@ Dht::searchSynchedNodeListen(const Sp<Search>& sr, SearchNode& n)
                         if (auto sn = sr->getNode(req.node))
                             sn->listenStatus.erase(query);
                 }
-            },
-            [this,ws,query](const Sp<Node>& node, net::RequestAnswer&& answer) mutable
-            { /* on new values */
-                if (auto sr = ws.lock()) {
-                    scheduler.edit(sr->nextSearchStep, scheduler.time());
-                    sr->insertNode(node, scheduler.time(), answer.ntoken);
-                    if (auto sn = sr->getNode(node)) {
-                        sn->onValues(query, std::move(answer), types, scheduler);
-                    }
-                }
             }
         );
         // Here the request may have failed and the CachedListenStatus removed
@@ -1743,7 +1751,7 @@ Dht::Dht() : store(), network_engine(logger_, rd, scheduler, {}) {}
 
 Dht::Dht(std::unique_ptr<net::DatagramSocket>&& sock, const Config& config, const Sp<Logger>& l)
     : DhtInterface(l),
-    myid(config.node_id ? config.node_id : InfoHash::getRandom()),
+    myid(config.node_id ? config.node_id : InfoHash::getRandom(rd)),
     store(),
     store_quota(),
     max_store_keys(config.max_store_size ? (int)config.max_store_size : MAX_HASHES),
@@ -1785,13 +1793,13 @@ Dht::Dht(std::unique_ptr<net::DatagramSocket>&& sock, const Config& config, cons
     secret = std::uniform_int_distribution<uint64_t>{}(rd);
     rotateSecrets();
 
+    if (not persistPath.empty())
+        loadState(persistPath);
+
     expire();
 
     if (logger_)
         logger_->d("DHT node initialised with ID %s", myid.toString().c_str());
-
-    if (not persistPath.empty())
-        loadState(persistPath);
 }
 
 bool
@@ -2070,13 +2078,13 @@ Dht::importValues(const std::vector<ValuesExport>& import)
 {
     const auto& now = scheduler.time();
 
-    for (const auto& node : import) {
-        if (node.second.empty())
+    for (const auto& value : import) {
+        if (value.second.empty())
             continue;
 
         try {
             msgpack::unpacked msg;
-            msgpack::unpack(msg, (const char*)node.second.data(), node.second.size());
+            msgpack::unpack(msg, (const char*)value.second.data(), value.second.size());
             auto valarr = msg.get();
             if (valarr.type != msgpack::type::ARRAY)
                 throw msgpack::type_error();
@@ -2091,15 +2099,15 @@ Dht::importValues(const std::vector<ValuesExport>& import)
                     tmp_val.msgpack_unpack(valel.via.array.ptr[1]);
                 } catch (const std::exception&) {
                     if (logger_)
-                        logger_->e(node.first, "Error reading value at %s", node.first.toString().c_str());
+                        logger_->e(value.first, "Error reading value at %s", value.first.toString().c_str());
                     continue;
                 }
                 val_time = std::min(val_time, now);
-                storageStore(node.first, std::make_shared<Value>(std::move(tmp_val)), val_time);
+                storageStore(value.first, std::make_shared<Value>(std::move(tmp_val)), val_time);
             }
         } catch (const std::exception&) {
             if (logger_)
-                logger_->e(node.first, "Error reading values at %s", node.first.toString().c_str());
+                logger_->e(value.first, "Error reading values at %s", value.first.toString().c_str());
             continue;
         }
     }
@@ -2175,7 +2183,6 @@ Dht::onError(Sp<net::Request> req, net::DhtProtocolException e) {
         if (logger_)
             logger_->e(node->id, "[node %s] token flush", node->toString().c_str());
         node->authError();
-        node->cancelRequest(req);
         for (auto& srp : searches(node->getFamily())) {
             auto& sr = srp.second;
             for (auto& n : sr->nodes) {
@@ -2504,12 +2511,24 @@ Dht::onAnnounceDone(const Sp<Node>& node, net::RequestAnswer& answer, Sp<Search>
 }
 
 
+struct DhtState {
+    unsigned v {1};
+    InfoHash id;
+    std::vector<NodeExport> nodes;
+    std::vector<ValuesExport> values;
+
+    MSGPACK_DEFINE_MAP(v, id, nodes, values)
+};
+
 void
 Dht::saveState(const std::string& path) const
 {
+    DhtState state;
+    state.id = myid;
+    state.nodes = exportNodes();
+    state.values = exportValues();
     std::ofstream file(path);
-    msgpack::pack(file, exportNodes());
-    msgpack::pack(file, exportValues());
+    msgpack::pack(file, state);    
 }
 
 void
@@ -2535,19 +2554,16 @@ Dht::loadState(const std::string& path)
         // Import nodes
         msgpack::object_handle oh;
         if (pac.next(oh)) {
-            {
-                auto imported_nodes = oh.get().as<std::vector<NodeExport>>();
-                if (logger_)
-                    logger_->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>>();
-                if (logger_)
-                    logger_->d("Importing %zu values", imported_values.size());
-                importValues(imported_values);
-            }
+            auto state = oh.get().as<DhtState>();
+            if (logger_)
+                logger_->d("Importing %zu nodes", state.nodes.size());
+            if (state.id)
+                myid = state.id;
+            std::vector<Sp<Node>> tmpNodes;
+            tmpNodes.reserve(state.nodes.size());
+            for (const auto& node : state.nodes)
+                tmpNodes.emplace_back(network_engine.insertNode(node.id, SockAddr(node.ss, node.sslen)));
+            importValues(state.values);
         }
     } catch (const std::exception& e) {
         if (logger_)

src/dht_proxy_server.cpp

@@ -350,11 +350,13 @@ DhtProxyServer::saveState(Os& stream) {
         pk.pack("puts");
         pk.pack(puts_);
     }
+#ifdef OPENDHT_PUSH_NOTIFICATIONS
     {
         std::lock_guard<std::mutex> lock(lockListener_);
         pk.pack("pushListeners");
         pk.pack(pushListeners_);
     }
+#endif
 }
 
 template <typename Is>
@@ -379,19 +381,25 @@ DhtProxyServer::loadState(Is& is, size_t size) {
                         pput.second.expireTimer = std::make_unique<asio::steady_timer>(io_context(), pput.second.expiration);
                         pput.second.expireTimer->async_wait(std::bind(&DhtProxyServer::handleCancelPermamentPut, this,
                                                 std::placeholders::_1, put.first, pput.first));
-                        auto jsonProvider = [infoHash=put.first.toString(), clientId=pput.second.clientId, vid = pput.first, sessionCtx = pput.second.sessionCtx](){
-                            Json::Value json;
-                            json["timeout"] = infoHash;
-                            json["to"] = clientId;
-                            json["vid"] = std::to_string(vid);
-                            std::lock_guard<std::mutex> l(sessionCtx->lock);
-                            json["s"] = sessionCtx->sessionId;
-                            return json;
-                        };
-                        pput.second.expireNotifyTimer = std::make_unique<asio::steady_timer>(io_context(), pput.second.expiration - proxy::OP_MARGIN);
-                        pput.second.expireNotifyTimer->async_wait(std::bind(
-                            &DhtProxyServer::handleNotifyPushListenExpire, this,
-                            std::placeholders::_1, pput.second.pushToken, std::move(jsonProvider), pput.second.type));
+#ifdef OPENDHT_PUSH_NOTIFICATIONS
+                        if (not pput.second.pushToken.empty()) {
+                            auto jsonProvider = [infoHash=put.first.toString(), clientId=pput.second.clientId, vid = pput.first, sessionCtx = pput.second.sessionCtx](){
+                                Json::Value json;
+                                json["timeout"] = infoHash;
+                                json["to"] = clientId;
+                                json["vid"] = std::to_string(vid);
+                                if (sessionCtx) {
+                                    std::lock_guard<std::mutex> l(sessionCtx->lock);
+                                    json["s"] = sessionCtx->sessionId;
+                                }
+                                return json;
+                            };
+                            pput.second.expireNotifyTimer = std::make_unique<asio::steady_timer>(io_context(), pput.second.expiration - proxy::OP_MARGIN);
+                            pput.second.expireNotifyTimer->async_wait(std::bind(
+                                &DhtProxyServer::handleNotifyPushListenExpire, this,
+                                std::placeholders::_1, pput.second.pushToken, std::move(jsonProvider), pput.second.type));
+                        }
+#endif
                         dht_->put(put.first, pput.second.value, DoneCallbackSimple{}, time_point::max(), true);
                     }
                 }
@@ -555,6 +563,10 @@ DhtProxyServer::updateStats(std::shared_ptr<NodeInfo> info) const
 #ifdef OPENDHT_PUSH_NOTIFICATIONS
     stats.pushListenersCount = pushListeners_.size();
 #endif
+    stats.totalPermanentPuts = 0;
+    std::for_each(puts_.begin(), puts_.end(), [&stats](const auto& put) {
+        stats.totalPermanentPuts += put.second.puts.size();
+    });
     stats.putCount = puts_.size();
     stats.listenCount = listeners_.size();
     stats.nodeInfo = std::move(info);
@@ -1142,8 +1154,8 @@ DhtProxyServer::put(restinio::request_handle_t request,
                 auto& sPuts = puts_[infoHash];
                 if (value->id == Value::INVALID_ID) {
                     for (auto& pp : sPuts.puts) {
-                        if (pp.second.pushToken == pushToken 
-                            and pp.second.clientId == clientId 
+                        if (pp.second.pushToken == pushToken
+                            and pp.second.clientId == clientId
                             and pp.second.value->contentEquals(*value))
                         {
                             pp.second.expireTimer->expires_at(timeout);

src/dhtrunner.cpp

@@ -32,6 +32,8 @@
 #include "dht_proxy_client.h"
 #endif
 
+#include <fstream>
+
 namespace dht {
 
 constexpr std::chrono::seconds DhtRunner::BOOTSTRAP_PERIOD;
@@ -90,16 +92,46 @@ DhtRunner::run(const char* ip4, const char* ip6, const char* service, const Conf
 {
     auto res4 = SockAddr::resolve(ip4, service);
     auto res6 = SockAddr::resolve(ip6, service);
-    run(res4.empty() ? SockAddr() : res4.front(),
-        res6.empty() ? SockAddr() : res6.front(), config, std::move(context));
+    if (res4.empty())
+        res4.emplace_back();
+    if (res6.empty())
+        res6.emplace_back();
+    run(res4.front(), res6.front(), config, std::move(context));
 }
 
 void
-DhtRunner::run(const SockAddr& local4, const SockAddr& local6, const Config& config, Context&& context)
+DhtRunner::run(SockAddr& local4, SockAddr& local6, const Config& config, Context&& context)
 {
     if (running == State::Idle) {
+        auto state_path = config.dht_config.node_config.persist_path;
+        if (not state_path.empty()) {
+            state_path += "_port.txt";
+            std::ifstream inConfig(state_path);
+            if (inConfig.is_open()) {
+                in_port_t port;
+                if (inConfig >> port) {
+                    if (context.logger)
+                        context.logger->d("[runner %p] Using IPv4 port %hu from saved configuration", this, port);
+                    if (local4.getPort() == 0)
+                        local4.setPort(port);
+                }
+                if (inConfig >> port) {
+                    if (context.logger)
+                        context.logger->d("[runner %p] Using IPv6 port %hu from saved configuration", this, port);
+                    if (local6.getPort() == 0)
+                        local6.setPort(port);
+                }
+            }
+        }
+
         if (not context.sock)
             context.sock.reset(new net::UdpSocket(local4, local6, context.logger));
+
+        if (not state_path.empty()) {
+            std::ofstream outConfig(state_path);
+            outConfig << context.sock->getBoundRef(AF_INET).getPort() << std::endl;
+            outConfig << context.sock->getBoundRef(AF_INET6).getPort() << std::endl;
+        }
         run(config, std::move(context));
     }
 }

src/network_engine.cpp

@@ -144,23 +144,25 @@ NetworkEngine::tellListenerRefreshed(Sp<Node> n, Tid socket_id, const InfoHash&,
     msgpack::sbuffer buffer;
     msgpack::packer<msgpack::sbuffer> pk(&buffer);
 
-    pk.pack_map(4 + (version >= 1 ? 2 : 0) + (config.network?1:0));
+    pk.pack_map(4 + (version >= 1 ? 1 : 0) + (config.network?1:0));
 
-    pk.pack(KEY_U);
-    pk.pack_map(1 + (not values.empty()?1:0) + (not token.empty()?1:0));
-    pk.pack(KEY_REQ_ID); pk.pack(myid);
-    if (not token.empty()) {
-        pk.pack(KEY_REQ_TOKEN); packToken(pk, token);
-    }
-    if (not values.empty()) {
-        pk.pack(KEY_REQ_REFRESHED);
-        pk.pack(values);
-        if (logger_)
-            logger_->d(n->id, "[node %s] sending %zu refreshed values", n->toString().c_str(), values.size());
-    }
+    pk.pack(version >= 1 ? KEY_A : KEY_U);
+        pk.pack_map(1 + (version >= 1 ? 1 : 0) + (not values.empty()?1:0) + (not token.empty()?1:0));
+        pk.pack(KEY_REQ_ID); pk.pack(myid);
+        if (version >= 1) {
+            pk.pack(KEY_REQ_SID);   pk.pack(socket_id);
+        }
+        if (not token.empty()) {
+            pk.pack(KEY_REQ_TOKEN); packToken(pk, token);
+        }
+        if (not values.empty()) {
+            pk.pack(KEY_REQ_REFRESHED);
+            pk.pack(values);
+            if (logger_)
+                logger_->d(n->id, "[node %s] sending %zu refreshed values", n->toString().c_str(), values.size());
+        }
 
-    pk.pack(KEY_TID); pk.pack(socket_id);
-    pk.pack(KEY_Y); pk.pack(KEY_R);
+    pk.pack(KEY_Y); pk.pack(version >= 1 ? KEY_Q : KEY_R);
     pk.pack(KEY_UA); pk.pack(my_v);
     if (config.network) {
         pk.pack(KEY_NETID); pk.pack(config.network);
@@ -168,9 +170,8 @@ NetworkEngine::tellListenerRefreshed(Sp<Node> n, Tid socket_id, const InfoHash&,
 
     if (version >= 1) {
         Tid tid (n->getNewTid());
-        Tid sid (socket_id);
 
-        pk.pack(KEY_REQ_SID);   pk.pack(sid);
+        pk.pack(KEY_Q);   pk.pack(QUERY_UPDATE);
         pk.pack(KEY_TID); pk.pack(tid);
 
         auto req = std::make_shared<Request>(MessageType::UpdateValue, tid, n,
@@ -182,6 +183,7 @@ NetworkEngine::tellListenerRefreshed(Sp<Node> n, Tid socket_id, const InfoHash&,
         ++out_stats.updateValue;
         return;
     }
+    pk.pack(KEY_TID); pk.pack(socket_id);
 
     // send response
     send(n->getAddr(), buffer.data(), buffer.size());
@@ -193,23 +195,25 @@ NetworkEngine::tellListenerExpired(Sp<Node> n, Tid socket_id, const InfoHash&, c
     msgpack::sbuffer buffer;
     msgpack::packer<msgpack::sbuffer> pk(&buffer);
 
-    pk.pack_map(4 + (version >= 1 ? 2 : 0) + (config.network?1:0));
+    pk.pack_map(4 + (version >= 1 ? 1 : 0) + (config.network?1:0));
 
-    pk.pack(KEY_U);
-    pk.pack_map(1 + (not values.empty()?1:0) + (not token.empty()?1:0));
-    pk.pack(KEY_REQ_ID); pk.pack(myid);
-    if (not token.empty()) {
-        pk.pack(KEY_REQ_TOKEN); packToken(pk, token);
-    }
-    if (not values.empty()) {
-        pk.pack(KEY_REQ_EXPIRED);
-        pk.pack(values);
-        if (logger_)
-            logger_->d(n->id, "[node %s] sending %zu expired values", n->toString().c_str(), values.size());
-    }
+    pk.pack(version >= 1 ? KEY_A : KEY_U);
+        pk.pack_map(1 + (version >= 1 ? 1 : 0) + (not values.empty()?1:0) + (not token.empty()?1:0));
+        pk.pack(KEY_REQ_ID); pk.pack(myid);
+        if (version >= 1) {
+            pk.pack(KEY_REQ_SID);   pk.pack(socket_id);
+        }
+        if (not token.empty()) {
+            pk.pack(KEY_REQ_TOKEN); packToken(pk, token);
+        }
+        if (not values.empty()) {
+            pk.pack(KEY_REQ_EXPIRED);
+            pk.pack(values);
+            if (logger_)
+                logger_->d(n->id, "[node %s] sending %zu expired values", n->toString().c_str(), values.size());
+        }
 
-    pk.pack(KEY_TID); pk.pack(socket_id);
-    pk.pack(KEY_Y); pk.pack(KEY_R);
+    pk.pack(KEY_Y); pk.pack(version >= 1 ? KEY_Q : KEY_R);
     pk.pack(KEY_UA); pk.pack(my_v);
     if (config.network) {
         pk.pack(KEY_NETID); pk.pack(config.network);
@@ -217,10 +221,9 @@ NetworkEngine::tellListenerExpired(Sp<Node> n, Tid socket_id, const InfoHash&, c
 
     if (version >= 1) {
         Tid tid (n->getNewTid());
-        Tid sid (socket_id);
 
-        pk.pack(KEY_REQ_SID);   pk.pack(sid);
-        pk.pack(KEY_TID); pk.pack(tid);
+        pk.pack(KEY_Q);   pk.pack(QUERY_UPDATE);
+        pk.pack(KEY_TID);     pk.pack(tid);
 
         auto req = std::make_shared<Request>(MessageType::UpdateValue, tid, n,
             Blob(buffer.data(), buffer.data() + buffer.size()),
@@ -231,6 +234,7 @@ NetworkEngine::tellListenerExpired(Sp<Node> n, Tid socket_id, const InfoHash&, c
         ++out_stats.updateValue;
         return;
     }
+    pk.pack(KEY_TID); pk.pack(socket_id);
 
     // send response
     send(n->getAddr(), buffer.data(), buffer.size());
@@ -556,7 +560,8 @@ NetworkEngine::process(std::unique_ptr<ParsedMessage>&& msg, const SockAddr& fro
             {
                 req->last_try = time_point::min();
                 req->reply_time = time_point::min();
-                onError(req, DhtProtocolException {msg->error_code});
+                if (not req->setError(DhtProtocolException {msg->error_code}))
+                    onError(req, DhtProtocolException {msg->error_code});
             } else {
                 if (logIncoming_)
                     if (logger_)
@@ -568,8 +573,11 @@ NetworkEngine::process(std::unique_ptr<ParsedMessage>&& msg, const SockAddr& fro
         case MessageType::Reply:
             if (req) { /* request reply */
                 auto& r = *req;
-                if (r.getType() == MessageType::AnnounceValue or r.getType() == MessageType::Listen)
+                if (r.getType() == MessageType::AnnounceValue
+                 or r.getType() == MessageType::Listen
+                 or r.getType() == MessageType::Refresh) {
                     r.node->authSuccess();
+                }
                 r.reply_time = scheduler.time();
 
                 deserializeNodes(*msg, from);
@@ -653,6 +661,7 @@ NetworkEngine::process(std::unique_ptr<ParsedMessage>&& msg, const SockAddr& fro
                     rsocket->on_receive(node, std::move(*msg));
                 else if (logger_)
                     logger_->e(msg->info_hash, node->id, "[node %s] 'update' request without socket for %s", node->toString().c_str(), msg->info_hash.toString().c_str());
+                sendListenConfirmation(from, msg->tid);
                 break;
             }
             default:
@@ -1064,27 +1073,11 @@ NetworkEngine::sendListen(Sp<Node> n,
         const InfoHash& hash,
         const Query& query,
         const Blob& token,
-        Sp<Request> previous,
+        Tid socketId,
         RequestCb&& on_done,
-        RequestExpiredCb&& on_expired,
-        SocketCb&& socket_cb)
+        RequestExpiredCb&& on_expired)
 {
-    Tid socket;
     Tid tid (n->getNewTid());
-    if (previous and previous->node == n) {
-        socket = previous->getSocket();
-    } else {
-        if (previous)
-            if (logger_)
-                logger_->e(hash, "[node %s] trying refresh listen contract with wrong node", previous->node->toString().c_str());
-        socket = n->openSocket(std::move(socket_cb));
-    }
-
-    if (not socket) {
-        if (logger_)
-            logger_->e(hash, "[node %s] unable to get a valid socket for listen. Aborting listen", n->toString().c_str());
-        return {};
-    }
     msgpack::sbuffer buffer;
     msgpack::packer<msgpack::sbuffer> pk(&buffer);
     pk.pack_map(5+(config.network?1:0));
@@ -1095,7 +1088,7 @@ NetworkEngine::sendListen(Sp<Node> n,
       pk.pack(KEY_VERSION);   pk.pack(1);
       pk.pack(KEY_REQ_H);     pk.pack(hash);
       pk.pack(KEY_REQ_TOKEN); packToken(pk, token);
-      pk.pack(KEY_REQ_SID);   pk.pack(socket);
+      pk.pack(KEY_REQ_SID);   pk.pack(socketId);
       if (has_query) {
           pk.pack(KEY_REQ_QUERY); pk.pack(query);
       }
@@ -1117,8 +1110,7 @@ NetworkEngine::sendListen(Sp<Node> n,
         [=](const Request& req_status, bool done) { /* on expired */
             if (on_expired)
                 on_expired(req_status, done);
-        },
-        socket
+        }
     );
     sendRequest(req);
     ++out_stats.listen;
@@ -1255,6 +1247,7 @@ NetworkEngine::sendRefreshValue(Sp<Node> n,
                 const Value::Id& vid,
                 const Blob& token,
                 RequestCb&& on_done,
+                RequestErrorCb&& on_error,
                 RequestExpiredCb&& on_expired)
 {
     Tid tid (n->getNewTid());
@@ -1290,6 +1283,7 @@ NetworkEngine::sendRefreshValue(Sp<Node> n,
                 }
             }
         },
+        on_error,
         [=](const Request& req_status, bool done) { /* on expired */
             if (on_expired) {
                 on_expired(req_status, done);
@@ -1331,7 +1325,7 @@ NetworkEngine::sendError(const SockAddr& addr,
 {
     msgpack::sbuffer buffer;
     msgpack::packer<msgpack::sbuffer> pk(&buffer);
-    pk.pack_map(4 + (include_id?1:0));
+    pk.pack_map(4 + (include_id?1:0) + (config.network?1:0));
 
     pk.pack(KEY_E); pk.pack_array(2);
       pk.pack(code);

src/node.cpp

@@ -116,7 +116,6 @@ Node::cancelRequest(const Sp<net::Request>& req)
 {
     if (req) {
         req->cancel();
-        closeSocket(req->closeSocket());
         requests_.erase(req->getTid());
     }
 }

src/request.h

@@ -27,6 +27,7 @@ struct Node;
 namespace net {
 
 class NetworkEngine;
+class DhtProtocolException;
 struct ParsedMessage;
 
 /*!
@@ -72,16 +73,19 @@ struct Request {
             Sp<Node> node,
             Blob&& msg,
             std::function<void(const Request&, ParsedMessage&&)> on_done,
-            std::function<void(const Request&, bool)> on_expired,
-            Tid socket = 0) :
-        node(node), tid(tid), type(type), on_done(on_done), on_expired(on_expired), msg(std::move(msg)), socket(socket) { }
+            std::function<void(const Request&, bool)> on_expired) :
+        node(node), tid(tid), type(type), on_done(on_done), on_expired(on_expired), msg(std::move(msg)) { }
+    Request(MessageType type, Tid tid,
+            Sp<Node> node,
+            Blob&& msg,
+            std::function<void(const Request&, ParsedMessage&&)> on_done,
+            std::function<bool(const Request&, DhtProtocolException&&)> on_error,
+            std::function<void(const Request&, bool)> on_expired) :
+        node(node), tid(tid), type(type), on_done(on_done), on_error(on_error), on_expired(on_expired), msg(std::move(msg)) { }
 
     Tid getTid() const { return tid; }
     MessageType getType() const { return type; }
 
-    Tid getSocket() { return socket; }
-    Tid closeSocket() { auto ret = socket; socket = 0; return ret; }
-
     void setExpired() {
         if (pending()) {
             state_ = Request::State::EXPIRED;
@@ -96,6 +100,15 @@ struct Request {
             clear();
         }
     }
+    bool setError(DhtProtocolException&& e) {
+        if (pending()) {
+            state_ = Request::State::EXPIRED;
+            bool handled = on_error and on_error(*this, std::forward<DhtProtocolException>(e));
+            clear();
+            return handled;
+        }
+        return true;
+    }
 
     void cancel() {
         if (pending()) {
@@ -113,6 +126,7 @@ private:
 
     void clear() {
         on_done = {};
+        on_error = {};
         on_expired = {};
         msg = {};
         parts = {};
@@ -128,11 +142,11 @@ private:
     time_point last_try {time_point::min()};   /* time of the last attempt to process the request. */
 
     std::function<void(const Request&, ParsedMessage&&)> on_done {};
+    std::function<bool(const Request&, DhtProtocolException&&)> on_error {};
     std::function<void(const Request&, bool)> on_expired {};
 
     Blob msg {};                      /* the serialized message. */
     std::vector<Blob> parts;
-    Tid socket;   /* the socket used for further reponses. */
 };
 
 } /* namespace net  */

src/search.h

@@ -65,10 +65,16 @@ struct Dht::SearchNode {
         ValueCache cache;
         Sp<Scheduler::Job> cacheExpirationJob {};
         Sp<net::Request> req {};
-        CachedListenStatus(ValueStateCallback&& cb, SyncCallback&& scb)
-         : cache(std::forward<ValueStateCallback>(cb), std::forward<SyncCallback>(scb)) {}
+        Tid socketId {0};
+        CachedListenStatus(ValueStateCallback&& cb, SyncCallback scb, Tid sid)
+         : cache(std::forward<ValueStateCallback>(cb), std::forward<SyncCallback>(scb)), socketId(sid) {}
         CachedListenStatus(CachedListenStatus&&) = delete;
         CachedListenStatus(const CachedListenStatus&) = delete;
+        ~CachedListenStatus() {
+            if (socketId and req and req->node) {
+                req->node->closeSocket(socketId);
+            }
+        }
     };
     using NodeListenerStatus = std::map<Sp<Query>, CachedListenStatus>;
 
@@ -344,13 +350,15 @@ struct Dht::SearchNode {
      * Assuming the node is synced, should the "listen" request with Query q be
      * sent to this node now ?
      */
-    time_point getListenTime(const Sp<Query>& q, duration listen_expire) const {
-        auto listen_status = listenStatus.find(q);
-        if (listen_status == listenStatus.end() or not listen_status->second.req)
+    time_point getListenTime(const decltype(listenStatus)::const_iterator listen_status, duration listen_expire) const {
+        if (listen_status == listenStatus.cend() or not listen_status->second.req)
             return time_point::min();
         return listen_status->second.req->pending() ? time_point::max() :
             listen_status->second.req->reply_time + listen_expire - REANNOUNCE_MARGIN;
     }
+    time_point getListenTime(const Sp<Query>& q, duration listen_expire) const {
+        return getListenTime(listenStatus.find(q), listen_expire);
+    }
 
     /**
      * Is this node expired or candidate

tools/dhtproxy_stats.py

@@ -7,7 +7,7 @@ stats_total = {"users":0, "pushListenersCount":0, "listenCount":0, "totalListene
 
 for i in range(80,101):
     print("Collecting stats for proxy " + str(i))
-    response = requests.request('STATS', 'http://dhtproxy.jami.net:' + str(i))
+    response = requests.request('STATS', 'http://127.0.0.1:' + str(i))
 
     if response.status_code == 200:
         result = response.json()
@@ -38,14 +38,11 @@ for i in range(80,101):
         except:
             pass
         try:
-            total = 0
-            for h,v in result["puts"].items():
-                total += int(v)
-            stats['totalPermanentPuts'] = total
-            stats_total['totalPermanentPuts'] += total
+            stats['totalPermanentPuts'] = int(result["totalPermanentPuts"])
+            stats_total['totalPermanentPuts'] += int(result["totalPermanentPuts"])
         except:
             pass
-        
+
         stats['timestamp'] = str(ts)
 
         #with open("stats_proxy_" + str(i), "a") as stat_file: