dht: add max bootstrap period

Also starts a bootstrap if disconnected.

CMakeLists.txt

@@ -1,16 +1,4 @@
-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()
+cmake_minimum_required (VERSION 3.1)
 project (opendht)
 
 include(CMakePackageConfigHelpers)
@@ -19,7 +7,7 @@ include(FindPkgConfig)
 include(cmake/CheckAtomic.cmake)
 
 set (opendht_VERSION_MAJOR 2)
-set (opendht_VERSION_MINOR 3.5)
+set (opendht_VERSION_MINOR 3.2)
 set (opendht_VERSION ${opendht_VERSION_MAJOR}.${opendht_VERSION_MINOR})
 set (PACKAGE_VERSION ${opendht_VERSION})
 set (VERSION "${opendht_VERSION}")
@@ -57,7 +45,7 @@ if (NOT MSVC)
     find_package (PkgConfig REQUIRED)
     find_package (GnuTLS 3.3 REQUIRED)
     pkg_search_module (Nettle REQUIRED nettle)
-    find_package (msgpack REQUIRED)
+    find_package (Msgpack 1.2 REQUIRED)
     if (OPENDHT_TOOLS)
         find_package (Readline 6 REQUIRED)
     endif ()
@@ -143,7 +131,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_NO_BOOST -DMSGPACK_DISABLE_LEGACY_NIL -DMSGPACK_DISABLE_LEGACY_CONVERT")
+set (CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -DMSGPACK_DISABLE_LEGACY_NIL -DMSGPACK_DISABLE_LEGACY_CONVERT")
 
 if (NOT CMAKE_BUILD_TYPE)
     set(CMAKE_BUILD_TYPE Release)
@@ -428,7 +416,6 @@ 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,20 +97,10 @@ 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)));
 }
@@ -257,7 +247,7 @@ void dht_runner_config_default(dht_runner_config* config) {
 }
 
 // dht::DhtRunner
-dht_runner* dht_runner_new(void) {
+dht_runner* dht_runner_new() {
     return reinterpret_cast<dht_runner*>(new dht::DhtRunner);
 }
 

c/opendht_c.h

@@ -82,7 +82,6 @@ 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);
@@ -90,7 +89,6 @@ 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);
@@ -135,7 +133,7 @@ OPENDHT_C_PUBLIC void dht_runner_config_default(dht_runner_config* config);
 // dht::DhtRunner
 struct OPENDHT_C_PUBLIC dht_runner;
 typedef struct dht_runner dht_runner;
-OPENDHT_C_PUBLIC dht_runner* dht_runner_new(void);
+OPENDHT_C_PUBLIC dht_runner* dht_runner_new();
 OPENDHT_C_PUBLIC void dht_runner_delete(dht_runner* runner);
 OPENDHT_C_PUBLIC void dht_runner_run(dht_runner* runner, in_port_t port);
 OPENDHT_C_PUBLIC void dht_runner_run_config(dht_runner* runner, in_port_t port, const dht_runner_config* config);

cmake/CheckAtomic.cmake

@@ -1,106 +1,95 @@
-# atomic builtins are required for threading support.
+# University of Illinois/NCSA
+# Open Source License
+#
+# Copyright (c) 2003-2017 University of Illinois at Urbana-Champaign.
+# All rights reserved.
+#
+# Developed by:
+#
+#     LLVM Team
+#
+#     University of Illinois at Urbana-Champaign
+#
+#     http://llvm.org
+#
+# Permission is hereby granted, free of charge, to any person obtaining a copy of
+# this software and associated documentation files (the "Software"), to deal with
+# the Software without restriction, including without limitation the rights to
+# use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies
+# of the Software, and to permit persons to whom the Software is furnished to do
+# so, subject to the following conditions:
+#
+#     * Redistributions of source code must retain the above copyright notice,
+#       this list of conditions and the following disclaimers.
+#
+#     * Redistributions in binary form must reproduce the above copyright notice,
+#       this list of conditions and the following disclaimers in the
+#       documentation and/or other materials provided with the distribution.
+#
+#     * Neither the names of the LLVM Team, University of Illinois at
+#       Urbana-Champaign, nor the names of its contributors may be used to
+#       endorse or promote products derived from this Software without specific
+#       prior written permission.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
+# FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.  IN NO EVENT SHALL THE
+# CONTRIBUTORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS WITH THE
+# SOFTWARE.
 
-INCLUDE(CheckCXXSourceCompiles)
-INCLUDE(CheckLibraryExists)
+include(CheckCXXSourceCompiles)
+include(CheckLibraryExists)
 
 # Sometimes linking against libatomic is required for atomic ops, if
 # the platform doesn't support lock-free atomics.
 
 function(check_working_cxx_atomics varname)
   set(OLD_CMAKE_REQUIRED_FLAGS ${CMAKE_REQUIRED_FLAGS})
-  set(CMAKE_REQUIRED_FLAGS "${CMAKE_REQUIRED_FLAGS} -std=c++11")
-  CHECK_CXX_SOURCE_COMPILES("
-#include <atomic>
-std::atomic<int> x;
-int main() {
-  return x;
-}
-" ${varname})
-  set(CMAKE_REQUIRED_FLAGS ${OLD_CMAKE_REQUIRED_FLAGS})
-endfunction(check_working_cxx_atomics)
-
-function(check_working_cxx_atomics64 varname)
-  set(OLD_CMAKE_REQUIRED_FLAGS ${CMAKE_REQUIRED_FLAGS})
-  set(CMAKE_REQUIRED_FLAGS "-std=c++11 ${CMAKE_REQUIRED_FLAGS}")
+  get_directory_property(compile_options COMPILE_OPTIONS)
+  set(CMAKE_REQUIRED_FLAGS ${compile_options})
   CHECK_CXX_SOURCE_COMPILES("
 #include <atomic>
 #include <cstdint>
-std::atomic<uint64_t> x (0);
-int main() {
-  uint64_t i = x.load(std::memory_order_relaxed);
-  return 0;
-}
-" ${varname})
+std::atomic<int64_t> v1;
+std::atomic<int64_t> v2;
+int main(int, char**) {
+  return v1 + v2;
+}" ${varname})
   set(CMAKE_REQUIRED_FLAGS ${OLD_CMAKE_REQUIRED_FLAGS})
-endfunction(check_working_cxx_atomics64)
+endfunction(check_working_cxx_atomics)
 
 
+if(NOT DEFINED PROXYGEN_COMPILER_IS_GCC_COMPATIBLE)
+  if(CMAKE_COMPILER_IS_GNUCXX)
+    set(PROXYGEN_COMPILER_IS_GCC_COMPATIBLE ON)
+  elseif(MSVC)
+    set(PROXYGEN_COMPILER_IS_GCC_COMPATIBLE OFF)
+  elseif("${CMAKE_CXX_COMPILER_ID}" MATCHES "Clang")
+    set(PROXYGEN_COMPILER_IS_GCC_COMPATIBLE ON)
+  elseif("${CMAKE_CXX_COMPILER_ID}" MATCHES "Intel")
+    set(PROXYGEN_COMPILER_IS_GCC_COMPATIBLE ON)
+  endif()
+endif()
+
 # This isn't necessary on MSVC, so avoid command-line switch annoyance
 # by only running on GCC-like hosts.
-if (LLVM_COMPILER_IS_GCC_COMPATIBLE)
+if(PROXYGEN_COMPILER_IS_GCC_COMPATIBLE)
   # First check if atomics work without the library.
   check_working_cxx_atomics(HAVE_CXX_ATOMICS_WITHOUT_LIB)
   # If not, check if the library exists, and atomics work with it.
   if(NOT HAVE_CXX_ATOMICS_WITHOUT_LIB)
     check_library_exists(atomic __atomic_fetch_add_4 "" HAVE_LIBATOMIC)
-    if( HAVE_LIBATOMIC )
+    if(HAVE_LIBATOMIC)
       list(APPEND CMAKE_REQUIRED_LIBRARIES "atomic")
       check_working_cxx_atomics(HAVE_CXX_ATOMICS_WITH_LIB)
       if (NOT HAVE_CXX_ATOMICS_WITH_LIB)
 	message(FATAL_ERROR "Host compiler must support std::atomic!")
       endif()
+      list(APPEND CMAKE_CXX_STANDARD_LIBRARIES -latomic)
     else()
       message(FATAL_ERROR "Host compiler appears to require libatomic, but cannot find it.")
     endif()
   endif()
 endif()
-
-# Check for 64 bit atomic operations.
-if(MSVC)
-  set(HAVE_CXX_ATOMICS64_WITHOUT_LIB True)
-else()
-  check_working_cxx_atomics64(HAVE_CXX_ATOMICS64_WITHOUT_LIB)
-endif()
-
-# If not, check if the library exists, and atomics work with it.
-if(NOT HAVE_CXX_ATOMICS64_WITHOUT_LIB)
-  check_library_exists(atomic __atomic_load_8 "" HAVE_CXX_LIBATOMICS64)
-  if(HAVE_CXX_LIBATOMICS64)
-    list(APPEND CMAKE_REQUIRED_LIBRARIES "atomic")
-    check_working_cxx_atomics64(HAVE_CXX_ATOMICS64_WITH_LIB)
-    if (NOT HAVE_CXX_ATOMICS64_WITH_LIB)
-      message(FATAL_ERROR "Host compiler must support 64-bit std::atomic!")
-    endif()
-  else()
-    message(FATAL_ERROR "Host compiler appears to require libatomic for 64-bit operations, but cannot find it.")
-  endif()
-endif()
-
-## TODO: This define is only used for the legacy atomic operations in
-## llvm's Atomic.h, which should be replaced.  Other code simply
-## assumes C++11 <atomic> works.
-CHECK_CXX_SOURCE_COMPILES("
-#ifdef _MSC_VER
-#include <windows.h>
-#endif
-int main() {
-#ifdef _MSC_VER
-        volatile LONG val = 1;
-        MemoryBarrier();
-        InterlockedCompareExchange(&val, 0, 1);
-        InterlockedIncrement(&val);
-        InterlockedDecrement(&val);
-#else
-        volatile unsigned long val = 1;
-        __sync_synchronize();
-        __sync_val_compare_and_swap(&val, 1, 0);
-        __sync_add_and_fetch(&val, 1);
-        __sync_sub_and_fetch(&val, 1);
-#endif
-        return 0;
-      }
-" LLVM_HAS_ATOMICS)
-
-if( NOT LLVM_HAS_ATOMICS )
-  message(STATUS "Warning: LLVM will be built thread-unsafe because atomic builtins are missing")
-endif()

cmake/FindMsgpack.cmake

@@ -0,0 +1,64 @@
+# - 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], 5)
+m4_define([opendht_patch_version], 2)
 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_NO_BOOST -DMSGPACK_DISABLE_LEGACY_NIL -DMSGPACK_DISABLE_LEGACY_CONVERT"
+CXXFLAGS="${CXXFLAGS} -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/dht.h

@@ -573,8 +573,7 @@ private:
      *
      * @param sr  The search to execute its operations.
      */
-    void searchStep(std::weak_ptr<Search> ws);
-
+    void searchStep(Sp<Search>);
     void searchSynchedNodeListen(const Sp<Search>&, SearchNode&);
 
     void dumpSearch(const Search& sr, std::ostream& out) const;

include/opendht/infohash.h

@@ -44,7 +44,6 @@ typedef uint16_t in_port_t;
 #include <stdexcept>
 #include <sstream>
 #include <cstring>
-#include <cstddef>
 
 namespace dht {
 

include/opendht/scheduler.h

@@ -36,9 +36,8 @@ namespace dht {
 class Scheduler {
 public:
     struct Job {
-        Job(std::function<void()>&& f, time_point t) : do_(std::move(f)), t_(t) {}
+        Job(std::function<void()>&& f) : do_(std::move(f)) {}
         std::function<void()> do_;
-        const time_point t_;
         void cancel() { do_ = {}; }
     };
 
@@ -51,12 +50,17 @@ public:
      * @return pointer to the newly scheduled job.
      */
     Sp<Scheduler::Job> add(time_point t, std::function<void()>&& job_func) {
-        auto job = std::make_shared<Job>(std::move(job_func), t);
+        auto job = std::make_shared<Job>(std::move(job_func));
         if (t != time_point::max())
             timers.emplace(std::move(t), job);
         return job;
     }
 
+    void add(const Sp<Scheduler::Job>& job, time_point t) {
+        if (t != time_point::max())
+            timers.emplace(std::move(t), job);
+    }
+
     /**
      * Reschedules a job.
      *
@@ -64,29 +68,16 @@ public:
      * @param t  The time at which the job shall be rescheduled.
      */
     void edit(Sp<Scheduler::Job>& job, time_point t) {
-        if (not job)
+        if (not job) {
             return;
+        }
         // std::function move doesn't garantee to leave the object empty.
         // Force clearing old value.
         auto task = std::move(job->do_);
-        cancel(job);
+        job->do_ = {};
         job = add(t, std::move(task));
     }
 
-    bool cancel(Sp<Scheduler::Job>& job) {
-        if (job) {
-            job->cancel();
-            for (auto r = timers.equal_range(job->t_); r.first != r.second; ++r.first) {
-                if (r.first->second == job) {
-                    timers.erase(r.first);
-                    job.reset();
-                    return true;
-                }
-            }
-        }
-        return false;
-    }
-
     /**
      * Runs the jobs to do up to now.
      *

include/opendht/thread_pool.h

@@ -65,14 +65,15 @@ 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 {std::chrono::minutes(10)};
+    duration expiration {60 * 10};
     StorePolicy storePolicy {DEFAULT_STORE_POLICY};
     EditPolicy editPolicy {DEFAULT_EDIT_POLICY};
 };

src/dht.cpp

@@ -288,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, ws));
+                srn->syncJob = scheduler.add(syncTime, std::bind(&Dht::searchStep, this, sr));
         }
         onGetValuesDone(req.node, answer, sr, query);
     }
@@ -490,15 +490,13 @@ 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];
-            scheduler.cancel(acked.refresh);
             /* 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;
-                acked = {
+                sn->acked[a.value->id] = {
                     network_engine.sendAnnounceValue(sn->node, sr->id, a.value, created, sn->token, onDone, onExpired),
                     next_refresh_time
                 };
@@ -506,7 +504,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);
-                acked = {
+                sn->acked[a.value->id] = {
                     network_engine.sendRefreshValue(sn->node, sr->id, a.value->id, sn->token, onDone,
                     [this, ws, node=sn->node, v=a.value,
                      onDone,
@@ -538,13 +536,17 @@ 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;
-                acked = {std::move(ack_req), next_refresh_time};
+                sn->acked[a.value->id] = std::make_pair(std::move(ack_req), next_refresh_time);
 
                 /* step to clear announces */
                 scheduler.edit(sr->nextSearchStep, now);
             }
             if (a.permanent) {
-                acked.refresh = scheduler.add(next_refresh_time - REANNOUNCE_MARGIN, std::bind(&Dht::searchStep, this, ws));
+                scheduler.add(next_refresh_time - REANNOUNCE_MARGIN, [this,ws] {
+                    if (auto sr = ws.lock()) {
+                        searchStep(sr);
+                    }
+                });
             }
         }
     };
@@ -642,8 +644,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)) {
-                        auto job = scheduler.add(sn->getListenTime(query, getListenExpiration()), std::bind(&Dht::searchStep, this, ws));
-                        sn->onListenSynced(query, true, std::move(job));
+                        scheduler.add(sn->getListenTime(query, getListenExpiration()), std::bind(&Dht::searchStep, this, sr));
+                        sn->onListenSynced(query);
                     }
                     onListenDone(req.node, answer, sr);
                 }
@@ -669,9 +671,8 @@ 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(std::weak_ptr<Search> ws)
+Dht::searchStep(Sp<Search> sr)
 {
-    auto sr = ws.lock();
     if (not sr or sr->expired or sr->done) return;
 
     const auto& now = scheduler.time();
@@ -824,7 +825,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, std::weak_ptr<Search>(sr)));
+        sr->nextSearchStep = scheduler.add(time_point::max(), std::bind(&Dht::searchStep, this, 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)
@@ -1277,9 +1278,8 @@ 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;
-        scheduler.cancel(vs->expiration_job);
         if (not permanent) {
-            vs->expiration_job = scheduler.add(expiration, std::bind(&Dht::expireStorage, this, id));
+            scheduler.add(expiration, std::bind(&Dht::expireStorage, this, id));
         }
         if (total_store_size > max_store_size) {
             expireStore();
@@ -1621,10 +1621,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.req) {
+                    if (ack == n.acked.end() or not ack->second.first) {
                         out << ' ';
                     } else {
-                        out << ack->second.req->getStateChar();
+                        out << ack->second.first->getStateChar();
                     }
                 }
                 out << "] ";
@@ -2056,7 +2056,8 @@ Dht::bootstrap()
                 logger_->e(myid, "Can't resolve %s:%s: %s", boootstrap.first.c_str(), boootstrap.second.c_str(), e.what());
         }
     }
-    scheduler.cancel(bootstrapJob);
+    if (bootstrapJob)
+        bootstrapJob->cancel();
     bootstrapJob = scheduler.add(scheduler.time() + bootstrap_period, std::bind(&Dht::bootstrap, this));
     bootstrap_period = std::min(bootstrap_period * 2, BOOTSTRAP_PERIOD_MAX);
 }
@@ -2071,7 +2072,10 @@ Dht::startBootstrap()
 void
 Dht::stopBootstrap()
 {
-    scheduler.cancel(bootstrapJob);
+    if (bootstrapJob) {
+        bootstrapJob->cancel();
+        bootstrapJob.reset();
+    }
     bootstrap_period = BOOTSTRAP_PERIOD;
 }
 
@@ -2189,25 +2193,25 @@ Dht::exportNodes() const
     if (b4 != dht4.buckets.end()) {
         for (auto& n : b4->nodes)
             if (n->isGood(now))
-                nodes.emplace_back(n->exportNode());
+                nodes.push_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.emplace_back(n->exportNode());
+                nodes.push_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.emplace_back(n->exportNode());
+                nodes.push_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.emplace_back(n->exportNode());
+                nodes.push_back(n->exportNode());
     }
     return nodes;
 }
@@ -2480,7 +2484,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 %016" PRIx64, hash.toString().c_str(), node.toString().c_str(), v->id);
+                    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());
             } else {
                 const auto& type = getType(lv->type);
                 if (type.editPolicy(hash, lv, vc, node.id, node.getAddr())) {
@@ -2523,7 +2527,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 %016" PRIx64, hash.toString().c_str(), node->toString().c_str(), vid);
+            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());
     } else {
         if (logger_)
             logger_->d(hash, node->id, "[store %s] [node %s] got refresh for unknown value",
@@ -2559,12 +2563,8 @@ Dht::storageRefresh(const InfoHash& id, Value::Id vid)
         }
 
         auto expiration = s->second.refresh(now, vid, types);
-        if (expiration.first) {
-            scheduler.cancel(expiration.first->expiration_job);
-            if (expiration.second != time_point::max()) {
-                expiration.first->expiration_job = scheduler.add(expiration.second, std::bind(&Dht::expireStorage, this, id));
-            }
-        }
+        if (expiration != time_point::max())
+            scheduler.add(expiration, std::bind(&Dht::expireStorage, this, id));
         return true;
     }
     return false;

