update to version 2.1.3

.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

@@ -434,8 +434,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,7 @@
 #include <random>
 #include <algorithm>
 #include <functional>
+#include <thread>
 
 namespace dht {
 namespace crypto {
@@ -111,11 +112,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

@@ -1743,7 +1743,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 +1785,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 +2070,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 +2091,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;
         }
     }
@@ -2504,12 +2504,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 +2547,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_client.cpp

@@ -98,7 +98,7 @@ getRandomSessionId(size_t length = 8) {
     static constexpr const char chars[] = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789!#$%&()*+,./:;<=>?@[]^_`{|}~";
     std::string str(length, 0);
     crypto::random_device rdev;
-    std::uniform_int_distribution<> dist(0, (sizeof(chars)/sizeof(char)) - 1);
+    std::uniform_int_distribution<> dist(0, (sizeof(chars)/sizeof(char)) - 2);
     std::generate_n( str.begin(), length, [&]{ return chars[dist(rdev)]; } );
     return str;
 }

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());
@@ -653,6 +657,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:

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: