thread pool: add ExecutionContext

include/opendht/dht.h

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

include/opendht/dht_interface.h

@@ -61,8 +61,10 @@ public:
 
     /**
      * Performs final operations before quitting.
+     * stop: if true, cancel ongoing operations and call their 'done'
+     *       callbacks synchronously.
      */
-    virtual void shutdown(ShutdownCallback cb) = 0;
+    virtual void shutdown(ShutdownCallback cb, bool stop = false) = 0;
 
     /**
      * Returns true if the node is running (have access to an open socket).

include/opendht/dht_proxy_client.h

@@ -85,7 +85,7 @@ public:
     /**
      * Performs final operations before quitting.
      */
-    void shutdown(ShutdownCallback cb) override;
+    void shutdown(ShutdownCallback cb, bool) override;
 
     /**
      * Returns true if the node is running (have access to an open socket).

include/opendht/dhtrunner.h

@@ -408,7 +408,7 @@ public:
     /**
      * Gracefuly disconnect from network.
      */
-    void shutdown(ShutdownCallback cb = {});
+    void shutdown(ShutdownCallback cb = {}, bool stop = false);
 
     /**
      * Quit and wait for all threads to terminate.

include/opendht/securedht.h

@@ -146,8 +146,8 @@ public:
     /**
      * SecureDht to Dht proxy
      */
-    void shutdown(ShutdownCallback cb) override {
-        dht_->shutdown(cb);
+    void shutdown(ShutdownCallback cb, bool stop = false) override {
+        dht_->shutdown(cb, stop);
     }
     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 > sizeof(sockaddr_storage))
+        if (length > static_cast<socklen_t>(sizeof(sockaddr_storage)))
             throw std::runtime_error("Socket address length is too large");
         set(sa, length);
     }

include/opendht/thread_pool.h

@@ -93,4 +93,71 @@ 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_;
+};
+
 }

src/dht.cpp

@@ -77,11 +77,27 @@ Dht::Kad::getStatus(time_point now) const
 }
 
 void
-Dht::shutdown(ShutdownCallback cb)
+Dht::shutdown(ShutdownCallback cb, bool stop)
 {
     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;

src/dht_proxy_client.cpp

@@ -242,7 +242,7 @@ DhtProxyClient::cancelAllListeners()
 }
 
 void
-DhtProxyClient::shutdown(ShutdownCallback cb)
+DhtProxyClient::shutdown(ShutdownCallback cb, bool)
 {
     stop();
     if (cb)

src/dhtrunner.cpp

@@ -119,22 +119,24 @@ 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()) {
+    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 (context.logger)
-                    context.logger->d("[runner %p] Using IPv4 port %hu from saved configuration", this, port);
-                if (local4.getPort() == 0)
+                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 (context.logger)
-                    context.logger->d("[runner %p] Using IPv6 port %hu from saved configuration", this, port);
-                if (local6.getPort() == 0)
+                if (local6.getPort() == 0) {
+                    if (context.logger)
+                        context.logger->d("[runner %p] Using IPv6 port %hu from saved configuration", this, port);
                     local6.setPort(port);
+                }
             }
         }
     }
@@ -266,29 +268,31 @@ DhtRunner::run(const Config& config, Context&& context)
 }
 
 void
-DhtRunner::shutdown(ShutdownCallback cb) {
+DhtRunner::shutdown(ShutdownCallback cb, bool stop) {
+    std::unique_lock<std::mutex> lck(storage_mtx);
     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) cb();
+        else if (cb) {
+            lck.unlock();
+            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_prio.emplace([=](SecureDht&) mutable {
+    pending_ops.emplace([=](SecureDht&) mutable {
         auto onShutdown = [this]{ opEnded(); };
 #ifdef OPENDHT_PROXY_CLIENT
         if (dht_via_proxy_)
-            dht_via_proxy_->shutdown(onShutdown);
+            dht_via_proxy_->shutdown(onShutdown, stop);
 #endif
         if (dht_)
-            dht_->shutdown(onShutdown);
+            dht_->shutdown(onShutdown, stop);
     });
     cv.notify_all();
 }
@@ -317,11 +321,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)
@@ -353,9 +357,13 @@ 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);
@@ -707,11 +715,12 @@ 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));
@@ -726,11 +735,12 @@ 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));
@@ -742,11 +752,12 @@ 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_++;
@@ -827,11 +838,12 @@ 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),
@@ -845,11 +857,12 @@ 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);
@@ -886,11 +899,12 @@ 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),
@@ -916,11 +930,12 @@ 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),
@@ -1006,11 +1021,12 @@ 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));
@@ -1021,9 +1037,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);
     });
@@ -1033,9 +1049,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);
@@ -1059,11 +1075,12 @@ 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){

src/value.cpp

@@ -496,10 +496,10 @@ Query::msgpack_unpack(const msgpack::object& o)
 }
 
 template <typename T>
-bool subset(std::vector<T> fds, std::vector<T> qfds)
+bool subset(const std::vector<T>& fds, const std::vector<T>& qfds)
 {
-    for (auto& fd : fds) {
-        if (std::find_if(qfds.begin(), qfds.end(), [&fd](T& _vfd) { return fd == _vfd; }) == qfds.end())
+    for (const auto& fd : fds) {
+        if (std::find_if(qfds.begin(), qfds.end(), [&fd](const 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, 5s, [&]{ return done; }));
+    CPPUNIT_ASSERT(cv.wait_for(lk, 15s, [&]{ return done; }));
     serverProxy.reset();
     nodeProxy.reset();
 }
@@ -289,4 +289,47 @@ 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,6 +36,7 @@ class DhtProxyTester : public CppUnit::TestFixture {
     CPPUNIT_TEST(testResubscribeGetValues);
     CPPUNIT_TEST(testPutGet40KChars);
     CPPUNIT_TEST(testFuzzy);
+    CPPUNIT_TEST(testShutdownStop);
     CPPUNIT_TEST_SUITE_END();
 
  public:
@@ -68,6 +69,8 @@ class DhtProxyTester : public CppUnit::TestFixture {
 
    void testFuzzy();
 
+   void testShutdownStop();
+
  private:
     dht::DhtRunner::Config clientConfig {};
     dht::DhtRunner nodePeer;

tests/threadpooltester.cpp

@@ -59,7 +59,7 @@ ThreadPoolTester::testExecutor()
     auto executor8 = std::make_shared<dht::Executor>(pool, 8);
 
     constexpr unsigned N = 64 * 1024;
-    std::atomic_uint count1 {0};
+    unsigned 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.load() != N ||
+    while ((count1 != N ||
             count4.load() != N ||
             count8.load() != N) && clock::now() - start < std::chrono::seconds(20))
     {
@@ -78,11 +78,28 @@ ThreadPoolTester::testExecutor()
     executor1.reset();
     executor4.reset();
     executor8.reset();
-    CPPUNIT_ASSERT_EQUAL(N, count1.load());
+    CPPUNIT_ASSERT_EQUAL(N, count1);
     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,6 +29,7 @@ class ThreadPoolTester : public CppUnit::TestFixture {
     CPPUNIT_TEST_SUITE(ThreadPoolTester);
     CPPUNIT_TEST(testThreadPool);
     CPPUNIT_TEST(testExecutor);
+    CPPUNIT_TEST(testContext);
     CPPUNIT_TEST_SUITE_END();
 
  public:
@@ -43,6 +44,7 @@ class ThreadPoolTester : public CppUnit::TestFixture {
 
     void testThreadPool();
     void testExecutor();
+    void testContext();
 };
 
 }  // namespace test