src/dhtrunner.cpp

@@ -105,96 +105,81 @@ DhtRunner::run(const Config& config, Context&& context)
     auto expected = State::Idle;
     if (not running.compare_exchange_strong(expected, State::Running)) {
         if (context.logger)
-            context.logger->w("[runner %p] Node is already running. Call join() first before calling run() again.", this);
+            context.logger->w("[runner %p] Node is already running. Call join() first before calling run() again.");
         return;
     }
 
-    try {
-        auto local4 = config.bind4;
-        auto local6 = config.bind6;
-        if (not local4 and not local6) {
-            if (context.logger)
-                context.logger->w("[runner %p] No address to bind specified in the configuration, using default addresses", this);
-            local4.setFamily(AF_INET);
-            local6.setFamily(AF_INET6);
-        }
-        auto state_path = config.dht_config.node_config.persist_path;
-        if (not state_path.empty() && (local4.getPort() == 0 || local6.getPort() == 0)) {
-            state_path += "_port.txt";
-            std::ifstream inConfig(state_path);
-            if (inConfig.is_open()) {
-                in_port_t port;
-                if (inConfig >> port) {
-                    if (local4.getPort() == 0) {
-                        if (context.logger)
-                            context.logger->d("[runner %p] Using IPv4 port %hu from saved configuration", this, port);
-                        local4.setPort(port);
-                    }
+    auto local4 = config.bind4;
+    auto local6 = config.bind6;
+    if (not local4 and not local6) {
+        if (context.logger)
+            context.logger->w("[runner %p] No address to bind specified in the configuration, using default addresses");
+        local4.setFamily(AF_INET);
+        local6.setFamily(AF_INET6);
+    }
+    auto state_path = config.dht_config.node_config.persist_path;
+    if (not state_path.empty() && (local4.getPort() == 0 || local6.getPort() == 0)) {
+        state_path += "_port.txt";
+        std::ifstream inConfig(state_path);
+        if (inConfig.is_open()) {
+            in_port_t port;
+            if (inConfig >> port) {
+                if (local4.getPort() == 0) {
+                    if (context.logger)
+                        context.logger->d("[runner %p] Using IPv4 port %hu from saved configuration", this, port);
+                    local4.setPort(port);
                 }
-                if (inConfig >> port) {
-                    if (local6.getPort() == 0) {
-                        if (context.logger)
-                            context.logger->d("[runner %p] Using IPv6 port %hu from saved configuration", this, port);
-                        local6.setPort(port);
-                    }
+            }
+            if (inConfig >> port) {
+                if (local6.getPort() == 0) {
+                    if (context.logger)
+                        context.logger->d("[runner %p] Using IPv6 port %hu from saved configuration", this, port);
+                    local6.setPort(port);
                 }
             }
         }
-
-        if (context.logger) {
-            logger_ = context.logger;
-        }
-
-        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;
-        }
-
-        if (context.logger) {
-            logger_->d("[runner %p] state changed to Running", this);
-        }
-
-        context.sock->setOnReceive([&] (net::PacketList&& pkts) {
-            net::PacketList ret;
-            {
-                std::lock_guard<std::mutex> lck(sock_mtx);
-                auto maxSize = net::RX_QUEUE_MAX_SIZE - pkts.size();
-                while (rcv.size() > maxSize) {
-                    if (logger_)
-                        logger_->e("Dropping packet: queue is full!");
-                    rcv.pop_front();
-                }
-
-                rcv.splice(rcv.end(), std::move(pkts));
-                ret = std::move(rcv_free);
-            }
-            cv.notify_all();
-            return ret;
-        });
-
-#ifdef OPENDHT_PROXY_CLIENT
-        config_ = config;
-        identityAnnouncedCb_ = context.identityAnnouncedCb;
-#endif
-        auto dht = std::unique_ptr<DhtInterface>(new Dht(std::move(context.sock), SecureDht::getConfig(config.dht_config), context.logger));
-        dht_ = std::unique_ptr<SecureDht>(new SecureDht(std::move(dht), config.dht_config, std::move(context.identityAnnouncedCb), context.logger));
-        enableProxy(not config.proxy_server.empty());
-    } catch(const std::exception& e) {
-        config_ = {};
-        identityAnnouncedCb_ = {};
-        dht_.reset();
-#ifdef OPENDHT_PROXY_CLIENT
-        dht_via_proxy_.reset();
-#endif
-        running = State::Idle;
-        throw std::runtime_error(std::string("Error starting the node: ") + e.what());
     }
 
+    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;
+    }
+
+    if (context.logger) {
+        logger_ = context.logger;
+        logger_->d("[runner %p] state changed to Running", this);
+    }
+
+    context.sock->setOnReceive([&] (net::PacketList&& pkts) {
+        net::PacketList ret;
+        {
+            std::lock_guard<std::mutex> lck(sock_mtx);
+            auto maxSize = net::RX_QUEUE_MAX_SIZE - pkts.size();
+            while (rcv.size() > maxSize) {
+                if (logger_)
+                    logger_->e("Dropping packet: queue is full!");
+                rcv.pop_front();
+            }
+
+            rcv.splice(rcv.end(), std::move(pkts));
+            ret = std::move(rcv_free);
+        }
+        cv.notify_all();
+        return ret;
+    });
+
+#ifdef OPENDHT_PROXY_CLIENT
+    config_ = config;
+    identityAnnouncedCb_ = context.identityAnnouncedCb;
+#endif
+    auto dht = std::unique_ptr<DhtInterface>(new Dht(std::move(context.sock), SecureDht::getConfig(config.dht_config), context.logger));
+    dht_ = std::unique_ptr<SecureDht>(new SecureDht(std::move(dht), config.dht_config, std::move(context.identityAnnouncedCb), context.logger));
+
+    enableProxy(not config.proxy_server.empty());
     if (context.logger and dht_via_proxy_) {
         dht_via_proxy_->setLogger(context.logger);
     }

src/search.h

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

src/storage.h

@@ -37,7 +37,8 @@ public:
         storedValues_.emplace(expiration, std::pair<InfoHash, Value::Id>(id, value.id));
     }
     void erase(const InfoHash& id, const Value& value, time_point expiration) {
-        totalSize_ -= value.size();
+        auto size = value.size();
+        totalSize_ -= 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) {
@@ -58,7 +59,6 @@ struct ValueStorage {
     Sp<Value> data {};
     time_point created {};
     time_point expiration {};
-    Sp<Scheduler::Job> expiration_job {};
     StorageBucket* store_bucket {nullptr};
 
     ValueStorage() {}
@@ -143,15 +143,14 @@ struct Storage {
      * @param vid  The value id
      * @return time of the next expiration, time_point::max() if no expiration
      */
-    std::pair<ValueStorage*, time_point>
-    refresh(const time_point& now, const Value::Id& vid, const TypeStore& types) {
+    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, vs.expiration};
+                return vs.expiration;
             }
-        return {nullptr, time_point::max()};
+        return time_point::max();
     }
 
     size_t listen(ValueCallback& cb, Value::Filter& f, const Sp<Query>& q);
@@ -199,9 +198,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)
@@ -215,6 +214,7 @@ 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,8 +240,6 @@ 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};
@@ -290,8 +288,6 @@ Storage::expire(const InfoHash& id, time_point now)
         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,6 +28,12 @@ namespace dht {
 
 constexpr const size_t IO_THREADS_MAX {64};
 
+struct ThreadPool::ThreadState
+{
+    std::thread thread {};
+    std::atomic_bool run {true};
+};
+
 ThreadPool&
 ThreadPool::computation()
 {
@@ -61,12 +67,14 @@ void
 ThreadPool::run(std::function<void()>&& cb)
 {
     std::unique_lock<std::mutex> l(lock_);
-    if (not cb or not running_) return;
+    if (not running_) return;
 
     // launch new thread if necessary
     if (not readyThreads_ && threads_.size() < maxThreads_) {
-        threads_.emplace_back(std::make_unique<std::thread>([this]() {
-            while (true) {
+        threads_.emplace_back(new ThreadState());
+        auto& t = *threads_.back();
+        t.thread = std::thread([&]() {
+            while (t.run) {
                 std::function<void()> task;
 
                 // pick task from queue
@@ -74,10 +82,10 @@ ThreadPool::run(std::function<void()>&& cb)
                     std::unique_lock<std::mutex> l(lock_);
                     readyThreads_++;
                     cv_.wait(l, [&](){
-                        return not running_ or not tasks_.empty();
+                        return not t.run or not tasks_.empty();
                     });
                     readyThreads_--;
-                    if (not running_)
+                    if (not t.run)
                         break;
                     task = std::move(tasks_.front());
                     tasks_.pop();
@@ -85,13 +93,14 @@ ThreadPool::run(std::function<void()>&& cb)
 
                 // run task
                 try {
-                    task();
+                    if (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
@@ -104,8 +113,12 @@ ThreadPool::run(std::function<void()>&& cb)
 void
 ThreadPool::stop()
 {
-    std::lock_guard<std::mutex> l(lock_);
-    running_ = false;
+    {
+        std::lock_guard<std::mutex> l(lock_);
+        running_ = false;
+    }
+    for (auto& t : threads_)
+        t->run = false;
     cv_.notify_all();
 }
 
@@ -114,7 +127,7 @@ ThreadPool::join()
 {
     stop();
     for (auto& t : threads_)
-        t->join();
+        t->thread.join();
     threads_.clear();
 }
 
@@ -134,7 +147,7 @@ Executor::run_(std::function<void()>&& task)
 {
     current_++;
     std::weak_ptr<Executor> w = shared_from_this();
-    threadPool_.get().run([w,task = std::move(task)] {
+    threadPool_.get().run([w,task] {
         try {
             task();
         } catch (const std::exception& e) {

src/value_cache.h

@@ -58,7 +58,8 @@ public:
         values.clear();
         CallbackQueue ret;
         if (not expired_values.empty() and callback) {
-            ret.emplace_back([expired_values = std::move(expired_values), cb = callback]{
+            auto cb = callback;
+            ret.emplace_back([expired_values, cb]{
                 cb(expired_values, true);
             });
         }
@@ -102,23 +103,24 @@ public:
         }
         CallbackQueue ret;
         if (not expired_values.empty() and callback) {
-            ret.emplace_back([cb = callback, expired_values = std::move(expired_values)]{
-                cb(expired_values, true);
+            auto cb = callback;
+            ret.emplace_back([cb, expired_values]{
+                if (cb) cb(expired_values, true);
             });
         }
         return ret;
     }
 
     time_point onValues
-        (const std::vector<Sp<Value>>& new_values,
+        (const std::vector<Sp<Value>>& values,
         const std::vector<Value::Id>& refreshed_values,
         const std::vector<Value::Id>& expired_values,
         const TypeStore& types, const time_point& now)
     {
         CallbackQueue cbs;
         time_point ret = time_point::max();
-        if (not new_values.empty())
-            cbs.splice(cbs.end(), addValues(new_values, types, now));
+        if (not values.empty())
+            cbs.splice(cbs.end(), addValues(values, types, now));
         for (const auto& vid : refreshed_values)
             refreshValue(vid, types, now);
         for (const auto& vid : expired_values)
@@ -182,10 +184,11 @@ private:
                 v->second.expiration = now + types.getType(v->second.data->type).expiration;
             }
         }
+        auto cb = callback;
         CallbackQueue ret;
-        if (callback and not nvals.empty())
-            ret.emplace_back([cb = callback, nvals = std::move(nvals)]{
-                cb(nvals, false);
+        if (not nvals.empty())
+            ret.emplace_back([cb, nvals]{
+                if (cb) cb(nvals, false);
             });
         return ret;
     }

tests/threadpooltester.cpp

@@ -48,7 +48,7 @@ ThreadPoolTester::testThreadPool() {
         std::this_thread::sleep_for(std::chrono::milliseconds(10));
 
     pool.join();
-    CPPUNIT_ASSERT_EQUAL(N, count.load());
+    CPPUNIT_ASSERT(count.load() == N);
 }
 
 void

tools/CMakeLists.txt

@@ -30,7 +30,6 @@ 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 <opendht_c.h>
+#include <c/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_from_string(data_str);
+    dht_value* val = dht_value_new(data_str, strlen(data_str));
     dht_runner_put(runner, &h, val, dht_done_callback, runner, false);
     dht_value_unref(val);
 

tools/dhtnode.cpp

@@ -156,7 +156,7 @@ void cmd_loop(std::shared_ptr<DhtRunner>& node, dht_params& params
                     if (auto stats = proxy.second->stats())
                         std::cout << "  " << stats->toString() << std::endl;
                     else
-                        std::cout << "  (stats not available yet)" << std::endl;
+                        std::cout << "  (stats not available yet)" << std::endl;                
                 }
 #endif
             });