tools/c: adapt include directory

This reverts commit 5d771e119c.

CMakeLists.txt

@@ -1,4 +1,16 @@
 cmake_minimum_required (VERSION 3.1)
+if(POLICY CMP0065)
+  cmake_policy(SET CMP0065 NEW)
+endif()
+if(POLICY CMP0069)
+  cmake_policy(SET CMP0069 NEW)
+endif()
+if(POLICY CMP0073)
+  cmake_policy(SET CMP0073 NEW)
+endif()
+if(POLICY CMP0074)
+  cmake_policy(SET CMP0074 NEW)
+endif()
 project (opendht)
 
 include(CMakePackageConfigHelpers)
@@ -131,7 +143,7 @@ else ()
     set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} ${DISABLE_MSC_WARNINGS}")
     set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} /MP")
 endif ()
-set (CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -DMSGPACK_DISABLE_LEGACY_NIL -DMSGPACK_DISABLE_LEGACY_CONVERT")
+set (CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -DMSGPACK_NO_BOOST -DMSGPACK_DISABLE_LEGACY_NIL -DMSGPACK_DISABLE_LEGACY_CONVERT")
 
 if (NOT CMAKE_BUILD_TYPE)
     set(CMAKE_BUILD_TYPE Release)
@@ -416,6 +428,7 @@ if (OPENDHT_C)
     )
     install (TARGETS opendht-c DESTINATION ${CMAKE_INSTALL_LIBDIR} EXPORT opendht-c)
     install (FILES ${CMAKE_CURRENT_BINARY_DIR}/opendht-c.pc DESTINATION ${CMAKE_INSTALL_LIBDIR}/pkgconfig)
+    install (FILES c/opendht_c.h DESTINATION ${CMAKE_INSTALL_PREFIX}/include/opendht)
 endif ()
 
 if (OPENDHT_TOOLS)

c/opendht.cpp

@@ -97,10 +97,20 @@ const char* dht_value_get_user_type(const dht_value* data) {
     return vsp->user_type.c_str();
 }
 
+void dht_value_set_user_type(dht_value* data, const char* user_type) {
+    (*reinterpret_cast<ValueSp*>(data))->user_type = user_type;
+}
+
 dht_value* dht_value_new(const uint8_t* data, size_t size) {
     return reinterpret_cast<dht_value*>(new ValueSp(std::make_shared<dht::Value>(data, size)));
 }
 
+dht_value* dht_value_new_from_string(const char* str) {
+    ValueSp value = std::make_shared<dht::Value>((const uint8_t*)str, strlen(str));
+    value->user_type = "text/plain";
+    return reinterpret_cast<dht_value*>(new ValueSp(std::move(value)));
+}
+
 dht_value* dht_value_ref(const dht_value* v) {
     return reinterpret_cast<dht_value*>(new ValueSp(*reinterpret_cast<const ValueSp*>(v)));
 }

c/opendht_c.h

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

configure.ac

@@ -175,7 +175,7 @@ AM_COND_IF([PROXY_CLIENT_OR_SERVER], [
     AS_IF([test "x$http_parser_headers" != "xyes"], AC_MSG_ERROR([Missing HttpParser headers files]))
 ])
 
-CXXFLAGS="${CXXFLAGS} -DMSGPACK_DISABLE_LEGACY_NIL -DMSGPACK_DISABLE_LEGACY_CONVERT"
+CXXFLAGS="${CXXFLAGS} -DMSGPACK_NO_BOOST -DMSGPACK_DISABLE_LEGACY_NIL -DMSGPACK_DISABLE_LEGACY_CONVERT"
 
 AC_ARG_ENABLE([tools], AS_HELP_STRING([--disable-tools],[Disable tools (CLI DHT node)]),,build_tools=yes)
 AM_CONDITIONAL(ENABLE_TOOLS, test x$build_tools == xyes)

include/opendht/dht.h

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

include/opendht/dhtrunner.h

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

include/opendht/thread_pool.h

@@ -65,15 +65,14 @@ public:
     void join();
 
 private:
-    struct ThreadState;
-    std::queue<std::function<void()>> tasks_ {};
-    std::vector<std::unique_ptr<ThreadState>> threads_;
-    unsigned readyThreads_ {0};
     std::mutex lock_ {};
     std::condition_variable cv_ {};
+    std::queue<std::function<void()>> tasks_ {};
+    std::vector<std::unique_ptr<std::thread>> threads_;
+    unsigned readyThreads_ {0};
+    bool running_ {true};
 
     const unsigned maxThreads_;
-    bool running_ {true};
 };
 
 class OPENDHT_PUBLIC Executor : public std::enable_shared_from_this<Executor> {

include/opendht/value.h

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

src/dht.cpp

@@ -38,10 +38,12 @@ using namespace std::placeholders;
 
 constexpr std::chrono::minutes Dht::MAX_STORAGE_MAINTENANCE_EXPIRE_TIME;
 constexpr std::chrono::minutes Dht::SEARCH_EXPIRE_TIME;
+constexpr std::chrono::seconds Dht::BOOTSTRAP_PERIOD;
 constexpr duration Dht::LISTEN_EXPIRE_TIME;
 constexpr duration Dht::LISTEN_EXPIRE_TIME_PUBLIC;
 constexpr duration Dht::REANNOUNCE_MARGIN;
 static constexpr size_t MAX_REQUESTS_PER_SEC {8 * 1024};
+static constexpr duration BOOTSTRAP_PERIOD_MAX {std::chrono::hours(24)};
 
 NodeStatus
 Dht::updateStatus(sa_family_t af)
@@ -51,9 +53,9 @@ Dht::updateStatus(sa_family_t af)
     d.status = d.getStatus(scheduler.time());
     if (d.status != old) {
         auto& other = dht(af == AF_INET ? AF_INET6 : AF_INET);
-        if (other.status == NodeStatus::Disconnected && d.status == NodeStatus::Disconnected)
+        if (other.status == NodeStatus::Disconnected && d.status == NodeStatus::Disconnected) {
             onDisconnected();
-        else if (other.status == NodeStatus::Connected || d.status == NodeStatus::Connected) {
+        } else if (other.status == NodeStatus::Connected || d.status == NodeStatus::Connected) {
             onConnected();
         }
     }
@@ -1437,6 +1439,7 @@ Dht::connectivityChanged(sa_family_t af)
     reported_addr.erase(std::remove_if(reported_addr.begin(), reported_addr.end(), [&](const ReportedAddr& addr){
         return addr.second.getFamily() == af;
     }), reported_addr.end());
+    startBootstrap(); // will only happen if disconnected
 }
 
 void
@@ -2018,11 +2021,7 @@ Dht::expire()
 void
 Dht::onConnected()
 {
-    if (bootstrapJob) {
-        bootstrapJob->cancel();
-        bootstrapJob.reset();
-    }
-    bootstrap_period = std::chrono::seconds(10);
+    stopBootstrap();
     auto callbacks = std::move(onConnectCallbacks_);
     while (not callbacks.empty()) {
         callbacks.front()();
@@ -2032,6 +2031,13 @@ Dht::onConnected()
 
 void
 Dht::onDisconnected()
+{
+    if (not bootstrapJob)
+        bootstrap();
+}
+
+void
+Dht::bootstrap()
 {
     if (dht4.status != NodeStatus::Disconnected || dht6.status != NodeStatus::Disconnected)
         return;
@@ -2052,8 +2058,25 @@ Dht::onDisconnected()
     }
     if (bootstrapJob)
         bootstrapJob->cancel();
-    bootstrapJob = scheduler.add(scheduler.time() + bootstrap_period, std::bind(&Dht::onDisconnected, this));
-    bootstrap_period *= 2;
+    bootstrapJob = scheduler.add(scheduler.time() + bootstrap_period, std::bind(&Dht::bootstrap, this));
+    bootstrap_period = std::min(bootstrap_period * 2, BOOTSTRAP_PERIOD_MAX);
+}
+
+void
+Dht::startBootstrap()
+{
+    stopBootstrap();
+    bootstrapJob = scheduler.add(scheduler.time(), std::bind(&Dht::bootstrap, this));
+}
+
+void
+Dht::stopBootstrap()
+{
+    if (bootstrapJob) {
+        bootstrapJob->cancel();
+        bootstrapJob.reset();
+    }
+    bootstrap_period = BOOTSTRAP_PERIOD;
 }
 
 void

src/dhtrunner.cpp

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

src/storage.h

@@ -37,8 +37,7 @@ public:
         storedValues_.emplace(expiration, std::pair<InfoHash, Value::Id>(id, value.id));
     }
     void erase(const InfoHash& id, const Value& value, time_point expiration) {
-        auto size = value.size();
-        totalSize_ -= size;
+        totalSize_ -= value.size();
         auto range = storedValues_.equal_range(expiration);
         for (auto rit = range.first; rit != range.second;) {
             if (rit->second.first == id && rit->second.second == value.id) {

src/thread_pool.cpp

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

tests/threadpooltester.cpp

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

tools/CMakeLists.txt

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

tools/dhtcnode.c

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