build: update version to 2.3.3

This reverts commit 5d771e119c.

.github/workflows/ccpp.yml

@@ -13,12 +13,9 @@ jobs:
       run: |
         sudo apt install 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
+        autotools-dev autoconf libfmt-dev libhttp-parser-dev libmsgpack-dev libargon2-0-dev
     - name: autogen
       run: ./autogen.sh
-    - name: argon2
-      run: |
-        cd argon2 && make && sudo make install && cd ..
     - name: asio
       run: |
         wget https://github.com/aberaud/asio/archive/b2b7a1c166390459e1c169c8ae9ef3234b361e3f.tar.gz \

CMakeLists.txt

@@ -1,4 +1,16 @@
-cmake_minimum_required (VERSION 3.1)
+cmake_minimum_required (VERSION 3.10)
+if(POLICY CMP0065)
+  cmake_policy(SET CMP0065 NEW)
+endif()
+if(POLICY CMP0069)
+  cmake_policy(SET CMP0069 NEW)
+endif()
+if(POLICY CMP0073)
+  cmake_policy(SET CMP0073 NEW)
+endif()
+if(POLICY CMP0074)
+  cmake_policy(SET CMP0074 NEW)
+endif()
 project (opendht)
 
 include(CMakePackageConfigHelpers)
@@ -7,7 +19,7 @@ include(FindPkgConfig)
 include(cmake/CheckAtomic.cmake)
 
 set (opendht_VERSION_MAJOR 2)
-set (opendht_VERSION_MINOR 3.0)
+set (opendht_VERSION_MINOR 3.3)
 set (opendht_VERSION ${opendht_VERSION_MAJOR}.${opendht_VERSION_MINOR})
 set (PACKAGE_VERSION ${opendht_VERSION})
 set (VERSION "${opendht_VERSION}")
@@ -45,7 +57,7 @@ if (NOT MSVC)
     find_package (PkgConfig REQUIRED)
     find_package (GnuTLS 3.3 REQUIRED)
     pkg_search_module (Nettle REQUIRED nettle)
-    find_package (Msgpack 1.2 REQUIRED)
+    find_package (msgpack REQUIRED)
     if (OPENDHT_TOOLS)
         find_package (Readline 6 REQUIRED)
     endif ()
@@ -131,7 +143,7 @@ else ()
     set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} ${DISABLE_MSC_WARNINGS}")
     set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} /MP")
 endif ()
