sockaddr: cleanup

CMakeLists.txt

@@ -71,11 +71,8 @@ if (NOT MSVC)
         )
     endif()
 
-    if (OPENDHT_HTTP OR OPENDHT_PEER_DISCOVERY)
-        find_path(ASIO_INCLUDE_DIR asio.hpp REQUIRED)
-    endif ()
-
     if (OPENDHT_HTTP)
+        find_path(ASIO_INCLUDE_DIR asio.hpp REQUIRED)
         find_package(Restinio REQUIRED)
         find_library(FMT_LIBRARY fmt)
         add_library(fmt SHARED IMPORTED)

docker/DockerfileDeps

@@ -17,9 +17,9 @@ RUN apt-get update && apt-get install -y \
 
 RUN echo "*** Downloading RESTinio ***" \
     && mkdir restinio && cd restinio \
-    && wget https://github.com/aberaud/restinio/archive/2224ffedef52cb2b74645d63d871d61dbd0f165e.tar.gz \
-    && ls -l && tar -xzf 2224ffedef52cb2b74645d63d871d61dbd0f165e.tar.gz \
-    && cd restinio-2224ffedef52cb2b74645d63d871d61dbd0f165e/dev \
+    && wget https://github.com/aberaud/restinio/archive/2c0b6f5e5ba04d7a74e8406a3df1fd433680599d.tar.gz \
+    && ls -l && tar -xzf 2c0b6f5e5ba04d7a74e8406a3df1fd433680599d.tar.gz \
+    && cd restinio-2c0b6f5e5ba04d7a74e8406a3df1fd433680599d/dev \
     && cmake -DCMAKE_INSTALL_PREFIX=/usr -DRESTINIO_TEST=OFF -DRESTINIO_SAMPLE=OFF \
              -DRESTINIO_INSTALL_SAMPLES=OFF -DRESTINIO_BENCH=OFF -DRESTINIO_INSTALL_BENCHES=OFF \
              -DRESTINIO_FIND_DEPS=ON -DRESTINIO_ALLOW_SOBJECTIZER=Off -DRESTINIO_USE_BOOST_ASIO=none . \

docker/DockerfileDepsLlvm

@@ -18,9 +18,9 @@ ENV CXX c++
 
 RUN echo "*** Downloading RESTinio ***" \
     && mkdir restinio && cd restinio \
-    && wget https://github.com/aberaud/restinio/archive/2224ffedef52cb2b74645d63d871d61dbd0f165e.tar.gz \
-    && ls -l && tar -xzf 2224ffedef52cb2b74645d63d871d61dbd0f165e.tar.gz \
-    && cd restinio-2224ffedef52cb2b74645d63d871d61dbd0f165e/dev \
+    && wget https://github.com/aberaud/restinio/archive/2c0b6f5e5ba04d7a74e8406a3df1fd433680599d.tar.gz \
+    && ls -l && tar -xzf 2c0b6f5e5ba04d7a74e8406a3df1fd433680599d.tar.gz \
+    && cd restinio-2c0b6f5e5ba04d7a74e8406a3df1fd433680599d/dev \
     && cmake -DCMAKE_INSTALL_PREFIX=/usr -DRESTINIO_TEST=OFF -DRESTINIO_SAMPLE=OFF \
              -DRESTINIO_INSTALL_SAMPLES=OFF -DRESTINIO_BENCH=OFF -DRESTINIO_INSTALL_BENCHES=OFF \
              -DRESTINIO_FIND_DEPS=ON -DRESTINIO_ALLOW_SOBJECTIZER=Off -DRESTINIO_USE_BOOST_ASIO=none . \

include/opendht/callbacks.h

@@ -164,7 +164,6 @@ using GetCallback = std::function<bool(const std::vector<std::shared_ptr<Value>>
 using ValueCallback = std::function<bool(const std::vector<std::shared_ptr<Value>>& values, bool expired)>;
 using GetCallbackSimple = std::function<bool(std::shared_ptr<Value> value)>;
 using ShutdownCallback = std::function<void()>;
-using IdentityAnnouncedCb = std::function<void(bool)>;
 
 using CertificateStoreQuery = std::function<std::vector<std::shared_ptr<crypto::Certificate>>(const InfoHash& pk_id)>;
 

include/opendht/crypto.h

@@ -80,8 +80,6 @@ struct OPENDHT_PUBLIC PublicKey
      * Takes ownership of an existing gnutls_pubkey.
      */
     PublicKey(gnutls_pubkey_t k) : pk(k) {}
-
-    /** Import public key from serialized data */
     PublicKey(const uint8_t* dat, size_t dat_size);
     PublicKey(const Blob& pk) : PublicKey(pk.data(), pk.size()) {}
     PublicKey(PublicKey&& o) noexcept : pk(o.pk) { o.pk = nullptr; };

include/opendht/dht.h

@@ -95,7 +95,7 @@ public:
     /**
      * Performs final operations before quitting.
      */
-    void shutdown(ShutdownCallback cb, bool stop = false) override;
+    void shutdown(ShutdownCallback cb) override;
 
     /**
      * Returns true if the node is running (have access to an open socket).
@@ -512,8 +512,6 @@ private:
 
     void confirmNodes();
     void expire();
-
-    void onConnected();
     void onDisconnected();
 
     /**

include/opendht/dht_interface.h

@@ -21,8 +21,6 @@
 #include "infohash.h"
 #include "log_enable.h"
 
-#include <queue>
-
 namespace dht {
 
 namespace net {
@@ -48,10 +46,6 @@ public:
     virtual NodeStatus getStatus(sa_family_t af) const = 0;
     virtual NodeStatus getStatus() const = 0;
 
-    void addOnConnectedCallback(std::function<void()> cb) {
-        onConnectCallbacks_.emplace(std::move(cb));
-    }
-
     virtual net::DatagramSocket* getSocket() const { return {}; };
 
     /**
@@ -61,10 +55,8 @@ public:
 
     /**
      * Performs final operations before quitting.
-     * stop: if true, cancel ongoing operations and call their 'done'
-     *       callbacks synchronously.
      */
-    virtual void shutdown(ShutdownCallback cb, bool stop = false) = 0;
+    virtual void shutdown(ShutdownCallback cb) = 0;
 
     /**
      * Returns true if the node is running (have access to an open socket).
@@ -272,7 +264,6 @@ public:
 
 protected:
     std::shared_ptr<Logger> logger_ {};
-    std::queue<std::function<void()>> onConnectCallbacks_ {};
 };
 
 } // namespace dht

include/opendht/dht_proxy_client.h

@@ -85,7 +85,7 @@ public:
     /**
      * Performs final operations before quitting.
      */
-    void shutdown(ShutdownCallback cb, bool) override;
+    void shutdown(ShutdownCallback cb) override;
 
     /**
      * Returns true if the node is running (have access to an open socket).
@@ -378,8 +378,8 @@ private:
      * Retrieve if we can connect to the proxy (update statusIpvX_)
      */
     void handleProxyConfirm(const asio::error_code &ec);
-    std::unique_ptr<asio::steady_timer> nextProxyConfirmationTimer_;
-    std::unique_ptr<asio::steady_timer> listenerRestartTimer_;
+    Sp<asio::steady_timer> nextProxyConfirmationTimer_;
+    Sp<asio::steady_timer> listenerRestartTimer_;
 
     /**
      * Relaunch LISTEN requests if the client disconnect/reconnect.

include/opendht/dht_proxy_server.h

@@ -445,6 +445,7 @@ private:
         std::map<InfoHash, std::vector<Listener>> listeners;
         MSGPACK_DEFINE_ARRAY(listeners)
     };
+    std::mutex lockPushListeners_;
     std::map<std::string, PushListener> pushListeners_;
     proxy::ListenToken tokenPushNotif_ {0};
 #endif //OPENDHT_PUSH_NOTIFICATIONS

include/opendht/dhtrunner.h

@@ -74,7 +74,6 @@ public:
         std::shared_ptr<PeerDiscovery> peerDiscovery {};
         StatusCallback statusChangedCallback {};
         CertificateStoreQuery certificateStore {};
-        IdentityAnnouncedCb identityAnnouncedCb {};
         Context() {}
     };
 
@@ -408,7 +407,7 @@ public:
     /**
      * Gracefuly disconnect from network.
      */
-    void shutdown(ShutdownCallback cb = {}, bool stop = false);
+    void shutdown(ShutdownCallback cb = {});
 
     /**
      * Quit and wait for all threads to terminate.
@@ -474,7 +473,6 @@ private:
 
     /** Current configuration */
     Config config_;
-    IdentityAnnouncedCb identityAnnouncedCb_;
 
     /**
      * reset dht clients

include/opendht/securedht.h

@@ -45,13 +45,15 @@ public:
         return c;
     }
 
+    SecureDht() {}
+
     /**
      * s, s6: bound socket descriptors for IPv4 and IPv6, respectively.
      *        For the Dht to be initialised, at least one of them must be >= 0.
      * id:    the identity to use for the crypto layer and to compute
      *        our own hash on the Dht.
      */
-    SecureDht(std::unique_ptr<DhtInterface> dht, Config config, IdentityAnnouncedCb iacb = {}, const std::shared_ptr<Logger>& l = {});
+    SecureDht(std::unique_ptr<DhtInterface> dht, Config config);
 
     virtual ~SecureDht();
 
@@ -146,8 +148,8 @@ public:
     /**
      * SecureDht to Dht proxy
      */
-    void shutdown(ShutdownCallback cb, bool stop = false) override {
-        dht_->shutdown(cb, stop);
+    void shutdown(ShutdownCallback cb) override {
+        dht_->shutdown(cb);
     }
     void dumpTables() const override {
         dht_->dumpTables();

include/opendht/sockaddr.h

@@ -68,7 +68,7 @@ public:
      * Build from existing address.
      */
     SockAddr(const sockaddr* sa, socklen_t length) {
-        if (length > static_cast<socklen_t>(sizeof(sockaddr_storage)))
+        if (length > sizeof(sockaddr_storage))
             throw std::runtime_error("Socket address length is too large");
         set(sa, length);
     }

include/opendht/thread_pool.h

@@ -93,71 +93,4 @@ private:
     void schedule();
 };
 
-class OPENDHT_PUBLIC ExecutionContext {
-public:
-    ExecutionContext(ThreadPool& pool)
-     : threadPool_(pool), state_(std::make_shared<SharedState>())
-    {}
-
-    ~ExecutionContext() {
-        state_->destroy();
-    }
-
-    /** Wait for ongoing tasks to complete execution and drop other pending tasks */
-    void stop() {
-        state_->destroy(false);
-    }
-
-    void run(std::function<void()>&& task) {
-        std::lock_guard<std::mutex> lock(state_->mtx);
-        if (state_->shutdown_) return;
-        state_->pendingTasks++;
-        threadPool_.get().run([task = std::move(task), state = state_] {
-            state->run(task);
-        });
-    }
-
-private:
-    struct SharedState {
-        std::mutex mtx {};
-        std::condition_variable cv {};
-        unsigned pendingTasks {0};
-        unsigned ongoingTasks {0};
-        /** When true, prevents new tasks to be scheduled */
-        bool shutdown_ {false};
-        /** When true, prevents scheduled tasks to be executed */
-        std::atomic_bool destroyed {false};
-
-        void destroy(bool wait = true) {
-            std::unique_lock<std::mutex> lock(mtx);
-            if (destroyed) return;
-            if (wait) {
-                cv.wait(lock, [this] { return pendingTasks == 0 && ongoingTasks == 0; });
-            }
-            shutdown_ = true;
-            if (not wait) {
-                cv.wait(lock, [this] { return ongoingTasks == 0; });
-            }
-            destroyed = true;
-        }
-
-        void run(const std::function<void()>& task) {
-            {
-                std::lock_guard<std::mutex> lock(mtx);
-                pendingTasks--;
-                ongoingTasks++;
-            }
-                if (destroyed) return;
-                task();
-            {
-                std::lock_guard<std::mutex> lock(mtx);
-                ongoingTasks--;
-                cv.notify_all();
-            }
-        }
-    };
-    std::reference_wrapper<ThreadPool> threadPool_;
-    std::shared_ptr<SharedState> state_;
-};
-
 }

python/opendht.pyx

@@ -36,7 +36,6 @@ from libcpp.memory cimport shared_ptr
 from cython.parallel import parallel, prange
 from cython.operator cimport dereference as deref, preincrement as inc, predecrement as dec
 from cpython cimport ref
-from datetime import timedelta
 
 cimport opendht_cpp as cpp
 
@@ -236,9 +235,8 @@ cdef class Where(object):
 
 cdef class Value(object):
     cdef shared_ptr[cpp.Value] _value
-    def __init__(self, bytes val=b'', cpp.uint16_t id=0):
-        self._value.reset(new cpp.Value(id, val, len(val)))
-
+    def __init__(self, bytes val=b''):
+        self._value.reset(new cpp.Value(val, len(val)))
     def __str__(self):
         return self._value.get().toString().decode()
     property owner:
@@ -272,14 +270,6 @@ cdef class Value(object):
         def __get__(self):
             return self._value.get().size()
 
-cdef class ValueType(object):
-    cdef cpp.ValueType * _value
-    def __init__(self, cpp.uint16_t id, str name, expiration: timedelta):
-        if not isinstance(expiration, timedelta):
-            raise TypeError("expiration argument must be of type timedelta")
-        cdef cpp.seconds duration = cpp.seconds(int(expiration.total_seconds()))
-        self._value =  new cpp.ValueType(id, name.encode(), duration)
-
 cdef class NodeSetIter(object):
     cdef map[cpp.InfoHash, shared_ptr[cpp.Node]]* _nodes
     cdef map[cpp.InfoHash, shared_ptr[cpp.Node]].iterator _curIter
@@ -523,8 +513,6 @@ cdef class DhtRunner(_WithID):
         cb_obj = {'shutdown':shutdown_cb}
         ref.Py_INCREF(cb_obj)
         self.thisptr.get().shutdown(cpp.bindShutdownCb(shutdown_callback, <void*>cb_obj))
-    def registerType(self, ValueType type):
-        self.thisptr.get().registerType(deref(type._value))
     def enableLogging(self):
         cpp.enableLogging(self.thisptr.get()[0])
     def disableLogging(self):

python/opendht_cpp.pxd

@@ -26,14 +26,6 @@ from libc.string cimport const_char, const_uchar
 ctypedef uint16_t in_port_t
 ctypedef unsigned short int sa_family_t;
 
-cdef extern from "<iostream>" namespace "std::chrono" nogil:
-    cdef cppclass duration[ulong]:
-        duration() except +
-
-    cdef cppclass seconds:
-        duration seconds(uint64_t) except +
-        duration seconds()
-
 cdef extern from "<memory>" namespace "std" nogil:
     cdef cppclass shared_ptr[T]:
         shared_ptr() except +
@@ -139,6 +131,7 @@ cdef extern from "opendht/value.h" namespace "dht::Value":
 cdef extern from "opendht/value.h" namespace "dht::Value::Field":
     cdef Field None
     cdef Field Id
+    cdef Field ValueType
     cdef Field OwnerPk
     cdef Field SeqNum
     cdef Field UserType
@@ -148,7 +141,7 @@ cdef extern from "opendht/value.h" namespace "dht":
     cdef cppclass Value:
         Value() except +
         Value(vector[uint8_t]) except +
-        Value(const uint16_t t, const uint8_t* dat_ptr, size_t dat_len) except +
+        Value(const uint8_t* dat_ptr, size_t dat_len) except +
         string toString() const
         size_t size() const
         uint64_t id
@@ -157,12 +150,6 @@ cdef extern from "opendht/value.h" namespace "dht":
         vector[uint8_t] data
         string user_type
 
-    cdef cppclass ValueType:
-        ValueType(uint16_t id, string name, seconds expiration) except +
-        uint16_t id
-        string name
-        seconds expiration
-
     cdef cppclass Query:
         Query() except +
         Query(Select s, Where w) except +
@@ -247,7 +234,6 @@ cdef extern from "opendht/dhtrunner.h" namespace "dht":
         void run(const_char*, const_char*, const_char*, Config config)
         void join()
         void shutdown(ShutdownCallback)
-        void registerType(ValueType& value);
         bool isRunning()
         SockAddr getBound(sa_family_t af) const
         string getStorageLog() const

src/crypto.cpp

@@ -710,13 +710,13 @@ CertificateRequest::sign(const PrivateKey& key, const std::string& password)
 }
 
 bool
-CertificateRequest::verify() const
+CertificateRequest::verify() const 
 {
     return gnutls_x509_crq_verify(request, 0) >= 0;
 }
 
-Blob
-CertificateRequest::pack() const
+Blob 
+CertificateRequest::pack() const 
 {
     gnutls_datum_t dat {nullptr, 0};
     if (auto err = gnutls_x509_crq_export2(request, GNUTLS_X509_FMT_PEM, &dat))
@@ -727,7 +727,7 @@ CertificateRequest::pack() const
 }
 
 std::string
-CertificateRequest::toString() const
+CertificateRequest::toString() const 
 {
     gnutls_datum_t dat {nullptr, 0};
     if (auto err = gnutls_x509_crq_export2(request, GNUTLS_X509_FMT_PEM, &dat))
@@ -1147,8 +1147,8 @@ void
 setRandomSerial(gnutls_x509_crt_t cert)
 {
     random_device rdev;
-    std::uniform_int_distribution<int64_t> dist{1};
-    int64_t cert_serial = dist(rdev);
+    std::uniform_int_distribution<uint64_t> dist{};
+    uint64_t cert_serial = dist(rdev);
     gnutls_x509_crt_set_serial(cert, &cert_serial, sizeof(cert_serial));
 }
 

src/dht.cpp

@@ -52,7 +52,12 @@ Dht::updateStatus(sa_family_t af)
         if (other.status == NodeStatus::Disconnected && d.status == NodeStatus::Disconnected)
             onDisconnected();
         else if (other.status == NodeStatus::Connected || d.status == NodeStatus::Connected) {
-            onConnected();
+            // On connected
+            if (bootstrapJob) {
+                bootstrapJob->cancel();
+                bootstrapJob.reset();
+            }
+            bootstrap_period = std::chrono::seconds(10);
         }
     }
     return d.status;
@@ -77,27 +82,11 @@ Dht::Kad::getStatus(time_point now) const
 }
 
 void
-Dht::shutdown(ShutdownCallback cb, bool stop)
+Dht::shutdown(ShutdownCallback cb)
 {
     if (not persistPath.empty())
         saveState(persistPath);
 
-    if (stop) {
-        for (auto dht : {&dht4, &dht6}) {
-            for (auto& sr : dht->searches) {
-                for (const auto& r : sr.second->callbacks)
-                    r.second.done_cb(false, {});
-                sr.second->callbacks.clear();
-                for (const auto& a : sr.second->announce) {
-                    if (a.callback) a.callback(false, {});
-                }
-                sr.second->announce.clear();
-                sr.second->listeners.clear();
-            }
-        }
-        network_engine.clear();
-    }
-
     if (not maintain_storage) {
         if (cb) cb();
         return;
@@ -502,9 +491,9 @@ void Dht::searchSendAnnounceValue(const Sp<Search>& sr) {
                         sr->id.toString().c_str(), sn->node->toString().c_str(), a.value->id);
                 sn->acked[a.value->id] = {
                     network_engine.sendRefreshValue(sn->node, sr->id, a.value->id, sn->token, onDone,
-                    [this, ws, node=sn->node, v=a.value,
-                     onDone,
-                     onExpired,
+                    [this, ws, node=sn->node, v=a.value, 
+                     onDone, 
+                     onExpired, 
                      created = a.permanent ? time_point::max() : a.created,
                      next_refresh_time
                     ](const net::Request& /*req*/, net::DhtProtocolException&& e){
@@ -602,7 +591,7 @@ Dht::searchSynchedNodeListen(const Sp<Search>& sr, SearchNode& n)
     std::weak_ptr<Search> ws = sr;
     for (const auto& l : sr->listeners) {
         const auto& query = l.second.query;
-
+        
         auto r = n.listenStatus.find(query);
         if (n.getListenTime(r, listenExp) > scheduler.time())
             continue;
@@ -736,7 +725,7 @@ Dht::searchStep(Sp<Search> sr)
 
     while (sr->currentlySolicitedNodeCount() < MAX_REQUESTED_SEARCH_NODES and searchSendGetValues(sr));
 
-
+    
     if (sr->getNumberOfConsecutiveBadNodes() >= std::min<size_t>(sr->nodes.size(), SEARCH_MAX_BAD_NODES))
     {
         if (logger_)
@@ -1992,21 +1981,6 @@ Dht::expire()
     scheduler.add(expire_stuff_time, std::bind(&Dht::expire, this));
 }
 
-void
-Dht::onConnected()
-{
-    if (bootstrapJob) {
-        bootstrapJob->cancel();
-        bootstrapJob.reset();
-    }
-    bootstrap_period = std::chrono::seconds(10);
-    auto callbacks = std::move(onConnectCallbacks_);
-    while (not callbacks.empty()) {
-        callbacks.front()();
-        callbacks.pop();
-    }
-}
-
 void
 Dht::onDisconnected()
 {
@@ -2552,7 +2526,7 @@ Dht::saveState(const std::string& path) const
     state.nodes = exportNodes();
     state.values = exportValues();
     std::ofstream file(path);
-    msgpack::pack(file, state);
+    msgpack::pack(file, state);    
 }
 
 void

src/dht_proxy_client.cpp

@@ -155,10 +155,10 @@ DhtProxyClient::startProxy()
     if (logger_)
         logger_->d("[proxy:client] start proxy with %s", proxyUrl_.c_str());
 
-    nextProxyConfirmationTimer_ = std::make_unique<asio::steady_timer>(httpContext_, std::chrono::steady_clock::now());
+    nextProxyConfirmationTimer_ = std::make_shared<asio::steady_timer>(httpContext_, std::chrono::steady_clock::now());
     nextProxyConfirmationTimer_->async_wait(std::bind(&DhtProxyClient::handleProxyConfirm, this, std::placeholders::_1));
 
-    listenerRestartTimer_ = std::make_unique<asio::steady_timer>(httpContext_);
+    listenerRestartTimer_ = std::make_shared<asio::steady_timer>(httpContext_);
 
     loopSignal_();
 }
@@ -242,7 +242,7 @@ DhtProxyClient::cancelAllListeners()
 }
 
 void
-DhtProxyClient::shutdown(ShutdownCallback cb, bool)
+DhtProxyClient::shutdown(ShutdownCallback cb)
 {
     stop();
     if (cb)
@@ -723,15 +723,6 @@ DhtProxyClient::onProxyInfos(const Json::Value& proxyInfos, const sa_family_t fa
         if (oldStatus == NodeStatus::Disconnected || oldStatus == NodeStatus::Connecting) {
             listenerRestartTimer_->expires_at(std::chrono::steady_clock::now());
             listenerRestartTimer_->async_wait(std::bind(&DhtProxyClient::restartListeners, this, std::placeholders::_1));
-            if (not onConnectCallbacks_.empty()) {
-                std::lock_guard<std::mutex> lock(lockCallbacks_);
-                callbacks_.emplace_back([cbs = std::move(onConnectCallbacks_)]() mutable {
-                    while (not cbs.empty()) {
-                        cbs.front()();
-                        cbs.pop();
-                    }
-                });
-            }
         }
         nextProxyConfirmationTimer_->expires_at(std::chrono::steady_clock::now() + std::chrono::minutes(15));
         nextProxyConfirmationTimer_->async_wait(std::bind(&DhtProxyClient::handleProxyConfirm, this, std::placeholders::_1));

src/dht_proxy_server.cpp

@@ -822,7 +822,7 @@ DhtProxyServer::subscribe(restinio::request_handle_t request,
             logger_->d("[proxy:server] [subscribe %s] [client %s] [session %s]", infoHash.toString().c_str(), clientId.c_str(), sessionId.c_str());
 
         // Insert new or return existing push listeners of a token
-        std::lock_guard<std::mutex> lock(lockListener_);
+        std::lock_guard<std::mutex> lock(lockPushListeners_);
         auto& pushListener = pushListeners_[pushToken];
         auto& pushListeners = pushListener.listeners[infoHash];
 

src/dhtrunner.cpp

@@ -104,11 +104,8 @@ DhtRunner::run(const Config& config, Context&& context)
 {
     std::lock_guard<std::mutex> lck(dht_mtx);
     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.");
+    if (not running.compare_exchange_strong(expected, State::Running))
         return;
-    }
 
     auto local4 = config.bind4;
     auto local6 = config.bind6;
@@ -119,24 +116,22 @@ DhtRunner::run(const Config& config, Context&& context)
         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)) {
+    if (not state_path.empty()) {
         state_path += "_port.txt";
         std::ifstream inConfig(state_path);
         if (inConfig.is_open()) {
             in_port_t port;
             if (inConfig >> port) {
-                if (local4.getPort() == 0) {
-                    if (context.logger)
-                        context.logger->d("[runner %p] Using IPv4 port %hu from saved configuration", this, port);
+                if (context.logger)
+                    context.logger->d("[runner %p] Using IPv4 port %hu from saved configuration", this, port);
+                if (local4.getPort() == 0)
                     local4.setPort(port);
-                }
             }
             if (inConfig >> port) {
-                if (local6.getPort() == 0) {
-                    if (context.logger)
-                        context.logger->d("[runner %p] Using IPv6 port %hu from saved configuration", this, port);
+                if (context.logger)
+                    context.logger->d("[runner %p] Using IPv6 port %hu from saved configuration", this, port);
+                if (local6.getPort() == 0)
                     local6.setPort(port);
-                }
             }
         }
     }
@@ -173,13 +168,12 @@ DhtRunner::run(const Config& config, Context&& context)
         return ret;
     });
 
+    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));
+
 #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);
@@ -268,31 +262,29 @@ DhtRunner::run(const Config& config, Context&& context)
 }
 
 void
-DhtRunner::shutdown(ShutdownCallback cb, bool stop) {
-    std::unique_lock<std::mutex> lck(storage_mtx);
+DhtRunner::shutdown(ShutdownCallback cb) {
     auto expected = State::Running;
     if (not running.compare_exchange_strong(expected, State::Stopping)) {
         if (expected == State::Stopping and ongoing_ops) {
+            std::lock_guard<std::mutex> lck(storage_mtx);
             shutdownCallbacks_.emplace_back(std::move(cb));
         }
-        else if (cb) {
-            lck.unlock();
-            cb();
-        }
+        else if (cb) cb();
         return;
     }
     if (logger_)
         logger_->d("[runner %p] state changed to Stopping, %zu ongoing ops", this, ongoing_ops.load());
+    std::lock_guard<std::mutex> lck(storage_mtx);
     ongoing_ops++;
     shutdownCallbacks_.emplace_back(std::move(cb));
-    pending_ops.emplace([=](SecureDht&) mutable {
+    pending_ops_prio.emplace([=](SecureDht&) mutable {
         auto onShutdown = [this]{ opEnded(); };
 #ifdef OPENDHT_PROXY_CLIENT
         if (dht_via_proxy_)
-            dht_via_proxy_->shutdown(onShutdown, stop);
+            dht_via_proxy_->shutdown(onShutdown);
 #endif
         if (dht_)
-            dht_->shutdown(onShutdown, stop);
+            dht_->shutdown(onShutdown);
     });
     cv.notify_all();
 }
@@ -321,11 +313,11 @@ DhtRunner::bindOpDoneCallback(DoneCallbackSimple&& cb) {
 
 bool
 DhtRunner::checkShutdown() {
+    if (running != State::Stopping or ongoing_ops)
+        return false;
     decltype(shutdownCallbacks_) cbs;
     {
         std::lock_guard<std::mutex> lck(storage_mtx);
-        if (running != State::Stopping or ongoing_ops)
-            return false;
         cbs = std::move(shutdownCallbacks_);
     }
     for (auto& cb : cbs)
@@ -357,13 +349,9 @@ DhtRunner::join()
 
     {
         std::lock_guard<std::mutex> lck(storage_mtx);
-        if (ongoing_ops and logger_) {
-            logger_->w("[runner %p] stopping with %zu remaining ops", this, ongoing_ops.load());
-        }
         pending_ops = decltype(pending_ops)();
         pending_ops_prio = decltype(pending_ops_prio)();
         ongoing_ops = 0;
-        shutdownCallbacks_.clear();
     }
     {
         std::lock_guard<std::mutex> lck(dht_mtx);
@@ -715,12 +703,11 @@ DhtRunner::loop_()
 void
 DhtRunner::get(InfoHash hash, GetCallback vcb, DoneCallback dcb, Value::Filter f, Where w)
 {
-    std::unique_lock<std::mutex> lck(storage_mtx);
     if (running != State::Running) {
-        lck.unlock();
         if (dcb) dcb(false, {});
         return;
     }
+    std::lock_guard<std::mutex> lck(storage_mtx);
     ongoing_ops++;
     pending_ops.emplace([=](SecureDht& dht) mutable {
         dht.get(hash, std::move(vcb), bindOpDoneCallback(std::move(dcb)), std::move(f), std::move(w));
@@ -735,12 +722,11 @@ DhtRunner::get(const std::string& key, GetCallback vcb, DoneCallbackSimple dcb,
 }
 void
 DhtRunner::query(const InfoHash& hash, QueryCallback cb, DoneCallback done_cb, Query q) {
-    std::unique_lock<std::mutex> lck(storage_mtx);
     if (running != State::Running) {
-        lck.unlock();
         if (done_cb) done_cb(false, {});
         return;
     }
+    std::lock_guard<std::mutex> lck(storage_mtx);
     ongoing_ops++;
     pending_ops.emplace([=](SecureDht& dht) mutable {
         dht.query(hash, std::move(cb), bindOpDoneCallback(std::move(done_cb)), std::move(q));
@@ -752,12 +738,11 @@ std::future<size_t>
 DhtRunner::listen(InfoHash hash, ValueCallback vcb, Value::Filter f, Where w)
 {
     auto ret_token = std::make_shared<std::promise<size_t>>();
-    std::unique_lock<std::mutex> lck(storage_mtx);
     if (running != State::Running) {
-        lck.unlock();
         ret_token->set_value(0);
         return ret_token->get_future();
     }
+    std::lock_guard<std::mutex> lck(storage_mtx);
     pending_ops.emplace([=](SecureDht& dht) mutable {
 #ifdef OPENDHT_PROXY_CLIENT
         auto tokenbGlobal = listener_token_++;
@@ -838,12 +823,11 @@ DhtRunner::cancelListen(InfoHash h, std::shared_future<size_t> ftoken)
 void
 DhtRunner::put(InfoHash hash, Value&& value, DoneCallback cb, time_point created, bool permanent)
 {
-    std::unique_lock<std::mutex> lck(storage_mtx);
     if (running != State::Running) {
-        lck.unlock();
         if (cb) cb(false, {});
         return;
     }
+    std::lock_guard<std::mutex> lck(storage_mtx);
     ongoing_ops++;
     pending_ops.emplace([=,
         cb = std::move(cb),
@@ -857,12 +841,11 @@ DhtRunner::put(InfoHash hash, Value&& value, DoneCallback cb, time_point created
 void
 DhtRunner::put(InfoHash hash, std::shared_ptr<Value> value, DoneCallback cb, time_point created, bool permanent)
 {
-    std::unique_lock<std::mutex> lck(storage_mtx);
     if (running != State::Running) {
-        lck.unlock();
         if (cb) cb(false, {});
         return;
     }
+    std::lock_guard<std::mutex> lck(storage_mtx);
     ongoing_ops++;
     pending_ops.emplace([=, cb = std::move(cb)](SecureDht& dht) mutable {
         dht.put(hash, value, bindOpDoneCallback(std::move(cb)), created, permanent);
@@ -899,12 +882,11 @@ DhtRunner::cancelPut(const InfoHash& h, const std::shared_ptr<Value>& value)
 void
 DhtRunner::putSigned(InfoHash hash, std::shared_ptr<Value> value, DoneCallback cb, bool permanent)
 {
-    std::unique_lock<std::mutex> lck(storage_mtx);
     if (running != State::Running) {
-        lck.unlock();
         if (cb) cb(false, {});
         return;
     }
+    std::lock_guard<std::mutex> lck(storage_mtx);
     ongoing_ops++;
     pending_ops.emplace([=,
         cb = std::move(cb),
@@ -930,12 +912,11 @@ DhtRunner::putSigned(const std::string& key, Value&& value, DoneCallbackSimple c
 void
 DhtRunner::putEncrypted(InfoHash hash, InfoHash to, std::shared_ptr<Value> value, DoneCallback cb, bool permanent)
 {
-    std::unique_lock<std::mutex> lck(storage_mtx);
     if (running != State::Running) {
-        lck.unlock();
         if (cb) cb(false, {});
         return;
     }
+    std::lock_guard<std::mutex> lck(storage_mtx);
     ongoing_ops++;
     pending_ops.emplace([=,
         cb = std::move(cb),
@@ -1021,12 +1002,11 @@ DhtRunner::bootstrap(std::vector<SockAddr> nodes, DoneCallbackSimple&& cb)
 void
 DhtRunner::bootstrap(const SockAddr& addr, DoneCallbackSimple&& cb)
 {
-    std::unique_lock<std::mutex> lck(storage_mtx);
     if (running != State::Running) {
-        lck.unlock();
         if (cb) cb(false);
         return;
     }
+    std::lock_guard<std::mutex> lck(storage_mtx);
     ongoing_ops++;
     pending_ops_prio.emplace([addr, cb = bindOpDoneCallback(std::move(cb))](SecureDht& dht) mutable {
         dht.pingNode(std::move(addr), std::move(cb));
@@ -1037,9 +1017,9 @@ DhtRunner::bootstrap(const SockAddr& addr, DoneCallbackSimple&& cb)
 void
 DhtRunner::bootstrap(const InfoHash& id, const SockAddr& address)
 {
-    std::lock_guard<std::mutex> lck(storage_mtx);
     if (running != State::Running)
         return;
+    std::unique_lock<std::mutex> lck(storage_mtx);
     pending_ops_prio.emplace([id, address](SecureDht& dht) mutable {
         dht.insertNode(id, address);
     });
@@ -1049,9 +1029,9 @@ DhtRunner::bootstrap(const InfoHash& id, const SockAddr& address)
 void
 DhtRunner::bootstrap(const std::vector<NodeExport>& nodes)
 {
-    std::lock_guard<std::mutex> lck(storage_mtx);
     if (running != State::Running)
         return;
+    std::lock_guard<std::mutex> lck(storage_mtx);
     pending_ops_prio.emplace([=](SecureDht& dht) {
         for (auto& node : nodes)
             dht.insertNode(node);
@@ -1075,12 +1055,11 @@ DhtRunner::connectivityChanged()
 
 void
 DhtRunner::findCertificate(InfoHash hash, std::function<void(const Sp<crypto::Certificate>&)> cb) {
-    std::unique_lock<std::mutex> lck(storage_mtx);
     if (running != State::Running) {
-        lck.unlock();
         cb({});
         return;
     }
+    std::lock_guard<std::mutex> lck(storage_mtx);
     ongoing_ops++;
     pending_ops.emplace([this, hash, cb = std::move(cb)] (SecureDht& dht) {
         dht.findCertificate(hash, [this, cb = std::move(cb)](const Sp<crypto::Certificate>& crt){
@@ -1156,7 +1135,7 @@ DhtRunner::enableProxy(bool proxify)
         if (not config_.push_token.empty())
             dht_via_proxy->setPushNotificationToken(config_.push_token);
 #endif
-        dht_via_proxy_ = std::unique_ptr<SecureDht>(new SecureDht(std::move(dht_via_proxy), config_.dht_config, identityAnnouncedCb_, logger_));
+        dht_via_proxy_ = std::unique_ptr<SecureDht>(new SecureDht(std::move(dht_via_proxy), config_.dht_config));
         // add current listeners
         for (auto& l: listeners_)
             l.second.tokenProxyDht = dht_via_proxy_->listen(l.second.hash, l.second.gcb, l.second.f, l.second.w);

src/http.cpp

@@ -561,63 +561,12 @@ void
 Connection::async_connect(std::vector<asio::ip::tcp::endpoint>&& endpoints, ConnectHandlerCb cb)
 {
     std::lock_guard<std::mutex> lock(mutex_);
-    if (!ssl_socket_ && !socket_) {
-        cb(asio::error::operation_aborted, {});
-        return;
-    }
-    auto& base = ssl_socket_? ssl_socket_->lowest_layer() : *socket_;
-
-    ConnectHandlerCb wcb = [&base, cb=std::move(cb)](const asio::error_code& ec, const asio::ip::tcp::endpoint& endpoint) {
-        if (!ec) {
-            auto socket = base.native_handle();
-            // Once connected, set a keep alive on the TCP socket with 30 seconds delay
-            // This will generate broken pipes as soon as possible.
-            // Note this needs to be done once connected to have a valid native_handle()
-            uint32_t start = 30;
-            uint32_t interval = 30;
-            uint32_t cnt = 1;
-#ifdef _WIN32
-            std::string val = "1";
-            setsockopt(socket, SOL_SOCKET, SO_KEEPALIVE, val.c_str(), sizeof(val));
-
-            // TCP_KEEPIDLE and TCP_KEEPINTVL are available since Win 10 version 1709
-            // TCP_KEEPCNT since Win 10 version 1703
-#ifdef TCP_KEEPIDLE
-            std::string start_str = std::to_string(start);
-            setsockopt(socket, IPPROTO_TCP, TCP_KEEPIDLE,
-                    start_str.c_str(), sizeof(start_str));
-#endif
-#ifdef TCP_KEEPINTVL
-            std::string interval_str = std::to_string(interval);
-            setsockopt(socket, IPPROTO_TCP, TCP_KEEPINTVL,
-                    interval_str.c_str(), sizeof(interval_str));
-#endif
-#ifdef TCP_KEEPCNT
-            std::string cnt_str = std::to_string(cnt);
-            setsockopt(socket, IPPROTO_TCP, TCP_KEEPCNT,
-                    cnt_str.c_str(), sizeof(cnt_str));
-#endif
-#else
-            uint32_t val = 1;
-            setsockopt(socket, SOL_SOCKET, SO_KEEPALIVE, &val, sizeof(uint32_t));
-#ifdef __APPLE__
-            // Apple devices only have one parameter
-            setsockopt(socket, IPPROTO_TCP, TCP_KEEPALIVE, &start, sizeof(uint32_t));
-#else
-            // Linux based systems
-            setsockopt(socket, SOL_TCP, TCP_KEEPIDLE, &start, sizeof(uint32_t));
-            setsockopt(socket, SOL_TCP, TCP_KEEPINTVL, &interval, sizeof(uint32_t));
-            setsockopt(socket, SOL_TCP, TCP_KEEPCNT, &cnt, sizeof(uint32_t));
-#endif
-#endif
-        }
-        if (cb)
-            cb(ec, endpoint);
-    };
     if (ssl_socket_)
-        asio::async_connect(ssl_socket_->lowest_layer(), std::move(endpoints), wrapCallabck(std::move(wcb)));
-    else
-        asio::async_connect(*socket_, std::move(endpoints), wrapCallabck(std::move(wcb)));
+        asio::async_connect(ssl_socket_->lowest_layer(), std::move(endpoints), wrapCallabck(std::move(cb)));
+    else if (socket_)
+        asio::async_connect(*socket_, std::move(endpoints), wrapCallabck(std::move(cb)));
+    else if (cb)
+        cb(asio::error::operation_aborted, {});
 }
 
 void
@@ -1305,11 +1254,9 @@ Request::terminate(const asio::error_code& ec)
         return;
 
     response_.aborted = ec == asio::error::operation_aborted;
-    if (ec == asio::error::basic_errors::broken_pipe)
-        response_.status_code = 0U; // Avoid to give a successful answer (happen with a broken pipe, takes the last status)
 
     if (logger_) {
-        if (ec and ec != asio::error::eof and !response_.aborted)
+        if (ec and ec != asio::error::eof and ec != asio::error::operation_aborted)
             logger_->e("[http:request:%i] end with error: %s", id_, ec.message().c_str());
         else
             logger_->d("[http:request:%i] done with status code %u", id_, response_.status_code);
@@ -1468,7 +1415,7 @@ Request::await()
     std::unique_lock<std::mutex> lock(mtx);
     std::condition_variable cv;
     bool ok {false};
-    add_on_done_callback([&](const Response&){
+    add_on_done_callback([&](const Response& resp){
         std::lock_guard<std::mutex> lk(mtx);
         ok = true;
         cv.notify_all();

src/securedht.cpp

@@ -36,8 +36,8 @@ extern "C" {
 
 namespace dht {
 
-SecureDht::SecureDht(std::unique_ptr<DhtInterface> dht, SecureDht::Config conf, IdentityAnnouncedCb iacb, const std::shared_ptr<Logger>& l)
-: DhtInterface(l), dht_(std::move(dht)), key_(conf.id.first), certificate_(conf.id.second), enableCache_(conf.cert_cache_all)
+SecureDht::SecureDht(std::unique_ptr<DhtInterface> dht, SecureDht::Config conf)
+: dht_(std::move(dht)), key_(conf.id.first), certificate_(conf.id.second), enableCache_(conf.cert_cache_all)
 {
     if (!dht_) return;
     for (const auto& type : DEFAULT_TYPES)
@@ -52,17 +52,16 @@ SecureDht::SecureDht(std::unique_ptr<DhtInterface> dht, SecureDht::Config conf,
         auto certId = certificate_->getId();
         if (key_ and certId != key_->getPublicKey().getId())
             throw DhtException("SecureDht: provided certificate doesn't match private key.");
-        dht_->addOnConnectedCallback([this, certId, cb=std::move(iacb)]{
-            dht_->put(certId, Value {
-                CERTIFICATE_TYPE,
-                *certificate_,
-                1
-            }, [this, certId, cb=std::move(cb)](bool ok) {
-                if (cb) cb(ok);
+
+        dht_->put(certId, Value {
+            CERTIFICATE_TYPE,
+            *certificate_,
+            1
+        }, [this, certId](bool ok) {
+            if (ok)
                 if (logger_)
-                    logger_->d(certId, "SecureDht: certificate announcement %s", ok ? "succeeded" : "failed");
-            }, {}, true);
-        });
+                    logger_->d(certId, "SecureDht: public key announced successfully");
+        }, {}, true);
     }
 }
 

src/value.cpp

@@ -496,10 +496,10 @@ Query::msgpack_unpack(const msgpack::object& o)
 }
 
 template <typename T>
-bool subset(const std::vector<T>& fds, const std::vector<T>& qfds)
+bool subset(std::vector<T> fds, std::vector<T> qfds)
 {
-    for (const auto& fd : fds) {
-        if (std::find_if(qfds.begin(), qfds.end(), [&fd](const T& _vfd) { return fd == _vfd; }) == qfds.end())
+    for (auto& fd : fds) {
+        if (std::find_if(qfds.begin(), qfds.end(), [&fd](T& _vfd) { return fd == _vfd; }) == qfds.end())
             return false;
     }
     return true;

tests/dhtproxytester.cpp

@@ -71,7 +71,7 @@ DhtProxyTester::tearDown() {
         cv.notify_all();
     });
     std::unique_lock<std::mutex> lk(cv_m);
-    CPPUNIT_ASSERT(cv.wait_for(lk, 15s, [&]{ return done; }));
+    CPPUNIT_ASSERT(cv.wait_for(lk, 5s, [&]{ return done; }));
     serverProxy.reset();
     nodeProxy.reset();
 }
@@ -289,47 +289,4 @@ DhtProxyTester::testFuzzy()
         CPPUNIT_ASSERT(value->data == mtu);
 }
 
-void
-DhtProxyTester::testShutdownStop()
-{
-    constexpr size_t N = 40000;
-    constexpr unsigned C = 100;
-
-    // Arrange
-    auto key = dht::InfoHash::get("testShutdownStop");
-    std::vector<std::shared_ptr<dht::Value>> values;
-    std::vector<uint8_t> mtu;
-    mtu.reserve(N);
-    for (size_t i = 0; i < N; i++)
-        mtu.emplace_back((i % 2) ? 'T' : 'M');
-
-    std::atomic_uint callback_count {0};
-
-    // Act
-    for (size_t i = 0; i < C; i++) {
-        auto nodeTest = std::make_shared<dht::DhtRunner>();
-        nodeTest->run(0, clientConfig);
-        nodeTest->put(key, dht::Value(mtu), [&](bool ok) {
-            callback_count++;
-        });
-        nodeTest->get(key, [&](const std::vector<std::shared_ptr<dht::Value>>& vals){
-            values.insert(values.end(), vals.begin(), vals.end());
-            return true;
-        },[&](bool ok){
-            callback_count++;
-        });
-        bool done = false;
-        std::condition_variable cv;
-        std::mutex cv_m;
-        nodeTest->shutdown([&]{
-            std::lock_guard<std::mutex> lk(cv_m);
-            done = true;
-            cv.notify_all();
-        }, true);
-        std::unique_lock<std::mutex> lk(cv_m);
-        CPPUNIT_ASSERT(cv.wait_for(lk, 10s, [&]{ return done; }));
-    }
-    CPPUNIT_ASSERT_EQUAL(2*C, callback_count.load());
-}
-
 }  // namespace test

tests/dhtproxytester.h

@@ -36,7 +36,6 @@ class DhtProxyTester : public CppUnit::TestFixture {
     CPPUNIT_TEST(testResubscribeGetValues);
     CPPUNIT_TEST(testPutGet40KChars);
     CPPUNIT_TEST(testFuzzy);
-    CPPUNIT_TEST(testShutdownStop);
     CPPUNIT_TEST_SUITE_END();
 
  public:
@@ -69,8 +68,6 @@ class DhtProxyTester : public CppUnit::TestFixture {
 
    void testFuzzy();
 
-   void testShutdownStop();
-
  private:
     dht::DhtRunner::Config clientConfig {};
     dht::DhtRunner nodePeer;

tests/dhtrunnertester.cpp

@@ -23,7 +23,6 @@
 #include <mutex>
 #include <condition_variable>
 using namespace std::chrono_literals;
-using namespace std::literals;
 
 namespace test {
 CPPUNIT_TEST_SUITE_REGISTRATION(DhtRunnerTester);
@@ -174,76 +173,6 @@ DhtRunnerTester::testListen() {
     node1.cancelListen(d, tokend);
 }
 
-void
-DhtRunnerTester::testIdOps() {
-    std::mutex mutex;
-    std::condition_variable cv;
-    unsigned valueCount(0);
-
-    dht::DhtRunner::Config config2;
-    config2.dht_config.node_config.max_peer_req_per_sec = -1;
-    config2.dht_config.node_config.max_req_per_sec = -1;
-    config2.dht_config.id = dht::crypto::generateIdentity();
-
-    dht::DhtRunner::Context context2;
-    context2.identityAnnouncedCb = [&](bool ok) {
-        CPPUNIT_ASSERT(ok);
-        std::lock_guard<std::mutex> lk(mutex);
-        valueCount++;
-        cv.notify_all();
-    };
-
-    node2.join();
-    node2.run(42232, config2, std::move(context2));
-    node2.bootstrap(node1.getBound());
-
-    node1.findCertificate(node2.getId(), [&](const std::shared_ptr<dht::crypto::Certificate>& crt){
-        CPPUNIT_ASSERT(crt);
-        std::lock_guard<std::mutex> lk(mutex);
-        valueCount++;
-        cv.notify_all();
-    });
-
-    {
-        std::unique_lock<std::mutex> lk(mutex);
-        CPPUNIT_ASSERT(cv.wait_for(lk, 20s, [&]{ return valueCount == 2; }));
-    }
-
-    dht::DhtRunner::Context context1;
-    context1.identityAnnouncedCb = [&](bool ok) {
-        CPPUNIT_ASSERT(ok);
-        std::lock_guard<std::mutex> lk(mutex);
-        valueCount++;
-        cv.notify_all();
-    };
-
-    config2.dht_config.id = dht::crypto::generateIdentity();
-    node1.join();
-    node1.run(42222, config2, std::move(context1));
-    node1.bootstrap(node2.getBound());
-
-    auto key = dht::InfoHash::get("key");
-    node1.putEncrypted(key, node2.getId(), dht::Value("yo"), [&](bool ok){
-        CPPUNIT_ASSERT(ok);
-        std::lock_guard<std::mutex> lk(mutex);
-        valueCount++;
-        cv.notify_all();
-    });
-
-    node2.listen<std::string>(key, [&](std::string&& value){
-        CPPUNIT_ASSERT_EQUAL("yo"s, value);
-        std::lock_guard<std::mutex> lk(mutex);
-        valueCount++;
-        cv.notify_all();
-        return false;
-    });
-
-    {
-        std::unique_lock<std::mutex> lk(mutex);
-        CPPUNIT_ASSERT(cv.wait_for(lk, 20s, [&]{ return valueCount == 4; }));
-    }
-}
-
 void
 DhtRunnerTester::testListenLotOfBytes() {
     std::mutex mutex;

tests/dhtrunnertester.h

@@ -33,7 +33,6 @@ class DhtRunnerTester : public CppUnit::TestFixture {
     CPPUNIT_TEST(testGetPut);
     CPPUNIT_TEST(testListen);
     CPPUNIT_TEST(testListenLotOfBytes);
-    CPPUNIT_TEST(testIdOps);
     CPPUNIT_TEST_SUITE_END();
 
     dht::DhtRunner node1 {};
@@ -59,10 +58,6 @@ class DhtRunnerTester : public CppUnit::TestFixture {
      * Test listen method
      */
     void testListen();
-    /**
-     * Test methods requiring a node identity
-     */
-    void testIdOps();
     /**
      * Test listen method with lot of datas
      */

tests/threadpooltester.cpp

@@ -59,7 +59,7 @@ ThreadPoolTester::testExecutor()
     auto executor8 = std::make_shared<dht::Executor>(pool, 8);
 
     constexpr unsigned N = 64 * 1024;
-    unsigned count1 {0};
+    std::atomic_uint count1 {0};
     std::atomic_uint count4 {0};
     std::atomic_uint count8 {0};
     for (unsigned i=0; i<N; i++) {
@@ -69,7 +69,7 @@ ThreadPoolTester::testExecutor()
     }
 
     auto start = clock::now();
-    while ((count1 != N ||
+    while ((count1.load() != N ||
             count4.load() != N ||
             count8.load() != N) && clock::now() - start < std::chrono::seconds(20))
     {
@@ -78,28 +78,11 @@ ThreadPoolTester::testExecutor()
     executor1.reset();
     executor4.reset();
     executor8.reset();
-    CPPUNIT_ASSERT_EQUAL(N, count1);
+    CPPUNIT_ASSERT_EQUAL(N, count1.load());
     CPPUNIT_ASSERT_EQUAL(N, count4.load());
     CPPUNIT_ASSERT_EQUAL(N, count8.load());
 }
 
-void
-ThreadPoolTester::testContext()
-{
-    std::atomic_uint count {0};
-    constexpr unsigned N = 64 * 1024;
-
-    {
-        dht::ExecutionContext ctx(dht::ThreadPool::computation());
-        for (unsigned i=0; i<N; i++) {
-            ctx.run([&] { count++; });
-        }
-    }
-
-    CPPUNIT_ASSERT_EQUAL(N, count.load());
-
-}
-
 void
 ThreadPoolTester::tearDown() {
 }

tests/threadpooltester.h

@@ -29,7 +29,6 @@ class ThreadPoolTester : public CppUnit::TestFixture {
     CPPUNIT_TEST_SUITE(ThreadPoolTester);
     CPPUNIT_TEST(testThreadPool);
     CPPUNIT_TEST(testExecutor);
-    CPPUNIT_TEST(testContext);
     CPPUNIT_TEST_SUITE_END();
 
  public:
@@ -44,7 +43,6 @@ class ThreadPoolTester : public CppUnit::TestFixture {
 
     void testThreadPool();
     void testExecutor();
-    void testContext();
 };
 
 }  // namespace test