-set (CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -DMSGPACK_DISABLE_LEGACY_NIL -DMSGPACK_DISABLE_LEGACY_CONVERT")
+set (CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -DMSGPACK_NO_BOOST -DMSGPACK_DISABLE_LEGACY_NIL -DMSGPACK_DISABLE_LEGACY_CONVERT")
 
 if (NOT CMAKE_BUILD_TYPE)
     set(CMAKE_BUILD_TYPE Release)
@@ -416,6 +428,7 @@ if (OPENDHT_C)
     )
     install (TARGETS opendht-c DESTINATION ${CMAKE_INSTALL_LIBDIR} EXPORT opendht-c)
     install (FILES ${CMAKE_CURRENT_BINARY_DIR}/opendht-c.pc DESTINATION ${CMAKE_INSTALL_LIBDIR}/pkgconfig)
+    install (FILES c/opendht_c.h DESTINATION ${CMAKE_INSTALL_PREFIX}/include/opendht)
 endif ()
 
 if (OPENDHT_TOOLS)

c/opendht.cpp

@@ -97,10 +97,20 @@ const char* dht_value_get_user_type(const dht_value* data) {
     return vsp->user_type.c_str();
 }
 
+void dht_value_set_user_type(dht_value* data, const char* user_type) {
+    (*reinterpret_cast<ValueSp*>(data))->user_type = user_type;
+}
+
 dht_value* dht_value_new(const uint8_t* data, size_t size) {
     return reinterpret_cast<dht_value*>(new ValueSp(std::make_shared<dht::Value>(data, size)));
 }
 
+dht_value* dht_value_new_from_string(const char* str) {
+    ValueSp value = std::make_shared<dht::Value>((const uint8_t*)str, strlen(str));
+    value->user_type = "text/plain";
+    return reinterpret_cast<dht_value*>(new ValueSp(std::move(value)));
+}
+
 dht_value* dht_value_ref(const dht_value* v) {
     return reinterpret_cast<dht_value*>(new ValueSp(*reinterpret_cast<const ValueSp*>(v)));
 }
@@ -173,7 +183,7 @@ int dht_privatekey_export(const dht_privatekey* k, char* out, size_t* out_size,
 
 dht_publickey* dht_privatekey_get_publickey(const dht_privatekey* k) {
     const auto& key = *reinterpret_cast<const PrivkeySp*>(k);
-    return reinterpret_cast<dht_publickey*>(new PubkeySp(std::make_shared<dht::crypto::PublicKey>(key->getPublicKey())));
+    return reinterpret_cast<dht_publickey*>(new PubkeySp(key->getSharedPublicKey()));
 }
 
 void dht_privatekey_delete(dht_privatekey* pk) {

c/opendht_c.h

@@ -82,6 +82,7 @@ struct OPENDHT_C_PUBLIC dht_value;
 typedef struct dht_value dht_value;
 typedef uint64_t dht_value_id;
 OPENDHT_C_PUBLIC dht_value* dht_value_new(const uint8_t* data, size_t size);
+OPENDHT_C_PUBLIC dht_value* dht_value_new_from_string(const char* str);
 OPENDHT_C_PUBLIC dht_value* dht_value_ref(const dht_value*);
 OPENDHT_C_PUBLIC void dht_value_unref(dht_value*);
 OPENDHT_C_PUBLIC dht_data_view dht_value_get_data(const dht_value* data);
@@ -89,6 +90,7 @@ OPENDHT_C_PUBLIC dht_value_id dht_value_get_id(const dht_value* data);
 OPENDHT_C_PUBLIC dht_publickey* dht_value_get_owner(const dht_value* data);
 OPENDHT_C_PUBLIC dht_infohash dht_value_get_recipient(const dht_value* data);
 OPENDHT_C_PUBLIC const char* dht_value_get_user_type(const dht_value* data);
+OPENDHT_C_PUBLIC void dht_value_set_user_type(dht_value* data, const char* user_type);
 
 // callbacks
 typedef bool (*dht_get_cb)(const dht_value* value, void* user_data);

cmake/FindMsgpack.cmake

@@ -1,64 +0,0 @@
-# - Try to find msgpack
-# Once done this will define
-#  MSGPACK_FOUND - System has msgpack
-#  MSGPACK_INCLUDE_DIRS - The msgpack include directories
-#  MSGPACK_LIBRARIES - The libraries needed to use msgpack
-
-if(NOT MSGPACK_USE_BUNDLED)
-  find_package(PkgConfig)
-  if (PKG_CONFIG_FOUND)
-    pkg_search_module(PC_MSGPACK QUIET
-      msgpackc>=${Msgpack_FIND_VERSION}
-      msgpack>=${Msgpack_FIND_VERSION})
-  endif()
-else()
-  set(PC_MSGPACK_INCLUDEDIR)
-  set(PC_MSGPACK_INCLUDE_DIRS)
-  set(PC_MSGPACK_LIBDIR)
-  set(PC_MSGPACK_LIBRARY_DIRS)
-  set(LIMIT_SEARCH NO_DEFAULT_PATH)
-endif()
-
-set(MSGPACK_DEFINITIONS ${PC_MSGPACK_CFLAGS_OTHER})
-
-find_path(MSGPACK_INCLUDE_DIR msgpack/version_master.h
-  HINTS ${PC_MSGPACK_INCLUDEDIR} ${PC_MSGPACK_INCLUDE_DIRS}
-  ${LIMIT_SEARCH})
-
-if(MSGPACK_INCLUDE_DIR)
-  file(READ ${MSGPACK_INCLUDE_DIR}/msgpack/version_master.h msgpack_version_h)
-  string(REGEX REPLACE ".*MSGPACK_VERSION_MAJOR +([0-9]+).*" "\\1" MSGPACK_VERSION_MAJOR "${msgpack_version_h}")
-  string(REGEX REPLACE ".*MSGPACK_VERSION_MINOR +([0-9]+).*" "\\1" MSGPACK_VERSION_MINOR "${msgpack_version_h}")
-  string(REGEX REPLACE ".*MSGPACK_VERSION_REVISION +([0-9]+).*" "\\1" MSGPACK_VERSION_REVISION "${msgpack_version_h}")
-  set(MSGPACK_VERSION_STRING "${MSGPACK_VERSION_MAJOR}.${MSGPACK_VERSION_MINOR}.${MSGPACK_VERSION_REVISION}")
-else()
-  set(MSGPACK_VERSION_STRING)
-endif()
-
-# If we're asked to use static linkage, add libmsgpack{,c}.a as a preferred library name.
-if(MSGPACK_USE_STATIC)
-  list(APPEND MSGPACK_NAMES
-    "${CMAKE_STATIC_LIBRARY_PREFIX}msgpackc${CMAKE_STATIC_LIBRARY_SUFFIX}"
-    "${CMAKE_STATIC_LIBRARY_PREFIX}msgpack${CMAKE_STATIC_LIBRARY_SUFFIX}")
-endif()
-
-list(APPEND MSGPACK_NAMES msgpackc msgpack)
-
-find_library(MSGPACK_LIBRARY NAMES ${MSGPACK_NAMES}
-  # Check each directory for all names to avoid using headers/libraries from
-  # different places.
-  NAMES_PER_DIR
-  HINTS ${PC_MSGPACK_LIBDIR} ${PC_MSGPACK_LIBRARY_DIRS}
-  ${LIMIT_SEARCH})
-
-mark_as_advanced(MSGPACK_INCLUDE_DIR MSGPACK_LIBRARY)
-
-set(MSGPACK_LIBRARIES ${MSGPACK_LIBRARY})
-set(MSGPACK_INCLUDE_DIRS ${MSGPACK_INCLUDE_DIR})
-
-include(FindPackageHandleStandardArgs)
-# handle the QUIETLY and REQUIRED arguments and set MSGPACK_FOUND to TRUE
-# if all listed variables are TRUE
-find_package_handle_standard_args(Msgpack
-  REQUIRED_VARS MSGPACK_LIBRARY MSGPACK_INCLUDE_DIR
-  VERSION_VAR MSGPACK_VERSION_STRING)

configure.ac

@@ -1,7 +1,7 @@
 dnl define macros
 m4_define([opendht_major_version], 2)
 m4_define([opendht_minor_version], 3)
-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])
 
@@ -175,7 +175,7 @@ AM_COND_IF([PROXY_CLIENT_OR_SERVER], [
     AS_IF([test "x$http_parser_headers" != "xyes"], AC_MSG_ERROR([Missing HttpParser headers files]))
 ])
 
-CXXFLAGS="${CXXFLAGS} -DMSGPACK_DISABLE_LEGACY_NIL -DMSGPACK_DISABLE_LEGACY_CONVERT"
+CXXFLAGS="${CXXFLAGS} -DMSGPACK_NO_BOOST -DMSGPACK_DISABLE_LEGACY_NIL -DMSGPACK_DISABLE_LEGACY_CONVERT"
 
 AC_ARG_ENABLE([tools], AS_HELP_STRING([--disable-tools],[Disable tools (CLI DHT node)]),,build_tools=yes)
 AM_CONDITIONAL(ENABLE_TOOLS, test x$build_tools == xyes)

include/opendht/crypto.h

@@ -541,8 +541,8 @@ struct OPENDHT_PUBLIC Certificate {
     void addRevocationList(RevocationList&&);
     void addRevocationList(std::shared_ptr<RevocationList>);
 
-    static Certificate generate(const PrivateKey& key, const std::string& name = "dhtnode", const Identity& ca = {}, bool is_ca = false);
-    static Certificate generate(const CertificateRequest& request, const Identity& ca);
+    static Certificate generate(const PrivateKey& key, const std::string& name = "dhtnode", const Identity& ca = {}, bool is_ca = false, int64_t validity = 0);
+    static Certificate generate(const CertificateRequest& request, const Identity& ca, int64_t validity = 0);
 
     gnutls_x509_crt_t getCopy() const {
         if (not cert)
@@ -592,12 +592,20 @@ struct OPENDHT_PUBLIC Certificate {
      */
     std::pair<std::string, Blob> generateOcspRequest(gnutls_x509_crt_t& issuer);
 
+    /**
+     * Change certificate's expiration
+     */
+    void setValidity(const Identity& ca, int64_t validity);
+    void setValidity(const PrivateKey& key, int64_t validity);
+
     gnutls_x509_crt_t cert {nullptr};
     std::shared_ptr<Certificate> issuer {};
     std::shared_ptr<OcspResponse> ocspResponse;
 private:
     Certificate(const Certificate&) = delete;
     Certificate& operator=(const Certificate&) = delete;
+    mutable InfoHash cachedId_ {};
+    mutable PkId cachedLongId_ {};
 
     struct crlNumberCmp {
         bool operator() (const std::shared_ptr<RevocationList>& lhs, const std::shared_ptr<RevocationList>& rhs) const {

include/opendht/dht.h

@@ -114,7 +114,7 @@ public:
 
     void addBootstrap(const std::string& host, const std::string& service) override {
         bootstrap_nodes.emplace_back(host, service);
-        onDisconnected();
+        startBootstrap();
     }
 
     void clearBootstrap() override {
@@ -356,6 +356,8 @@ private:
 
     static constexpr duration REANNOUNCE_MARGIN {std::chrono::seconds(10)};
 
+    static constexpr std::chrono::seconds BOOTSTRAP_PERIOD {10};
+
     static constexpr size_t TOKEN_SIZE {32};
 
     // internal structures
@@ -393,7 +395,7 @@ private:
     Kad dht6 {};
 
     std::vector<std::pair<std::string,std::string>> bootstrap_nodes {};
-    std::chrono::steady_clock::duration bootstrap_period {std::chrono::seconds(10)};
+    std::chrono::steady_clock::duration bootstrap_period {BOOTSTRAP_PERIOD};
     Sp<Scheduler::Job> bootstrapJob {};
 
     std::map<InfoHash, Storage> store;
@@ -480,6 +482,9 @@ private:
     void sendCachedPing(Bucket& b);
     bool bucketMaintenance(RoutingTable&);
     void dumpBucket(const Bucket& b, std::ostream& out) const;
+    void bootstrap();
+    void startBootstrap();
+    void stopBootstrap();
 
     // Nodes
     void onNewNode(const Sp<Node>& node, int confirm);
@@ -487,7 +492,6 @@ private:
     bool trySearchInsert(const Sp<Node>& node);
 
     // Searches
-
     inline SearchMap& searches(sa_family_t af) { return dht(af).searches; }
     inline const SearchMap& searches(sa_family_t af) const { return dht(af).searches; }
 
@@ -569,7 +573,8 @@ private:
      *
      * @param sr  The search to execute its operations.
      */
-    void searchStep(Sp<Search>);
+    void searchStep(std::weak_ptr<Search> ws);
+
     void searchSynchedNodeListen(const Sp<Search>&, SearchNode&);
 
     void dumpSearch(const Search& sr, std::ostream& out) const;

include/opendht/dht_proxy_client.h

@@ -337,6 +337,7 @@ private:
     NodeStats stats6_ {};
     SockAddr publicAddressV4_;
     SockAddr publicAddressV6_;
+    std::atomic_bool launchConnectedCbs_ {false};
 
     InfoHash myid {};
 

include/opendht/dhtrunner.h

@@ -456,8 +456,6 @@ public:
     void forwardAllMessages(bool forward);
 
 private:
-    static constexpr std::chrono::seconds BOOTSTRAP_PERIOD {10};
-
     enum class State {
         Idle,
         Running,

include/opendht/thread_pool.h

@@ -27,6 +27,8 @@
 #include <future>
 #include <functional>
 
+#include <ciso646> // fix windows compiler bug
+
 namespace dht {
 
 class OPENDHT_PUBLIC ThreadPool {
@@ -63,15 +65,14 @@ public:
     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_ {};
+    std::queue<std::function<void()>> tasks_ {};
+    std::vector<std::unique_ptr<std::thread>> threads_;
+    unsigned readyThreads_ {0};
+    bool running_ {true};
 
     const unsigned maxThreads_;
-    bool running_ {true};
 };
 
 class OPENDHT_PUBLIC Executor : public std::enable_shared_from_this<Executor> {

include/opendht/value.h

@@ -117,7 +117,7 @@ struct OPENDHT_PUBLIC ValueType {
 
     Id id {0};
     std::string name {};
-    duration expiration {60 * 10};
+    duration expiration {std::chrono::minutes(10)};
     StorePolicy storePolicy {DEFAULT_STORE_POLICY};
     EditPolicy editPolicy {DEFAULT_EDIT_POLICY};
 };

src/crypto.cpp

@@ -69,7 +69,7 @@ constexpr gnutls_digest_algorithm_t gnutlsHashAlgo(size_t min_res) {
                                GNUTLS_DIG_SHA1));
 }
 
-constexpr size_t gnutlsHashSize(int algo) {
+constexpr size_t gnutlsHashSize(gnutls_digest_algorithm_t algo) {
     return (algo == GNUTLS_DIG_SHA512) ? 512/8 : (
            (algo == GNUTLS_DIG_SHA256) ? 256/8 : (
            (algo == GNUTLS_DIG_SHA1)   ? 160/8 : 0 ));
@@ -853,12 +853,15 @@ Certificate::getId() const
 {
     if (not cert)
         return {};
+    if (cachedId_)
+        return cachedId_;
     InfoHash id;
     size_t sz = id.size();
     if (auto err = gnutls_x509_crt_get_key_id(cert, 0, id.data(), &sz))
         throw CryptoException(std::string("Can't get certificate public key ID: ") + gnutls_strerror(err));
     if (sz != id.size())
         throw CryptoException("Can't get certificate public key ID: wrong output length.");
+    cachedId_ = id;
     return id;
 }
 
@@ -867,6 +870,8 @@ Certificate::getLongId() const
 {
     if (not cert)
         return {};
+    if (cachedLongId_)
+        return cachedLongId_;
 #if GNUTLS_VERSION_NUMBER < 0x030401
     throw CryptoException("Can't get certificate 256 bits public key ID: GnuTLS 3.4.1 or higher required.");
 #else
@@ -876,6 +881,7 @@ Certificate::getLongId() const
         throw CryptoException(std::string("Can't get certificate 256 bits public key ID: ") + gnutls_strerror(err));
     if (sz != id.size())
         throw CryptoException("Can't get certificate 256 bits public key ID: wrong output length.");
+    cachedLongId_ = id;
     return id;
 #endif
 }
@@ -1157,14 +1163,14 @@ setRandomSerial(gnutls_x509_crt_t cert)
 }
 
 Certificate
-Certificate::generate(const PrivateKey& key, const std::string& name, const Identity& ca, bool is_ca)
+Certificate::generate(const PrivateKey& key, const std::string& name, const Identity& ca, bool is_ca, int64_t validity)
 {
     gnutls_x509_crt_t cert;
     if (not key.x509_key or gnutls_x509_crt_init(&cert) != GNUTLS_E_SUCCESS)
         return {};
     Certificate ret {cert};
 
-    setValidityPeriod(cert, 10 * 365 * 24 * 60 * 60);
+    setValidityPeriod(cert, validity <= 0 ? 10 * 365 * 24 * 60 * 60 : validity);
     if (int err = gnutls_x509_crt_set_key(cert, key.x509_key)) {
         throw CryptoException(std::string("Error when setting certificate key ") + gnutls_strerror(err));
     }
@@ -1210,7 +1216,7 @@ Certificate::generate(const PrivateKey& key, const std::string& name, const Iden
 }
 
 Certificate
-Certificate::generate(const CertificateRequest& request, const Identity& ca)
+Certificate::generate(const CertificateRequest& request, const Identity& ca, int64_t validity)
 {
     gnutls_x509_crt_t cert;
     if (auto err = gnutls_x509_crt_init(&cert))
@@ -1223,7 +1229,7 @@ Certificate::generate(const CertificateRequest& request, const Identity& ca)
         throw CryptoException(std::string("Can't set certificate version: ") + gnutls_strerror(err));
     }
 
-    setValidityPeriod(cert, 10 * 365 * 24 * 60 * 60);
+    setValidityPeriod(cert, validity <= 0 ? 10 * 365 * 24 * 60 * 60 : validity);
     setRandomSerial(cert);
 
     if (auto err = gnutls_x509_crt_privkey_sign(cert, ca.second->cert, ca.first->key, ca.second->getPreferredDigest(), 0)) {
@@ -1234,6 +1240,32 @@ Certificate::generate(const CertificateRequest& request, const Identity& ca)
     return ret.getPacked();
 }
 
+void
+Certificate::setValidity(const Identity& ca, int64_t validity)
+{
+    setValidityPeriod(cert, validity);
+    setRandomSerial(cert);
+    if (ca.first && ca.second) {
+        if (not ca.second->isCA()) {
+            throw CryptoException("Signing certificate must be CA");
+        }
+        if (int err = gnutls_x509_crt_privkey_sign(cert, ca.second->cert, ca.first->key, ca.second->getPreferredDigest(), 0)) {
+            throw CryptoException(std::string("Error when signing certificate ") + gnutls_strerror(err));
+        }
+    }
+}
+
+void
+Certificate::setValidity(const PrivateKey& key, int64_t validity)
+{
+    setValidityPeriod(cert, validity);
+    setRandomSerial(cert);
+    const auto& pk = key.getPublicKey();
+    if (int err = gnutls_x509_crt_privkey_sign(cert, cert, key.key, pk.getPreferredDigest(), 0)) {
+        throw CryptoException(std::string("Error when signing certificate ") + gnutls_strerror(err));
+    }
+}
+
 std::vector<std::shared_ptr<RevocationList>>
 Certificate::getRevocationLists() const
 {

src/dht.cpp

@@ -38,10 +38,12 @@ using namespace std::placeholders;
 
 constexpr std::chrono::minutes Dht::MAX_STORAGE_MAINTENANCE_EXPIRE_TIME;
 constexpr std::chrono::minutes Dht::SEARCH_EXPIRE_TIME;
+constexpr std::chrono::seconds Dht::BOOTSTRAP_PERIOD;
 constexpr duration Dht::LISTEN_EXPIRE_TIME;
 constexpr duration Dht::LISTEN_EXPIRE_TIME_PUBLIC;
 constexpr duration Dht::REANNOUNCE_MARGIN;
 static constexpr size_t MAX_REQUESTS_PER_SEC {8 * 1024};
+static constexpr duration BOOTSTRAP_PERIOD_MAX {std::chrono::hours(24)};
 
 NodeStatus
 Dht::updateStatus(sa_family_t af)
@@ -51,9 +53,9 @@ Dht::updateStatus(sa_family_t af)
     d.status = d.getStatus(scheduler.time());
     if (d.status != old) {
         auto& other = dht(af == AF_INET ? AF_INET6 : AF_INET);
-        if (other.status == NodeStatus::Disconnected && d.status == NodeStatus::Disconnected)
+        if (other.status == NodeStatus::Disconnected && d.status == NodeStatus::Disconnected) {
             onDisconnected();
-        else if (other.status == NodeStatus::Connected || d.status == NodeStatus::Connected) {
+        } else if (other.status == NodeStatus::Connected || d.status == NodeStatus::Connected) {
             onConnected();
         }
     }
@@ -286,7 +288,7 @@ Dht::searchNodeGetDone(const net::Request& req,
             if (srn->syncJob)
                 scheduler.edit(srn->syncJob, syncTime);
             else
-                srn->syncJob = scheduler.add(syncTime, std::bind(&Dht::searchStep, this, sr));
+                srn->syncJob = scheduler.add(syncTime, std::bind(&Dht::searchStep, this, ws));
         }
         onGetValuesDone(req.node, answer, sr, query);
     }
@@ -488,13 +490,18 @@ void Dht::searchSendAnnounceValue(const Sp<Search>& sr) {
             } catch (std::out_of_range&) { }
 
             auto next_refresh_time = now + getType(a.value->type).expiration;
+            auto& acked = sn->acked[a.value->id];
+            if (acked.refresh) {
+                acked.refresh->cancel();
+                acked.refresh.reset();
+            }
             /* only put the value if the node doesn't already have it */
             if (not hasValue or seq_no < a.value->seq) {
                 if (logger_)
                     logger_->d(sr->id, sn->node->id, "[search %s] [node %s] sending 'put' (vid: %016" PRIx64 ")",
                         sr->id.toString().c_str(), sn->node->toString().c_str(), a.value->id);
                 auto created = a.permanent ? time_point::max() : a.created;
-                sn->acked[a.value->id] = {
+                acked = {
                     network_engine.sendAnnounceValue(sn->node, sr->id, a.value, created, sn->token, onDone, onExpired),
                     next_refresh_time
                 };
@@ -502,7 +509,7 @@ void Dht::searchSendAnnounceValue(const Sp<Search>& sr) {
                 if (logger_)
                     logger_->w(sr->id, sn->node->id, "[search %s] [node %s] sending 'refresh' (vid: %016" PRIx64 ")",
                         sr->id.toString().c_str(), sn->node->toString().c_str(), a.value->id);
-                sn->acked[a.value->id] = {
+                acked = {
                     network_engine.sendRefreshValue(sn->node, sr->id, a.value->id, sn->token, onDone,
                     [this, ws, node=sn->node, v=a.value,
                      onDone,
@@ -534,17 +541,13 @@ void Dht::searchSendAnnounceValue(const Sp<Search>& sr) {
                         sr->id.toString().c_str(), sn->node->toString().c_str(), a.value->id);
                 auto ack_req = std::make_shared<net::Request>(net::Request::State::COMPLETED);
                 ack_req->reply_time = now;
-                sn->acked[a.value->id] = std::make_pair(std::move(ack_req), next_refresh_time);
+                acked = {std::move(ack_req), next_refresh_time};
 
                 /* step to clear announces */
                 scheduler.edit(sr->nextSearchStep, now);
             }
             if (a.permanent) {
-                scheduler.add(next_refresh_time - REANNOUNCE_MARGIN, [this,ws] {
-                    if (auto sr = ws.lock()) {
-                        searchStep(sr);
-                    }
-                });
+                acked.refresh = scheduler.add(next_refresh_time - REANNOUNCE_MARGIN, std::bind(&Dht::searchStep, this, ws));
             }
         }
     };
@@ -642,8 +645,8 @@ Dht::searchSynchedNodeListen(const Sp<Search>& sr, SearchNode& n)
                 if (auto sr = ws.lock()) {
                     scheduler.edit(sr->nextSearchStep, scheduler.time());
                     if (auto sn = sr->getNode(req.node)) {
-                        scheduler.add(sn->getListenTime(query, getListenExpiration()), std::bind(&Dht::searchStep, this, sr));
-                        sn->onListenSynced(query);
+                        auto job = scheduler.add(sn->getListenTime(query, getListenExpiration()), std::bind(&Dht::searchStep, this, ws));
+                        sn->onListenSynced(query, true, std::move(job));
                     }
                     onListenDone(req.node, answer, sr);
                 }
@@ -669,8 +672,9 @@ Dht::searchSynchedNodeListen(const Sp<Search>& sr, SearchNode& n)
 /* When a search is in progress, we periodically call search_step to send
    further requests. */
 void
-Dht::searchStep(Sp<Search> sr)
+Dht::searchStep(std::weak_ptr<Search> ws)
 {
+    auto sr = ws.lock();
     if (not sr or sr->expired or sr->done) return;
 
     const auto& now = scheduler.time();
@@ -823,7 +827,7 @@ Dht::search(const InfoHash& id, sa_family_t af, GetCallback gcb, QueryCallback q
         sr->expired = false;
         sr->nodes.clear();
         sr->nodes.reserve(SEARCH_NODES+1);
-        sr->nextSearchStep = scheduler.add(time_point::max(), std::bind(&Dht::searchStep, this, sr));
+        sr->nextSearchStep = scheduler.add(time_point::max(), std::bind(&Dht::searchStep, this, std::weak_ptr<Search>(sr)));
         if (logger_)
             logger_->w(id, "[search %s IPv%c] new search", id.toString().c_str(), (af == AF_INET) ? '4' : '6');
         if (search_id == 0)
@@ -1276,8 +1280,10 @@ Dht::storageStore(const InfoHash& id, const Sp<Value>& value, time_point created
     if (auto vs = store.first) {
         total_store_size += store.second.size_diff;
         total_values += store.second.values_diff;
+        if (vs->expiration_job)
+            vs->expiration_job->cancel();
         if (not permanent) {
-            scheduler.add(expiration, std::bind(&Dht::expireStorage, this, id));
+            vs->expiration_job = scheduler.add(expiration, std::bind(&Dht::expireStorage, this, id));
         }
         if (total_store_size > max_store_size) {
             expireStore();
@@ -1437,6 +1443,7 @@ Dht::connectivityChanged(sa_family_t af)
     reported_addr.erase(std::remove_if(reported_addr.begin(), reported_addr.end(), [&](const ReportedAddr& addr){
         return addr.second.getFamily() == af;
     }), reported_addr.end());
+    startBootstrap(); // will only happen if disconnected
 }
 
 void
@@ -1618,10 +1625,10 @@ Dht::dumpSearch(const Search& sr, std::ostream& out) const
                 out << "[";
                 for (const auto& a : sr.announce) {
                     auto ack = n.acked.find(a.value->id);
-                    if (ack == n.acked.end() or not ack->second.first) {
+                    if (ack == n.acked.end() or not ack->second.req) {
                         out << ' ';
                     } else {
-                        out << ack->second.first->getStateChar();
+                        out << ack->second.req->getStateChar();
                     }
                 }
                 out << "] ";
@@ -2018,11 +2025,7 @@ Dht::expire()
 void
 Dht::onConnected()
 {
-    if (bootstrapJob) {
-        bootstrapJob->cancel();
-        bootstrapJob.reset();
-    }
-    bootstrap_period = std::chrono::seconds(10);
+    stopBootstrap();
     auto callbacks = std::move(onConnectCallbacks_);
     while (not callbacks.empty()) {
         callbacks.front()();
@@ -2032,6 +2035,13 @@ Dht::onConnected()
 
 void
 Dht::onDisconnected()
+{
+    if (not bootstrapJob)
+        bootstrap();
+}
+
+void
+Dht::bootstrap()
 {
     if (dht4.status != NodeStatus::Disconnected || dht6.status != NodeStatus::Disconnected)
         return;
@@ -2052,8 +2062,25 @@ Dht::onDisconnected()
     }
     if (bootstrapJob)
         bootstrapJob->cancel();
-    bootstrapJob = scheduler.add(scheduler.time() + bootstrap_period, std::bind(&Dht::onDisconnected, this));
-    bootstrap_period *= 2;
+    bootstrapJob = scheduler.add(scheduler.time() + bootstrap_period, std::bind(&Dht::bootstrap, this));
+    bootstrap_period = std::min(bootstrap_period * 2, BOOTSTRAP_PERIOD_MAX);
+}
+
+void
+Dht::startBootstrap()
+{
+    stopBootstrap();
+    bootstrapJob = scheduler.add(scheduler.time(), std::bind(&Dht::bootstrap, this));
+}
+
+void
+Dht::stopBootstrap()
+{
+    if (bootstrapJob) {
+        bootstrapJob->cancel();
+        bootstrapJob.reset();
+    }
+    bootstrap_period = BOOTSTRAP_PERIOD;
 }
 
 void
@@ -2170,25 +2197,25 @@ Dht::exportNodes() const
     if (b4 != dht4.buckets.end()) {
         for (auto& n : b4->nodes)
             if (n->isGood(now))
-                nodes.push_back(n->exportNode());
+                nodes.emplace_back(n->exportNode());
     }
     const auto b6 = dht6.buckets.findBucket(myid);
     if (b6 != dht6.buckets.end()) {
         for (auto& n : b6->nodes)
             if (n->isGood(now))
-                nodes.push_back(n->exportNode());
+                nodes.emplace_back(n->exportNode());
     }
     for (auto b = dht4.buckets.begin(); b != dht4.buckets.end(); ++b) {
         if (b == b4) continue;
         for (auto& n : b->nodes)
             if (n->isGood(now))
-                nodes.push_back(n->exportNode());
+                nodes.emplace_back(n->exportNode());
     }
     for (auto b = dht6.buckets.begin(); b != dht6.buckets.end(); ++b) {
         if (b == b6) continue;
         for (auto& n : b->nodes)
             if (n->isGood(now))
-                nodes.push_back(n->exportNode());
+                nodes.emplace_back(n->exportNode());
     }
     return nodes;
 }
@@ -2461,7 +2488,7 @@ Dht::onAnnounce(Sp<Node> n,
             if (*lv == *vc) {
                 storageRefresh(hash, v->id);
                 if (logger_)
-                    logger_->d(hash, node.id, "[store %s] [node %s] refreshed value %s", hash.toString().c_str(), node.toString().c_str(), std::to_string(v->id).c_str());
+                    logger_->d(hash, node.id, "[store %s] [node %s] refreshed value %016" PRIx64, hash.toString().c_str(), node.toString().c_str(), v->id);
             } else {
                 const auto& type = getType(lv->type);
                 if (type.editPolicy(hash, lv, vc, node.id, node.getAddr())) {
@@ -2504,7 +2531,7 @@ Dht::onRefresh(Sp<Node> node, const InfoHash& hash, const Blob& token, const Val
     }
     if (storageRefresh(hash, vid)) {
         if (logger_)
-            logger_->d(hash, node->id, "[store %s] [node %s] refreshed value %s", hash.toString().c_str(), node->toString().c_str(), std::to_string(vid).c_str());
+            logger_->d(hash, node->id, "[store %s] [node %s] refreshed value %016" PRIx64, hash.toString().c_str(), node->toString().c_str(), vid);
     } else {
         if (logger_)
             logger_->d(hash, node->id, "[store %s] [node %s] got refresh for unknown value",
@@ -2540,8 +2567,15 @@ Dht::storageRefresh(const InfoHash& id, Value::Id vid)
         }
 
         auto expiration = s->second.refresh(now, vid, types);
-        if (expiration != time_point::max())
-            scheduler.add(expiration, std::bind(&Dht::expireStorage, this, id));
+        if (expiration.first) {
+            if (expiration.first->expiration_job) {
+                expiration.first->expiration_job->cancel();
+                expiration.first->expiration_job.reset();
+            }
+            if (expiration.second != time_point::max()) {
+                expiration.first->expiration_job = scheduler.add(expiration.second, std::bind(&Dht::expireStorage, this, id));
+            }
+        }
         return true;
     }
     return false;

src/dht_proxy_client.cpp

@@ -720,7 +720,8 @@ DhtProxyClient::onProxyInfos(const Json::Value& proxyInfos, const sa_family_t fa
     }
     auto newStatus = std::max(statusIpv4_, statusIpv6_);
     if (newStatus == NodeStatus::Connected) {
-        if (oldStatus == NodeStatus::Disconnected || oldStatus == NodeStatus::Connecting) {
+        if (oldStatus == NodeStatus::Disconnected || oldStatus == NodeStatus::Connecting || launchConnectedCbs_) {
+            launchConnectedCbs_ = false;
             listenerRestartTimer_->expires_at(std::chrono::steady_clock::now());
             listenerRestartTimer_->async_wait(std::bind(&DhtProxyClient::restartListeners, this, std::placeholders::_1));
             if (not onConnectCallbacks_.empty()) {
@@ -1160,6 +1161,9 @@ DhtProxyClient::pushNotificationReceived(const std::map<std::string, std::string
     {
         // If a push notification is received, the proxy is up and running
         std::lock_guard<std::mutex> l(lockCurrentProxyInfos_);
+        auto oldStatus = std::max(statusIpv4_, statusIpv6_);
+        if (oldStatus != NodeStatus::Connected)
+            launchConnectedCbs_ = true;
         statusIpv4_ = NodeStatus::Connected;
         statusIpv6_ = NodeStatus::Connected;
     }
@@ -1265,11 +1269,7 @@ DhtProxyClient::resubscribe(const InfoHash& key, const size_t token, Listener& l
         logger_->d("[proxy:client] [resubscribe] [search %s]", key.to_c_str());
 
     auto opstate = listener.opstate;
-    opstate->stop = true;
-    if (listener.request){
-        listener.request.reset();
-    }
-    opstate->stop = false;
+    listener.request.reset(); // This will update ok to false
     opstate->ok = true;
 
     restinio::http_request_header_t header;

src/dht_proxy_server.cpp

@@ -1004,7 +1004,13 @@ DhtProxyServer::sendPushNotification(const std::string& token, Json::Value&& jso
         notification["platform"] = type == PushType::Android ? 2 : 1;
         notification["data"] = std::move(json);
         notification["priority"] = highPriority ? "high" : "normal";
-        notification["time_to_live"] = 600;
+        if (type == PushType::Android)
+            notification["time_to_live"] = 3600 * 24; // 24 hours
+        else {
+            const auto expiration = std::chrono::system_clock::now() + std::chrono::hours(24);
+            uint32_t exp = std::chrono::duration_cast<std::chrono::seconds>(expiration.time_since_epoch()).count();
+            notification["expiration"] = exp;
+        }
 
         Json::Value notifications(Json::arrayValue);
         notifications[0] = notification;

src/dhtrunner.cpp

@@ -36,7 +36,6 @@
 
 namespace dht {
 
-constexpr std::chrono::seconds DhtRunner::BOOTSTRAP_PERIOD;
 static const std::string PEER_DISCOVERY_DHT_SERVICE = "dht";
 
 struct DhtRunner::Listener {
@@ -705,8 +704,8 @@ DhtRunner::loop_()
             rcv_free.splice(rcv_free.end(), std::move(received_treated));
     }
 
-    if (dropped)
-        std::cerr << "Dropped " << dropped << " packets with high delay" << std::endl;
+    if (dropped && logger_)
+        logger_->e("[runner %p] Dropped %zu packets with high delay.", this, dropped);
 
     NodeStatus nstatus4 = dht->updateStatus(AF_INET);
     NodeStatus nstatus6 = dht->updateStatus(AF_INET6);

src/search.h

@@ -49,12 +49,21 @@ struct Dht::Announce {
 };
 
 struct Dht::SearchNode {
+
+    struct AnnounceStatus {
+        Sp<net::Request> req {};
+        Sp<Scheduler::Job> refresh {};
+        time_point refresh_time;
+        AnnounceStatus(){};
+        AnnounceStatus(Sp<net::Request> r, time_point t): req(std::move(r)), refresh_time(t)
+         {}
+    };
     /**
      * Foreach value id, we keep track of a pair (net::Request, time_point) where the
      * request is the request returned by the network engine and the time_point
      * is the next time at which the value must be refreshed.
      */
-    using AnnounceStatus = std::map<Value::Id, std::pair<Sp<net::Request>, time_point>>;
+    using AnnounceStatusMap = std::map<Value::Id, AnnounceStatus>;
     /**
      * Foreach Query, we keep track of the request returned by the network
      * engine when we sent the "get".
@@ -63,6 +72,7 @@ struct Dht::SearchNode {
 
     struct CachedListenStatus {
         ValueCache cache;
+        Sp<Scheduler::Job> refresh {};
         Sp<Scheduler::Job> cacheExpirationJob {};
         Sp<net::Request> req {};
         Tid socketId {0};
@@ -87,7 +97,7 @@ struct Dht::SearchNode {
 
     SyncStatus getStatus {};    /* get/sync status */
     NodeListenerStatus listenStatus {}; /* listen status */
-    AnnounceStatus acked {};    /* announcement status for a given value id */
+    AnnounceStatusMap acked {};    /* announcement status for a given value id */
 
     Blob token {};                                 /* last token the node sent to us after a get request */
     time_point last_get_reply {time_point::min()}; /* last time received valid token */
@@ -239,9 +249,12 @@ struct Dht::SearchNode {
         }
     }
 
-    void onListenSynced(const Sp<Query>& q, bool synced = true) {
+    void onListenSynced(const Sp<Query>& q, bool synced = true, Sp<Scheduler::Job> refreshJob = {}) {
         auto l = listenStatus.find(q);
         if (l != listenStatus.end()) {
+            if (l->second.refresh)
+                l->second.refresh->cancel();
+            l->second.refresh = std::move(refreshJob);
             l->second.cache.onSynced(synced);
         }
     }
@@ -292,16 +305,18 @@ struct Dht::SearchNode {
 
     bool isAnnounced(Value::Id vid) const {
         auto ack = acked.find(vid);
-        if (ack == acked.end() or not ack->second.first)
+        if (ack == acked.end() or not ack->second.req)
             return false;
-        return ack->second.first->completed();
+        return ack->second.req->completed();
     }
     void cancelAnnounce() {
         for (const auto& status : acked) {
-            const auto& req = status.second.first;
+            const auto& req = status.second.req;
             if (req and req->pending()) {
                 node->cancelRequest(req);
             }
+            if (status.second.refresh)
+                status.second.refresh->cancel();
         }
         acked.clear();
     }
@@ -328,14 +343,21 @@ struct Dht::SearchNode {
         return listen_status->second.req->reply_time + listen_expire > now;
     }
     void cancelListen() {
-        for (const auto& status : listenStatus)
+        for (const auto& status : listenStatus) {
             node->cancelRequest(status.second.req);
-        listenStatus.clear();
+            if (status.second.refresh)
+                status.second.refresh->cancel();
+            if (status.second.cacheExpirationJob)
+                status.second.cacheExpirationJob->cancel();
+        }
+        listenStatus.clear(); 
     }
     void cancelListen(const Sp<Query>& query) {
         auto it = listenStatus.find(query);
         if (it != listenStatus.end()) {
             node->cancelRequest(it->second.req);
+            if (it->second.refresh)
+                it->second.refresh->cancel();
             listenStatus.erase(it);
         }
     }
@@ -345,13 +367,13 @@ struct Dht::SearchNode {
      */
     time_point getAnnounceTime(Value::Id vid) const {
         const auto& ack = acked.find(vid);
-        if (ack == acked.cend() or not ack->second.first) {
+        if (ack == acked.cend() or not ack->second.req) {
             return time_point::min();
         }
-        if (ack->second.first->completed()) {
-            return ack->second.second - REANNOUNCE_MARGIN;
+        if (ack->second.req->completed()) {
+            return ack->second.refresh_time - REANNOUNCE_MARGIN;
         }
-        return ack->second.first->pending() ? time_point::max() : time_point::min();
+        return ack->second.req->pending() ? time_point::max() : time_point::min();
     }
 
     /**
@@ -541,17 +563,15 @@ struct Dht::Search {
         if (not opExpirationJob)
             opExpirationJob = scheduler.add(time_point::max(), [this,&scheduler]{
                 auto nextExpire = cache.expire(scheduler.time(), [&](size_t t){
-                    Sp<Query> query;
                     const auto& ll = listeners.find(t);
                     if (ll != listeners.cend()) {
-                        query = ll->second.query;
+                        auto query = ll->second.query;
                         listeners.erase(ll);
-                    }
-                    for (auto& sn : nodes) {
-                        if (listeners.empty())
-                            sn->cancelListen();
-                        else if (query)
-                            sn->cancelListen(query);
+                        if (listeners.empty()) {
+                            for (auto& sn : nodes) sn->cancelListen();
+                        } else if (query) {
+                            for (auto& sn : nodes) sn->cancelListen(query);
+                        }
                     }
                 });
                 scheduler.edit(opExpirationJob, nextExpire);
@@ -588,8 +608,10 @@ struct Dht::Search {
         for (auto& n : nodes) {
             auto ackIt = n->acked.find(vid);
             if (ackIt != n->acked.end()) {
-                if (ackIt->second.first)
-                    ackIt->second.first->cancel();
+                if (ackIt->second.req)
+                    ackIt->second.req->cancel();
+                if (ackIt->second.refresh)
+                    ackIt->second.refresh->cancel();
                 n->acked.erase(ackIt);
             }
         }
@@ -606,7 +628,7 @@ struct Dht::Search {
             announce.emplace_back(Announce {permanent, value, created, callback});
             for (auto& n : nodes) {
                 n->probe_query.reset();
-                n->acked[value->id].first.reset();
+                n->acked[value->id].req.reset();
             }
         } else {
             a_sr->permanent = permanent;
@@ -614,7 +636,7 @@ struct Dht::Search {
             if (a_sr->value != value) {
                 a_sr->value = value;
                 for (auto& n : nodes) {
-                    n->acked[value->id].first.reset();
+                    n->acked[value->id].req.reset();
                     n->probe_query.reset();
                 }
             }
@@ -738,8 +760,8 @@ struct Dht::Search {
             for (auto& n : nodes) {
                 auto ackIt = n->acked.find(it->value->id);
                 if (ackIt != n->acked.end()) {
-                    if (ackIt->second.first)
-                        ackIt->second.first->cancel();
+                    if (ackIt->second.req)
+                        ackIt->second.req->cancel();
                     n->acked.erase(ackIt);
                 }
             }

src/storage.h

@@ -37,8 +37,7 @@ public:
         storedValues_.emplace(expiration, std::pair<InfoHash, Value::Id>(id, value.id));
     }
     void erase(const InfoHash& id, const Value& value, time_point expiration) {
-        auto size = value.size();
-        totalSize_ -= size;
+        totalSize_ -= value.size();
         auto range = storedValues_.equal_range(expiration);
         for (auto rit = range.first; rit != range.second;) {
             if (rit->second.first == id && rit->second.second == value.id) {
@@ -59,6 +58,7 @@ struct ValueStorage {
     Sp<Value> data {};
     time_point created {};
     time_point expiration {};
+    Sp<Scheduler::Job> expiration_job {};
     StorageBucket* store_bucket {nullptr};
 
     ValueStorage() {}
@@ -143,14 +143,15 @@ struct Storage {
      * @param vid  The value id
      * @return time of the next expiration, time_point::max() if no expiration
      */
-    time_point refresh(const time_point& now, const Value::Id& vid, const TypeStore& types) {
+    std::pair<ValueStorage*, time_point>
+    refresh(const time_point& now, const Value::Id& vid, const TypeStore& types) {
         for (auto& vs : values)
             if (vs.data->id == vid) {
                 vs.created = now;
                 vs.expiration = std::max(vs.expiration, now + types.getType(vs.data->type).expiration);
-                return vs.expiration;
+                return {&vs, vs.expiration};
             }
-        return time_point::max();
+        return {nullptr, time_point::max()};
     }
 
     size_t listen(ValueCallback& cb, Value::Filter& f, const Sp<Query>& q);
@@ -198,9 +199,9 @@ Storage::store(const InfoHash& id, const Sp<Value>& value, time_point created, t
     if (it != values.end()) {
         /* Already there, only need to refresh */
         it->created = created;
-        size_t size_old = it->data->size();
-        ssize_t size_diff = size_new - (ssize_t)size_old;
         if (it->data != value) {
+            size_t size_old = it->data->size();
+            ssize_t size_diff = size_new - (ssize_t)size_old;
             //DHT_LOG.DEBUG("Updating %s -> %s", id.toString().c_str(), value->toString().c_str());
             // clear quota for previous value
             if (it->store_bucket)
@@ -214,7 +215,6 @@ Storage::store(const InfoHash& id, const Sp<Value>& value, time_point created, t
             total_size += size_diff;
             return std::make_pair(&(*it), StoreDiff{size_diff, 0, 0});
         }
-        return std::make_pair(nullptr, StoreDiff{});
     } else {
         //DHT_LOG.DEBUG("Storing %s -> %s", id.toString().c_str(), value->toString().c_str());
         if (values.size() < MAX_VALUES) {
@@ -225,8 +225,8 @@ Storage::store(const InfoHash& id, const Sp<Value>& value, time_point created, t
                 sb->insert(id, *value, expiration);
             return std::make_pair(&values.back(), StoreDiff{size_new, 1, 0});
         }
-        return std::make_pair(nullptr, StoreDiff{});
     }
+    return std::make_pair(nullptr, StoreDiff{});
 }
 
 Storage::StoreDiff
@@ -240,6 +240,8 @@ Storage::remove(const InfoHash& id, Value::Id vid)
     ssize_t size = it->data->size();
     if (it->store_bucket)
         it->store_bucket->erase(id, *it->data, it->expiration);
+    if (it->expiration_job)
+        it->expiration_job->cancel();
     total_size -= size;
     values.erase(it);
     return {-size, -1, 0};
@@ -283,11 +285,13 @@ Storage::expire(const InfoHash& id, time_point now)
     });
     std::vector<Sp<Value>> ret;
     ret.reserve(std::distance(r, values.end()));
-    ssize_t size_diff {};
+    ssize_t size_diff {0};
     std::for_each(r, values.end(), [&](const ValueStorage& v) {
         size_diff -= v.data->size();
         if (v.store_bucket)
             v.store_bucket->erase(id, *v.data, v.expiration);
+        if (v.expiration_job)
+            v.expiration_job->cancel();
         ret.emplace_back(std::move(v.data));
     });
     total_size += size_diff;

src/thread_pool.cpp

@@ -28,12 +28,6 @@ namespace dht {
 
 constexpr const size_t IO_THREADS_MAX {64};
 
-struct ThreadPool::ThreadState
-{
-    std::thread thread {};
-    std::atomic_bool run {true};
-};
-
 ThreadPool&
 ThreadPool::computation()
 {
@@ -67,14 +61,12 @@ void
 ThreadPool::run(std::function<void()>&& cb)
 {
     std::unique_lock<std::mutex> l(lock_);
-    if (not running_) return;
+    if (not cb or 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) {
+        threads_.emplace_back(std::make_unique<std::thread>([this]() {
+            while (true) {
                 std::function<void()> task;
 
                 // pick task from queue
@@ -82,10 +74,10 @@ ThreadPool::run(std::function<void()>&& cb)
                     std::unique_lock<std::mutex> l(lock_);
                     readyThreads_++;
                     cv_.wait(l, [&](){
-                        return not t.run or not tasks_.empty();
+                        return not running_ or not tasks_.empty();
                     });
                     readyThreads_--;
-                    if (not t.run)
+                    if (not running_)
                         break;
                     task = std::move(tasks_.front());
                     tasks_.pop();
@@ -93,14 +85,13 @@ ThreadPool::run(std::function<void()>&& cb)
 
                 // run task
                 try {
-                    if (task)
-                        task();
+                    task();
                 } catch (const std::exception& e) {
                     // LOG_ERR("Exception running task: %s", e.what());
                     std::cerr << "Exception running task: " << e.what() << std::endl;
                 }
             }
-        });
+        }));
     }
 
     // push task to queue
@@ -113,12 +104,8 @@ ThreadPool::run(std::function<void()>&& cb)
 void
 ThreadPool::stop()
 {
-    {
-        std::lock_guard<std::mutex> l(lock_);
-        running_ = false;
-    }
-    for (auto& t : threads_)
-        t->run = false;
+    std::lock_guard<std::mutex> l(lock_);
+    running_ = false;
     cv_.notify_all();
 }
 
@@ -127,7 +114,7 @@ ThreadPool::join()
 {
     stop();
     for (auto& t : threads_)
-        t->thread.join();
+        t->join();
     threads_.clear();
 }
 
@@ -147,7 +134,7 @@ Executor::run_(std::function<void()>&& task)
 {
     current_++;
     std::weak_ptr<Executor> w = shared_from_this();
-    threadPool_.get().run([w,task] {
+    threadPool_.get().run([w,task = std::move(task)] {
         try {
             task();
         } catch (const std::exception& e) {

tests/threadpooltester.cpp

@@ -21,6 +21,7 @@
 
 #include "opendht/thread_pool.h"
 #include <atomic>
+#include <thread>
 
 namespace test {
 CPPUNIT_TEST_SUITE_REGISTRATION(ThreadPoolTester);
@@ -47,7 +48,7 @@ ThreadPoolTester::testThreadPool() {
         std::this_thread::sleep_for(std::chrono::milliseconds(10));
 
     pool.join();
-    CPPUNIT_ASSERT(count.load() == N);
+    CPPUNIT_ASSERT_EQUAL(N, count.load());
 }
 
 void

tools/CMakeLists.txt

@@ -30,6 +30,7 @@ endif ()
 if (OPENDHT_C)
     add_executable (dhtcnode dhtcnode.c)
     target_link_libraries (dhtcnode LINK_PUBLIC opendht-c ${READLINE_LIBRARIES})
+    target_include_directories (dhtcnode SYSTEM PRIVATE ${PROJECT_SOURCE_DIR}/c)
 endif ()
 
 if (NOT DEFINED CMAKE_INSTALL_BINDIR)

tools/dhtcnode.c

@@ -1,4 +1,4 @@
-#include <c/opendht_c.h>
+#include <opendht_c.h>
 
 #include <stdio.h>
 #include <string.h>
@@ -88,7 +88,7 @@ int main()
 
     // Put data
     const char* data_str = "yo, this is some data";
-    dht_value* val = dht_value_new(data_str, strlen(data_str));
+    dht_value* val = dht_value_new_from_string(data_str);
     dht_runner_put(runner, &h, val, dht_done_callback, runner, false);
     dht_value_unref(val);
 

tools/perftest.cpp

@@ -102,7 +102,7 @@ benchPingPong(unsigned netSize, unsigned n_parallel) {
 
     auto start = clock::now();
 
-    for (unsigned i=0; i<n_parallel; i++) 
+    for (unsigned i=0; i<n_parallel; i++)
         ping(pong_node, locs[i].first);
 
     {