update version to 2.4.4

Fix typo in 2.4.1 release

CMakeLists.txt

@@ -19,7 +19,7 @@ include(FindPkgConfig)
 include(cmake/CheckAtomic.cmake)
 
 set (opendht_VERSION_MAJOR 2)
-set (opendht_VERSION_MINOR 4.0)
+set (opendht_VERSION_MINOR 4.4)
 set (opendht_VERSION ${opendht_VERSION_MAJOR}.${opendht_VERSION_MINOR})
 set (PACKAGE_VERSION ${opendht_VERSION})
 set (VERSION "${opendht_VERSION}")
@@ -403,15 +403,23 @@ if (OPENDHT_SHARED)
 endif ()
 
 if (OPENDHT_C)
-    add_library (opendht-c SHARED
-        c/opendht.cpp
-        c/opendht_c.h
-    )
-    target_compile_definitions(opendht-c PRIVATE OPENDHT_C_BUILD)
     if (OPENDHT_SHARED)
-        target_link_libraries(opendht-c opendht)
-    else ()
-        target_link_libraries(opendht-c opendht-static)
+        add_library (opendht-c SHARED
+            c/opendht.cpp
+            c/opendht_c.h
+        )
+        target_compile_definitions(opendht-c PRIVATE OPENDHT_C_BUILD)
+        target_link_libraries(opendht-c PRIVATE opendht)
+        install (TARGETS opendht-c DESTINATION ${CMAKE_INSTALL_LIBDIR} EXPORT opendht-c)
+    endif ()
+
+    if (OPENDHT_STATIC)
+        add_library (opendht-c-static STATIC
+            c/opendht.cpp
+            c/opendht_c.h
+        )
+        target_link_libraries(opendht-c-static PRIVATE opendht-static)
+        install (TARGETS opendht-c-static DESTINATION ${CMAKE_INSTALL_LIBDIR} EXPORT opendht-c-static)
     endif ()
 
     # PkgConfig module
@@ -420,7 +428,6 @@ if (OPENDHT_C)
         opendht-c.pc
         @ONLY
     )
-    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 ()

c/opendht.cpp

@@ -32,6 +32,11 @@ extern "C" {
 
 #include <errno.h>
 
+const char* dht_version()
+{
+    return dht::version();
+}
+
 // dht::InfoHash
 
 inline dht_infohash dht_infohash_to_c(const dht::InfoHash& h) {
@@ -71,7 +76,11 @@ bool dht_infohash_is_zero(const dht_infohash* h) {
 }
 
 void dht_infohash_from_hex(dht_infohash* h, const char* dat) {
-    *h = dht_infohash_to_c(dht::InfoHash(std::string(dat, HASH_LEN*2)));
+    *h = dht_infohash_to_c(dht::InfoHash(std::string_view(dat, HASH_LEN*2)));
+}
+
+void dht_infohash_from_hex_null(dht_infohash* h, const char* dat) {
+    *h = dht_infohash_to_c(dht::InfoHash(std::string_view(dat)));
 }
 
 const char* dht_pkid_print(const dht_pkid* h) {

c/opendht_c.h

@@ -36,6 +36,8 @@ struct OPENDHT_C_PUBLIC dht_data_view {
 };
 typedef struct dht_data_view dht_data_view;
 
+OPENDHT_C_PUBLIC const char* dht_version(void);
+
 // dht::Blob
 struct OPENDHT_C_PUBLIC dht_blob;
 typedef struct dht_blob dht_blob;
@@ -48,6 +50,7 @@ typedef struct dht_infohash dht_infohash;
 OPENDHT_C_PUBLIC void dht_infohash_zero(dht_infohash* h);
 OPENDHT_C_PUBLIC void dht_infohash_random(dht_infohash* h);
 OPENDHT_C_PUBLIC void dht_infohash_from_hex(dht_infohash* h, const char* dat);
+OPENDHT_C_PUBLIC void dht_infohash_from_hex_null(dht_infohash* h, const char* dat);
 OPENDHT_C_PUBLIC void dht_infohash_get(dht_infohash* h, const uint8_t* dat, size_t dat_size);
 OPENDHT_C_PUBLIC void dht_infohash_get_from_string(dht_infohash* h, const char* str);
 OPENDHT_C_PUBLIC const char* dht_infohash_print(const dht_infohash* h);

configure.ac

@@ -1,7 +1,7 @@
 dnl define macros
 m4_define([opendht_major_version], 2)
 m4_define([opendht_minor_version], 4)
-m4_define([opendht_patch_version], 0)
+m4_define([opendht_patch_version], 3)
 m4_define([opendht_version],
 		  [opendht_major_version.opendht_minor_version.opendht_patch_version])
 

include/opendht/callbacks.h

@@ -82,6 +82,8 @@ struct OPENDHT_PUBLIC NodeInfo {
     NodeStats ipv4 {};
     NodeStats ipv6 {};
     size_t ongoing_ops {0};
+    size_t storage_values {0};
+    size_t storage_size {0};
     in_port_t bound4 {0};
     in_port_t bound6 {0};
 
@@ -132,6 +134,9 @@ struct OPENDHT_PUBLIC Config {
     /* If non-0, overrides the default maximum store size. -1 means no limit.  */
     ssize_t max_store_size {0};
 
+    /* If non-0, overrides the default maximum store key count. -1 means no limit.  */
+    ssize_t max_store_keys {0};
+
     /**
      * Use appropriate bahavior for a public IP, stable node:
      *   - No connectivity change triggered when a search fails

include/opendht/dht.h

@@ -302,6 +302,9 @@ public:
     void setStorageLimit(size_t limit = DEFAULT_STORAGE_LIMIT) override {
         max_store_size = limit;
     }
+    size_t getStorageLimit() const override {
+        return max_store_size;
+    }
 
     /**
      * Returns the total memory usage of stored values and the number
@@ -340,7 +343,7 @@ private:
     static constexpr unsigned LISTEN_NODES {4};
 
     /* The maximum number of hashes we're willing to track. */
-    static constexpr unsigned MAX_HASHES {1024 * 1024};
+    static constexpr unsigned MAX_HASHES {1024 * 1024 * 1024};
 
     /* The maximum number of searches we keep data about. */
     static constexpr unsigned MAX_SEARCHES {1024 * 1024};
@@ -447,13 +450,13 @@ private:
     // Storage
     void storageAddListener(const InfoHash& id, const Sp<Node>& node, size_t tid, Query&& = {}, int version = 0);
     bool storageStore(const InfoHash& id, const Sp<Value>& value, time_point created, const SockAddr& sa = {}, bool permanent = false);
-    bool storageErase(const InfoHash& id, Value::Id vid);
     bool storageRefresh(const InfoHash& id, Value::Id vid);
     void expireStore();
     void expireStorage(InfoHash h);
     void expireStore(decltype(store)::iterator);
 
-    void storageChanged(const InfoHash& id, Storage& st, ValueStorage&, bool newValue);
+    void storageRemoved(const InfoHash& id, Storage& st, const std::vector<Sp<Value>>& values, size_t totalSize);
+    void storageChanged(const InfoHash& id, Storage& st, const Sp<Value>&, bool newValue);
     std::string printStorageLog(const decltype(store)::value_type&) const;
 
     /**

include/opendht/dht_interface.h

@@ -221,6 +221,7 @@ public:
      * Set the in-memory storage limit in bytes
      */
     virtual void setStorageLimit(size_t limit = DEFAULT_STORAGE_LIMIT) = 0;
+    virtual size_t getStorageLimit() const = 0;
 
     /**
      * Returns the total memory usage of stored values and the number

include/opendht/dht_proxy_client.h

@@ -263,6 +263,7 @@ public:
     void dumpTables() const override {}
     std::vector<unsigned> getNodeMessageStats(bool) override { return {}; }
     void setStorageLimit(size_t) override {}
+    virtual size_t getStorageLimit() const { return 0; }
     void connectivityChanged(sa_family_t) override {
         getProxyInfos();
     }

include/opendht/dhtrunner.h

@@ -335,6 +335,7 @@ public:
 
     std::pair<size_t, size_t> getStoreSize() const;
 
+    void getStorageLimit() const;
     void setStorageLimit(size_t limit = DEFAULT_STORAGE_LIMIT);
 
     std::vector<NodeExport> exportNodes() const;

include/opendht/network_utils.h

@@ -42,7 +42,7 @@ namespace dht {
 namespace net {
 
 static const constexpr in_port_t DHT_DEFAULT_PORT = 4222;
-static const constexpr size_t RX_QUEUE_MAX_SIZE = 1024 * 16;
+static const constexpr size_t RX_QUEUE_MAX_SIZE = 1024 * 64;
 static const constexpr std::chrono::milliseconds RX_QUEUE_MAX_DELAY(650);
 
 int bindSocket(const SockAddr& addr, SockAddr& bound);

include/opendht/securedht.h

@@ -173,6 +173,10 @@ public:
     void setStorageLimit(size_t limit = DEFAULT_STORAGE_LIMIT) override {
         dht_->setStorageLimit(limit);
     }
+    size_t getStorageLimit() const override {
+        return dht_->getStorageLimit();
+    }
+
     std::vector<NodeExport> exportNodes() const override {
         return dht_->exportNodes();
     }

include/opendht/value.h

@@ -638,8 +638,10 @@ struct OPENDHT_PUBLIC Value
      */
     unsigned priority {0};
 
+    bool checkSignature();
+    Sp<Value> decrypt(const crypto::PrivateKey& key);
+
 private:
-    friend class SecureDht;
     /* Cache for crypto ops */
     bool signatureChecked {false};
     bool signatureValid {false};

src/dht.cpp

@@ -1194,15 +1194,20 @@ Dht::cancelPut(const InfoHash& id, const Value::Id& vid)
     };
     canceled |= sr_cancel_put(dht4.searches);
     canceled |= sr_cancel_put(dht6.searches);
-    if (canceled)
-        storageErase(id, vid);
+    if (canceled) {
+        auto st = store.find(id);
+        if (st != store.end()) {
+            if (auto value = st->second.remove(id, vid))
+                storageRemoved(id, st->second, {value}, value->size());
+        }
+    }
     return canceled;
 }
 
 // Storage
 
 void
-Dht::storageChanged(const InfoHash& id, Storage& st, ValueStorage& v, bool newValue)
+Dht::storageChanged(const InfoHash& id, Storage& st, const Sp<Value>& v, bool newValue)
 {
     if (newValue) {
         if (not st.local_listeners.empty()) {
@@ -1212,8 +1217,8 @@ Dht::storageChanged(const InfoHash& id, Storage& st, ValueStorage& v, bool newVa
             cbs.reserve(st.local_listeners.size());
             for (const auto& l : st.local_listeners) {
                 std::vector<Sp<Value>> vals;
-                if (not l.second.filter or l.second.filter(*v.data))
-                    vals.push_back(v.data);
+                if (not l.second.filter or l.second.filter(*v))
+                    vals.push_back(v);
                 if (not vals.empty()) {
                     if (logger_)
                         logger_->d(id, "[store %s] sending update local listener with token %lu",
@@ -1234,14 +1239,14 @@ Dht::storageChanged(const InfoHash& id, Storage& st, ValueStorage& v, bool newVa
         for (const auto& node_listeners : st.listeners) {
             for (const auto& l : node_listeners.second) {
                 auto f = l.second.query.where.getFilter();
-                if (f and not f(*v.data))
+                if (f and not f(*v))
                     continue;
                 if (logger_)
                     logger_->w(id, node_listeners.first->id, "[store %s] [node %s] sending update",
                         id.toString().c_str(),
                         node_listeners.first->toString().c_str());
                 std::vector<Sp<Value>> vals {};
-                vals.push_back(v.data);
+                vals.push_back(v);
                 Blob ntoken = makeToken(node_listeners.first->getAddr(), false);
                 network_engine.tellListener(node_listeners.first, l.first, id, 0, ntoken, {}, {},
                         std::move(vals), l.second.query, l.second.version);
@@ -1282,26 +1287,18 @@ Dht::storageStore(const InfoHash& id, const Sp<Value>& value, time_point created
             vs->expiration_job = scheduler.add(expiration, std::bind(&Dht::expireStorage, this, id));
         }
         if (total_store_size > max_store_size) {
+            auto value = vs->data;
+            auto value_diff = store.second.values_diff;
             expireStore();
+            storageChanged(id, st->second, value, value_diff > 0);
+        } else {
+            storageChanged(id, st->second, vs->data, store.second.values_diff > 0);
         }
-        storageChanged(id, st->second, *vs, store.second.values_diff > 0);
     }
 
     return std::get<0>(store);
 }
 
-bool
-Dht::storageErase(const InfoHash& id, Value::Id vid)
-{
-    auto st = store.find(id);
-    if (st == store.end())
-        return false;
-    auto ret = st->second.remove(id, vid);
-    total_store_size += ret.size_diff;
-    total_values += ret.values_diff;
-    return ret.values_diff;
-}
-
 void
 Dht::storageAddListener(const InfoHash& id, const Sp<Node>& node, size_t socket_id, Query&& query, int version)
 {
@@ -1333,36 +1330,8 @@ Dht::expireStore(decltype(store)::iterator i)
     const auto& id = i->first;
     auto& st = i->second;
     auto stats = st.expire(id, scheduler.time());
-    total_store_size += stats.first;
-    total_values -= stats.second.size();
     if (not stats.second.empty()) {
-        if (logger_)
-            logger_->d(id, "[store %s] discarded %ld expired values (%ld bytes)",
-            id.toString().c_str(), stats.second.size(), -stats.first);
-
-        if (not st.listeners.empty()) {
-            if (logger_)
-                logger_->d(id, "[store %s] %lu remote listeners", id.toString().c_str(), st.listeners.size());
-
-            std::vector<Value::Id> ids;
-            ids.reserve(stats.second.size());
-            for (const auto& v : stats.second)
-                ids.emplace_back(v->id);
-
-            for (const auto& node_listeners : st.listeners) {
-                for (const auto& l : node_listeners.second) {
-                    if (logger_)
-                        logger_->w(id, node_listeners.first->id, "[store %s] [node %s] sending expired",
-                            id.toString().c_str(),
-                            node_listeners.first->toString().c_str());
-                    Blob ntoken = makeToken(node_listeners.first->getAddr(), false);
-                    network_engine.tellListenerExpired(node_listeners.first, l.first, id, ntoken, ids, l.second.version);
-                }
-            }
-        }
-        for (const auto& local_listeners : st.local_listeners) {
-            local_listeners.second.get_cb(stats.second, true);
-        }
+        storageRemoved(id, st, stats.second, -stats.first);
     }
 }
 
@@ -1374,6 +1343,41 @@ Dht::expireStorage(InfoHash h)
         expireStore(i);
 }
 
+void
+Dht::storageRemoved(const InfoHash& id, Storage& st, const std::vector<Sp<Value>>& values, size_t totalSize)
+{
+    if (logger_)
+        logger_->d(id, "[store %s] discarded %ld values (%ld bytes)",
+        id.toString().c_str(), values.size(), totalSize);
+
+    total_store_size -= totalSize;
+    total_values -= values.size();
+
+    if (not st.listeners.empty()) {
+        if (logger_)
+            logger_->d(id, "[store %s] %lu remote listeners", id.toString().c_str(), st.listeners.size());
+
+        std::vector<Value::Id> ids;
+        ids.reserve(values.size());
+        for (const auto& v : values)
+            ids.emplace_back(v->id);
+
+        for (const auto& node_listeners : st.listeners) {
+            for (const auto& l : node_listeners.second) {
+                if (logger_)
+                    logger_->w(id, node_listeners.first->id, "[store %s] [node %s] sending expired",
+                        id.toString().c_str(),
+                        node_listeners.first->toString().c_str());
+                Blob ntoken = makeToken(node_listeners.first->getAddr(), false);
+                network_engine.tellListenerExpired(node_listeners.first, l.first, id, ntoken, ids, l.second.version);
+            }
+        }
+    }
+    for (const auto& local_listeners : st.local_listeners) {
+        local_listeners.second.get_cb(values, true);
+    }
+}
+
 void
 Dht::expireStore()
 {
@@ -1399,23 +1403,23 @@ Dht::expireStore()
             break;
         }
         auto largest = store_quota.begin();
-        for (auto it = ++largest; it != store_quota.end(); ++it) {
+        for (auto it = std::next(largest); it != store_quota.end(); ++it) {
             if (it->second.size() > largest->second.size())
                 largest = it;
         }
-        if (logger_)
-            logger_->w("No space left: discarding value of largest consumer %s", largest->first.toString().c_str());
-        while (true) {
-            auto exp_value = largest->second.getOldest();
-            auto storage = store.find(exp_value.first);
-            if (storage != store.end()) {
-                auto ret = storage->second.remove(exp_value.first, exp_value.second);
-                total_store_size += ret.size_diff;
-                total_values += ret.values_diff;
-                if (logger_)
-                    logger_->w("Discarded %ld bytes, still %ld used", largest->first.toString().c_str(), total_store_size);
-                if (ret.values_diff)
-                    break;
+        if (largest != store_quota.end()) {
+            while (true) {
+                auto exp_value = largest->second.getOldest();
+                auto storage = store.find(exp_value.first);
+                if (storage != store.end()) {
+                    if (logger_)
+                        logger_->w("Storage quota full: discarding value from %s at %s %016" PRIx64, largest->first.toString().c_str(), exp_value.first.to_c_str(), exp_value.second);
+
+                    if (auto value = storage->second.remove(exp_value.first, exp_value.second)) {
+                        storageRemoved(storage->first, storage->second, {value}, value->size());
+                        break;
+                    }
+                }
             }
         }
     }
@@ -1664,7 +1668,7 @@ Dht::dumpTables() const
 std::string
 Dht::getStorageLog() const
 {
-    std::stringstream out;
+    std::ostringstream out;
     for (const auto& s : store)
         out << printStorageLog(s);
     out << std::endl << std::endl;
@@ -1789,7 +1793,8 @@ Dht::Dht(std::unique_ptr<net::DatagramSocket>&& sock, const Config& config, cons
     myid(config.node_id ? config.node_id : InfoHash::getRandom(rd)),
     store(),
     store_quota(),
-    max_store_keys(config.max_store_size ? (int)config.max_store_size : MAX_HASHES),
+    max_store_keys(config.max_store_keys ? (int)config.max_store_keys : MAX_HASHES),
+    max_store_size(config.max_store_size ? (int)config.max_store_size : DEFAULT_STORAGE_LIMIT),
     max_searches(config.max_searches ? (int)config.max_searches : MAX_SEARCHES),
     network_engine(myid, fromDhtConfig(config), std::move(sock), logger_, rd, scheduler,
             std::bind(&Dht::onError, this, _1, _2),
@@ -2558,7 +2563,7 @@ Dht::storageRefresh(const InfoHash& id, Value::Id vid)
             }
         }
 
-        auto expiration = s->second.refresh(now, vid, types);
+        auto expiration = s->second.refresh(id, now, vid, types);
         if (expiration.first) {
             scheduler.cancel(expiration.first->expiration_job);
             if (expiration.second != time_point::max()) {

src/dhtrunner.cpp

@@ -163,14 +163,15 @@ DhtRunner::run(const Config& config, Context&& context)
             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));
+                size_t dropped = 0;
+                while (rcv.size() > net::RX_QUEUE_MAX_SIZE) {
+                    rcv.pop_front();
+                    dropped++;
+                }
+                if (dropped and logger_) {
+                    logger_->w("[runner %p] dropped %zu packets: queue is full!", this, dropped);
+                }
                 ret = std::move(rcv_free);
             }
             cv.notify_all();
@@ -297,16 +298,24 @@ DhtRunner::shutdown(ShutdownCallback cb, bool stop) {
     }
     if (logger_)
         logger_->d("[runner %p] state changed to Stopping, %zu ongoing ops", this, ongoing_ops.load());
+#ifdef OPENDHT_PROXY_CLIENT
+    ongoing_ops += 2;
+#else
     ongoing_ops++;
+#endif
     shutdownCallbacks_.emplace_back(std::move(cb));
     pending_ops.emplace([=](SecureDht&) mutable {
         auto onShutdown = [this]{ opEnded(); };
 #ifdef OPENDHT_PROXY_CLIENT
         if (dht_via_proxy_)
             dht_via_proxy_->shutdown(onShutdown, stop);
+        else
+            opEnded();
 #endif
         if (dht_)
             dht_->shutdown(onShutdown, stop);
+        else
+            opEnded();
     });
     cv.notify_all();
 }
@@ -563,6 +572,7 @@ DhtRunner::getNodeInfo(std::function<void(std::shared_ptr<NodeInfo>)> cb)
         info.node_id = dht.getNodeId();
         info.ipv4 = dht.getNodesStats(AF_INET);
         info.ipv6 = dht.getNodesStats(AF_INET6);
+        std::tie(info.storage_size, info.storage_values) = dht.getStoreSize();
         if (auto sock = dht.getSocket()) {
             info.bound4 = sock->getBoundRef(AF_INET).getPort();
             info.bound6 = sock->getBoundRef(AF_INET6).getPort();
@@ -823,12 +833,16 @@ DhtRunner::cancelListen(InfoHash h, size_t token)
 #ifdef OPENDHT_PROXY_CLIENT
     pending_ops.emplace([=](SecureDht&) {
         auto it = listeners_.find(token);
-        if (it == listeners_.end()) return;
-        if (it->second.tokenClassicDht)
-            dht_->cancelListen(h, it->second.tokenClassicDht);
-        if (it->second.tokenProxyDht and dht_via_proxy_)
-            dht_via_proxy_->cancelListen(h, it->second.tokenProxyDht);
-        listeners_.erase(it);
+        if (it != listeners_.end()) {
+            if (it->second.tokenClassicDht)
+                dht_->cancelListen(h, it->second.tokenClassicDht);
+            if (it->second.tokenProxyDht and dht_via_proxy_)
+                dht_via_proxy_->cancelListen(h, it->second.tokenProxyDht);
+            listeners_.erase(it);
+        } else {
+            if (logger_)
+                logger_->w("[runner %p] cancelListen: unknown token %zu.", this, token);
+        }
         opEnded();
     });
 #else
@@ -849,13 +863,18 @@ DhtRunner::cancelListen(InfoHash h, std::shared_future<size_t> ftoken)
     ongoing_ops++;
 #ifdef OPENDHT_PROXY_CLIENT
     pending_ops.emplace([this, h, ftoken = std::move(ftoken)](SecureDht&) {
-        auto it = listeners_.find(ftoken.get());
-        if (it == listeners_.end()) return;
-        if (it->second.tokenClassicDht)
-            dht_->cancelListen(h, it->second.tokenClassicDht);
-        if (it->second.tokenProxyDht and dht_via_proxy_)
-            dht_via_proxy_->cancelListen(h, it->second.tokenProxyDht);
-        listeners_.erase(it);
+        auto token = ftoken.get();
+        auto it = listeners_.find(token);
+        if (it != listeners_.end()) {
+            if (it->second.tokenClassicDht)
+                dht_->cancelListen(h, it->second.tokenClassicDht);
+            if (it->second.tokenProxyDht and dht_via_proxy_)
+                dht_via_proxy_->cancelListen(h, it->second.tokenProxyDht);
+            listeners_.erase(it);
+        } else {
+            if (logger_)
+                logger_->w("[runner %p] cancelListen: unknown token %zu.", this, token);
+        }
         opEnded();
     });
 #else

src/network_engine.cpp

@@ -1155,11 +1155,12 @@ NetworkEngine::sendAnnounceValue(const Sp<Node>& n,
     msgpack::packer<msgpack::sbuffer> pk(&buffer);
     pk.pack_map(5+(config.network?1:0));
 
-    pk.pack(KEY_A); pk.pack_map((created < scheduler.time() ? 5 : 4));
+    bool add_created = created < scheduler.time();
+    pk.pack(KEY_A); pk.pack_map(add_created ? 5 : 4);
       pk.pack(KEY_REQ_ID);     pk.pack(myid);
       pk.pack(KEY_REQ_H);      pk.pack(infohash);
       auto v = packValueHeader(buffer, {value});
-      if (created < scheduler.time()) {
+      if (add_created) {
           pk.pack(KEY_REQ_CREATION);
           pk.pack(to_time_t(created));
       }

src/securedht.cpp

@@ -72,36 +72,20 @@ ValueType
 SecureDht::secureType(ValueType&& type)
 {
     type.storePolicy = [this,type](InfoHash id, Sp<Value>& v, const InfoHash& nid, const SockAddr& a) {
-        if (v->isSigned()) {
-            if (!v->signatureChecked) {
-                v->signatureChecked = true;
-                v->signatureValid = v->owner and v->owner->checkSignature(v->getToSign(), v->signature);
-            }
-            if (!v->signatureValid) {
-                if (logger_)
-                    logger_->w("Signature verification failed");
-                return false;
-            }
-        }
+        if (v->isSigned())
+            return v->checkSignature();
         return type.storePolicy(id, v, nid, a);
     };
     type.editPolicy = [this,type](InfoHash id, const Sp<Value>& o, Sp<Value>& n, const InfoHash& nid, const SockAddr& a) {
-        if (!o->isSigned())
+        if (not o->isSigned())
             return type.editPolicy(id, o, n, nid, a);
-        if (o->owner != n->owner) {
-            if (logger_)
-                logger_->w("Edition forbidden: owner changed.");
-            return false;
-        }
-        if (!n->signatureChecked) {
-            n->signatureChecked = true;
-            n->signatureValid = o->owner and o->owner->checkSignature(n->getToSign(), n->signature);
-        }
-        if (!n->signatureValid) {
+        if (o->owner != n->owner or not n->isSigned()) {
             if (logger_)
                 logger_->w("Edition forbidden: signature verification failed.");
             return false;
         }
+        if (not n->checkSignature())
+            return false;
         if (o->seq == n->seq) {
             // If the data is exactly the same,
             // it can be reannounced, possibly by someone else.
@@ -253,19 +237,12 @@ SecureDht::checkValue(const Sp<Value>& v)
 #endif
             return {};
         }
-        if (v->decrypted) {
-            return v->decryptedValue;
-        }
-        v->decrypted = true;
         try {
-            Value decrypted_val (decrypt(*v));
-            if (decrypted_val.recipient == getId()) {
-                if (decrypted_val.owner)
-                    nodesPubKeys_[decrypted_val.owner->getId()] = decrypted_val.owner;
-                v->decryptedValue = std::make_shared<Value>(std::move(decrypted_val));
-                return v->decryptedValue;
+            if (auto decrypted_val = v->decrypt(*key_)) {
+                if (decrypted_val->owner)
+                    nodesPubKeys_[decrypted_val->owner->getId()] = decrypted_val->owner;
+                return decrypted_val;
             }
-            // Ignore values belonging to other people
         } catch (const std::exception& e) {
             if (logger_)
                 logger_->w("Could not decrypt value %s : %s", v->toString().c_str(), e.what());
@@ -273,17 +250,11 @@ SecureDht::checkValue(const Sp<Value>& v)
     }
     // Check signed values
     else if (v->isSigned()) {
-        if (v->signatureChecked) {
-            return v->signatureValid ? v : Sp<Value>{};
-        }
-        v->signatureChecked = true;
-        if (v->owner and v->owner->checkSignature(v->getToSign(), v->signature)) {
-            v->signatureValid = true;
-            if (enableCache_)
+        if (v->checkSignature()) {
+            if (enableCache_ and v->owner)
                 nodesPubKeys_[v->owner->getId()] = v->owner;
             return v;
-        }
-        else if (logger_)
+        } else if (logger_)
             logger_->w("Signature verification failed for %s", v->toString().c_str());
     }
     // Forward normal values

src/storage.h

@@ -37,18 +37,32 @@ 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 range = storedValues_.equal_range(expiration);
         for (auto rit = range.first; rit != range.second;) {
             if (rit->second.first == id && rit->second.second == value.id) {
+                totalSize_ -= value.size();
                 storedValues_.erase(rit);
-                break;
+                return;
             } else
                 ++rit;
         }
+        // printf("StorageBucket::erase can't find value %s %016" PRIx64 "\n", id.to_c_str(), value.id);
+    }
+    void refresh(const InfoHash& id, const Value& value, time_point old_expiration, time_point expiration) {
+        auto range = storedValues_.equal_range(old_expiration);
+        for (auto rit = range.first; rit != range.second;) {
+            if (rit->second.first == id && rit->second.second == value.id) {
+                storedValues_.erase(rit);
+                storedValues_.emplace(expiration, std::pair<InfoHash, Value::Id>(id, value.id));
+                return;
+            } else
+                ++rit;
+        }
+        // printf("StorageBucket::refresh can't find value %s %016" PRIx64 "\n", id.to_c_str(), value.id);
+        insert(id, value, expiration);
     }
     size_t size() const { return totalSize_; }
-    std::pair<InfoHash, Value::Id> getOldest() const { return storedValues_.begin()->second; }
+    std::pair<InfoHash, Value::Id> getOldest() const { return storedValues_.empty() ? std::pair<InfoHash, Value::Id>{} : storedValues_.begin()->second; }
 private:
     std::multimap<time_point, std::pair<InfoHash, Value::Id>> storedValues_;
     size_t totalSize_ {0};
@@ -74,7 +88,7 @@ struct Storage {
     size_t listener_token {1};
 
     /* The maximum number of values we store for a given hash. */
-    static constexpr unsigned MAX_VALUES {1024};
+    static constexpr unsigned MAX_VALUES {64 * 1024};
 
     /**
      * Changes caused by an operation on the storage.
@@ -144,11 +158,14 @@ struct Storage {
      * @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) {
+    refresh(const InfoHash& id, 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);
+                auto oldExp = vs.expiration;
+                vs.expiration = std::max(oldExp, now + types.getType(vs.data->type).expiration);
+                if (vs.store_bucket)
+                    vs.store_bucket->refresh(id, *vs.data, oldExp, vs.expiration);
                 return {&vs, vs.expiration};
             }
         return {nullptr, time_point::max()};
@@ -160,7 +177,7 @@ struct Storage {
         local_listeners.erase(token);
     }
 
-    StoreDiff remove(const InfoHash& id, Value::Id);
+    Sp<Value> remove(const InfoHash& id, Value::Id);
 
     std::pair<ssize_t, std::vector<Sp<Value>>> expire(const InfoHash& id, time_point now);
 
@@ -229,7 +246,7 @@ Storage::store(const InfoHash& id, const Sp<Value>& value, time_point created, t
     return std::make_pair(nullptr, StoreDiff{});
 }
 
-Storage::StoreDiff
+Sp<Value>
 Storage::remove(const InfoHash& id, Value::Id vid)
 {
     auto it = std::find_if (values.begin(), values.end(), [&](const ValueStorage& vr) {
@@ -243,8 +260,9 @@ Storage::remove(const InfoHash& id, Value::Id vid)
     if (it->expiration_job)
         it->expiration_job->cancel();
     total_size -= size;
+    auto value = it->data;
     values.erase(it);
-    return {-size, -1, 0};
+    return value;
 }
 
 Storage::StoreDiff

src/utils.cpp

@@ -106,7 +106,7 @@ print_addr(const sockaddr* sa, socklen_t slen)
 {
     char hbuf[NI_MAXHOST];
     char sbuf[NI_MAXSERV];
-    std::stringstream out;
+    std::ostringstream out;
     if (sa and slen and !getnameinfo(sa, slen, hbuf, sizeof(hbuf), sbuf, sizeof(sbuf), NI_NUMERICHOST | NI_NUMERICSERV)) {
         if (sa->sa_family == AF_INET6)
             out << "[" << hbuf << "]";

src/value.cpp

@@ -239,6 +239,41 @@ Value::toJson() const
     return val;
 }
 
+bool
+Value::checkSignature()
+{
+    if (!signatureChecked) {
+        signatureChecked = true;
+        if (isSigned()) {
+            signatureValid = owner and owner->checkSignature(getToSign(), signature);
+        } else {
+            signatureValid = true;
+        }
+    }
+    return signatureValid;
+}
+
+Sp<Value>
+Value::decrypt(const crypto::PrivateKey& key)
+{
+    if (not decrypted) {
+        decrypted = true;
+        if (isEncrypted()) {
+            auto decryptedBlob = key.decrypt(cypher);
+            std::unique_ptr<Value> v {new Value(id)};
+            auto msg = msgpack::unpack((const char*)decryptedBlob.data(), decryptedBlob.size());
+            v->msgpack_unpack_body(msg.get());
+            if (v->recipient != key.getPublicKey().getId())
+                throw crypto::DecryptError("Recipient mismatch");
+            // Ignore values belonging to other people
+            if (not v->owner or not v->owner->checkSignature(v->getToSign(), v->signature))
+                throw crypto::DecryptError("Signature mismatch");
+            decryptedValue = std::move(v);
+        }
+    }
+    return decryptedValue;
+}
+
 uint64_t
 unpackId(const Json::Value& json, const std::string& key) {
     uint64_t ret = 0;

tests/dhtrunnertester.cpp

@@ -35,9 +35,11 @@ DhtRunnerTester::setUp() {
     dht::DhtRunner::Config config;
     config.dht_config.node_config.max_peer_req_per_sec = -1;
     config.dht_config.node_config.max_req_per_sec = -1;
+    config.dht_config.node_config.max_store_size = -1;
+    config.dht_config.node_config.max_store_keys = -1;
 
-    node1.run(42222, config);
-    node2.run(42232, config);
+    node1.run(0, config);
+    node2.run(0, config);
     node2.bootstrap(node1.getBound());
 }
 
@@ -54,15 +56,17 @@ DhtRunnerTester::tearDown() {
     node1.shutdown(shutdown);
     node2.shutdown(shutdown);
     std::unique_lock<std::mutex> lk(cv_m);
-    CPPUNIT_ASSERT(cv.wait_for(lk, 12s, [&]{ return done == 2; }));
+    CPPUNIT_ASSERT(cv.wait_for(lk, 30s, [&]{ return done == 2u; }));
     node1.join();
     node2.join();
 }
 
 void
 DhtRunnerTester::testConstructors() {
-    CPPUNIT_ASSERT(node1.getBoundPort() == 42222);
-    CPPUNIT_ASSERT(node2.getBoundPort() == 42232);
+    CPPUNIT_ASSERT(node1.getBoundPort());
+    CPPUNIT_ASSERT_EQUAL(node1.getBoundPort(), node1.getBound().getPort());
+    CPPUNIT_ASSERT(node2.getBoundPort());
+    CPPUNIT_ASSERT_EQUAL(node2.getBoundPort(), node2.getBound().getPort());
 
     dht::DhtRunner::Config config {};
     dht::DhtRunner::Context context {};
@@ -91,33 +95,53 @@ DhtRunnerTester::testListen() {
     std::mutex mutex;
     std::condition_variable cv;
     std::atomic_uint valueCount(0);
+    std::atomic_uint valueCounta(0);
+    std::atomic_uint valueCountb(0);
+    std::atomic_uint valueCountc(0);
+    std::atomic_uint valueCountd(0);
+
     unsigned putCount(0);
-    unsigned putOkCount(0);
+    unsigned putOkCount1(0);
+    unsigned putOkCount2(0);
+    unsigned putOkCount3(0);
 
     auto a = dht::InfoHash::get("234");
     auto b = dht::InfoHash::get("2345");
     auto c = dht::InfoHash::get("23456");
     auto d = dht::InfoHash::get("234567");
-    constexpr unsigned N = 256;
+    constexpr unsigned N = 2048;
     constexpr unsigned SZ = 56 * 1024;
 
-    auto ftokena = node1.listen(a, [&](const std::shared_ptr<dht::Value>&) {
-        valueCount++;
+    auto ftokena = node1.listen(a, [&](const std::vector<std::shared_ptr<dht::Value>>& values, bool expired) {
+        if (expired)
+            valueCounta -= values.size();
+        else
+            valueCounta += values.size();
         return true;
     });
 
     auto ftokenb = node1.listen(b, [&](const std::shared_ptr<dht::Value>&) {
-        valueCount++;
+        /*if (expired)
+            valueCountb -= values.size();
+        else
+            valueCountb += values.size();*/
+        valueCountb++;
         return false;
     });
 
-    auto ftokenc = node1.listen(c, [&](const std::shared_ptr<dht::Value>&) {
-        valueCount++;
+    auto ftokenc = node1.listen(c, [&](const std::vector<std::shared_ptr<dht::Value>>& values, bool expired) {
+        if (expired)
+            valueCountc -= values.size();
+        else
+            valueCountc += values.size();
         return true;
     });
 
-    auto ftokend = node1.listen(d, [&](const std::shared_ptr<dht::Value>&) {
-        valueCount++;
+    auto ftokend = node1.listen(d, [&](const std::vector<std::shared_ptr<dht::Value>>& values, bool expired) {
+        if (expired)
+            valueCountd -= values.size();
+        else
+            valueCountd += values.size();
         return true;
     });
 
@@ -130,21 +154,21 @@ DhtRunnerTester::testListen() {
         node2.put(a, dht::Value("v1"), [&](bool ok) {
             std::lock_guard<std::mutex> lock(mutex);
             putCount++;
-            if (ok) putOkCount++;
+            if (ok) putOkCount1++;
             cv.notify_all();
         });
         node2.put(b, dht::Value("v2"), [&](bool ok) {
             std::lock_guard<std::mutex> lock(mutex);
             putCount++;
-            if (ok) putOkCount++;
+            if (ok) putOkCount2++;
             cv.notify_all();
         });
         auto bigVal = std::make_shared<dht::Value>();
         bigVal->data = mtu;
-        node2.put(c, bigVal, [&](bool ok) {
+        node2.put(c, std::move(bigVal), [&](bool ok) {
             std::lock_guard<std::mutex> lock(mutex);
             putCount++;
-            if (ok) putOkCount++;
+            if (ok) putOkCount3++;
             cv.notify_all();
         });
     }
@@ -152,7 +176,9 @@ DhtRunnerTester::testListen() {
     {
         std::unique_lock<std::mutex> lk(mutex);
         CPPUNIT_ASSERT(cv.wait_for(lk, 30s, [&]{ return putCount == N * 3u; }));
-        CPPUNIT_ASSERT_EQUAL(N * 3u, putOkCount);
+        CPPUNIT_ASSERT_EQUAL(N, putOkCount1);
+        CPPUNIT_ASSERT_EQUAL(N, putOkCount2);
+        CPPUNIT_ASSERT_EQUAL(N, putOkCount3);
     }
 
     CPPUNIT_ASSERT(ftokena.valid());
@@ -168,7 +194,10 @@ DhtRunnerTester::testListen() {
     CPPUNIT_ASSERT(tokena);
     CPPUNIT_ASSERT(tokenc);
     CPPUNIT_ASSERT(tokend);
-    CPPUNIT_ASSERT_EQUAL(N * 2u + 1u, valueCount.load());
+    CPPUNIT_ASSERT_EQUAL(N, valueCounta.load());
+    CPPUNIT_ASSERT_EQUAL(1u, valueCountb.load());
+    CPPUNIT_ASSERT_EQUAL(N, valueCountc.load());
+    CPPUNIT_ASSERT_EQUAL(0u, valueCountd.load());
 
     node1.cancelListen(a, tokena);
     node1.cancelListen(b, std::move(ftokenb));

tools/CMakeLists.txt

@@ -1,7 +1,9 @@
 if (OPENDHT_SHARED)
     set (OPENDHT_LIBS opendht)
+    set (OPENDHT_C_LIBS opendht-c)
 else ()
     set (OPENDHT_LIBS opendht-static)
+    set (OPENDHT_C_LIBS opendht-c-static)
     if (MSVC)
         set (MSC_COMPAT_SOURCES ${MSC_COMPAT_DIR}/wingetopt.c)
     endif ()
@@ -9,8 +11,8 @@ endif ()
 
 function (configure_tool name extra_files)
     add_executable (${name} ${name}.cpp ${extra_files})
-    target_link_libraries (${name} LINK_PUBLIC ${READLINE_LIBRARIES})
-    target_link_libraries (${name} LINK_PUBLIC ${OPENDHT_LIBS})
+    add_dependencies(${name} ${OPENDHT_LIBS})
+    target_link_libraries (${name} LINK_PUBLIC ${OPENDHT_LIBS} ${READLINE_LIBRARIES})
     if (MSVC)
         target_sources(${name} PRIVATE ${MSC_COMPAT_SOURCES})
         target_include_directories (${name} PRIVATE ${MSC_COMPAT_DIR})
@@ -29,7 +31,8 @@ endif ()
 
 if (OPENDHT_C)
     add_executable (dhtcnode dhtcnode.c)
-    target_link_libraries (dhtcnode LINK_PUBLIC opendht-c ${READLINE_LIBRARIES})
+    add_dependencies(dhtcnode ${OPENDHT_C_LIBS})
+    target_link_libraries (dhtcnode LINK_PUBLIC ${OPENDHT_C_LIBS} ${READLINE_LIBRARIES})
     target_include_directories (dhtcnode SYSTEM PRIVATE ${PROJECT_SOURCE_DIR}/c)
 endif ()
 

tools/dhtcnode.c

@@ -1,9 +1,22 @@
-#include <opendht_c.h>
+/*
+ *  Copyright (C) 2014-2022 Savoir-faire Linux Inc.
+ *  Author : Adrien Béraud <adrien.beraud@savoirfairelinux.com>
+ *
+ *  This program is free software; you can redistribute it and/or modify
+ *  it under the terms of the GNU General Public License as published by
+ *  the Free Software Foundation; either version 3 of the License, or
+ *  (at your option) any later version.
+ *
+ *  This program is distributed in the hope that it will be useful,
+ *  but WITHOUT ANY WARRANTY; without even the implied warranty of
+ *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ *  GNU General Public License for more details.
+ *
+ *  You should have received a copy of the GNU General Public License
+ *  along with this program. If not, see <https://www.gnu.org/licenses/>.
+ */
 
-#include <getopt.h>
-#include <readline/readline.h>
-#include <readline/history.h>
-#include <arpa/inet.h>
+#include <opendht_c.h>
 
 #include <stdio.h>
 #include <string.h>
@@ -11,6 +24,11 @@
 #include <stdlib.h>
 #include <stdatomic.h>
 
+#include <getopt.h>
+#include <readline/readline.h>
+#include <readline/history.h>
+#include <arpa/inet.h>
+
 struct op_context {
     dht_runner* runner;
     atomic_bool stop;
@@ -18,12 +36,18 @@ struct op_context {
 struct listen_context {
     dht_runner* runner;
     dht_op_token* token;
+    size_t count;
 };
 
 bool dht_value_callback(const dht_value* value, bool expired, void* user_data)
 {
+    struct listen_context* ctx = (struct listen_context*) user_data;
+    if (expired)
+        ctx->count--;
+    else
+        ctx->count++;
     dht_data_view data = dht_value_get_data(value);
-    printf("Value callback %s: %.*s.\n", expired ? "expired" : "new", (int)data.size, data.data);
+    printf("Listen: %s value: %.*s (total %zu).\n", expired ? "expired" : "new", (int)data.size, data.data, ctx->count);
     return true;
 }
 
@@ -96,24 +120,26 @@ struct dht_params {
 };
 
 static const struct option long_options[] = {
-    {"help",                    no_argument      , NULL, 'h'},
-    {"port",                    required_argument, NULL, 'p'},
-    {"net",                     required_argument, NULL, 'n'},
-    {"bootstrap",               required_argument, NULL, 'b'},
-    {"identity",                no_argument      , NULL, 'i'},
-    {"verbose",                 no_argument      , NULL, 'v'},
-    {"service",                 no_argument      , NULL, 's'},
-    {"peer-discovery",          no_argument      , NULL, 'D'},
-    {"no-rate-limit",           no_argument      , NULL, 'U'},
-    {"persist",                 required_argument, NULL, 'f'},
-    {"logfile",                 required_argument, NULL, 'l'},
-    {"syslog",                  no_argument      , NULL, 'L'},
-    {NULL,                      0                , NULL,  0}
+    {"help",                no_argument      , NULL, 'h'},
+    {"port",                required_argument, NULL, 'p'},
+    {"net",                 required_argument, NULL, 'n'},
+    {"bootstrap",           required_argument, NULL, 'b'},
+    {"identity",            no_argument      , NULL, 'i'},
+    {"verbose",             no_argument      , NULL, 'v'},
+    {"service",             no_argument      , NULL, 's'},
+    {"peer-discovery",      no_argument      , NULL, 'D'},
+    {"no-rate-limit",       no_argument      , NULL, 'U'},
+    {"persist",             required_argument, NULL, 'f'},
+    {"logfile",             required_argument, NULL, 'l'},
+    {"syslog",              no_argument      , NULL, 'L'},
+    {"version",             no_argument      , NULL, 'V'},
+    {NULL,                  0                , NULL,  0}
 };
 
 struct dht_params
 parse_args(int argc, char **argv) {
     struct dht_params params;
+    memset(&params, 0, sizeof params);
     int opt;
     while ((opt = getopt_long(argc, argv, "hisvDp:n:b:f:l:", long_options, NULL)) != -1) {
         switch (opt) {
@@ -144,6 +170,9 @@ parse_args(int argc, char **argv) {
         case 's':
             params.service = true;
             break;
+        case 'V':
+            params.version = true;
+            break;
         default:
             break;
         }
@@ -151,15 +180,30 @@ parse_args(int argc, char **argv) {
     return params;
 }
 
+dht_infohash parse_key(const char* key_str) {
+    dht_infohash key;
+    dht_infohash_from_hex_null(&key, key_str);
+    if (dht_infohash_is_zero(&key)) {
+        dht_infohash_get_from_string(&key, key_str);
+        printf("Using h(%s) = %s\n", key_str, dht_infohash_print(&key));
+    }
+    return key;
+}
+
 int main(int argc, char **argv)
 {
     struct dht_params params = parse_args(argc, argv);
 
+    if (params.version) {
+        printf("OpenDHT version %s\n", dht_version());
+        return EXIT_SUCCESS;
+    }
+
     dht_runner* runner = dht_runner_new();
     dht_runner_config dht_config;
-	dht_runner_config_default(&dht_config);
-	dht_config.peer_discovery = params.peer_discovery; // Look for other peers on the network
-	dht_config.peer_publish = params.peer_discovery; // Publish our own peer info
+    dht_runner_config_default(&dht_config);
+    dht_config.peer_discovery = params.peer_discovery; // Look for other peers on the network
+    dht_config.peer_publish = params.peer_discovery; // Publish our own peer info
     dht_config.dht_config.node_config.network = params.network;
     dht_config.log = params.log;
     dht_runner_run_config(runner, params.port, &dht_config);
@@ -172,20 +216,19 @@ int main(int argc, char **argv)
     char cmd[64];
     char arg[64];
     char value[256];
+    dht_infohash key;
     while (true) {
         const char* line_read = readline("> ");
-        if (line_read && *line_read)
-            add_history(line_read);
         if (!line_read)
             break;
-        if (!strcmp(line_read, "\0"))
+        if (!*line_read)
             continue;
+        add_history(line_read);
 
-        bzero(cmd, sizeof cmd);
-        bzero(arg, sizeof arg);
-        bzero(value, sizeof value);
+        memset(cmd, 0, sizeof cmd);
+        memset(arg, 0, sizeof arg);
+        memset(value, 0, sizeof value);
         sscanf(line_read, "%64s %64s %256s", cmd, arg, value);
-        //printf("%s -> %s\n", cmd, arg);
 
         if (!strcmp(cmd, "la")) {
             struct sockaddr** addrs = dht_runner_get_public_address(runner);
@@ -201,34 +244,44 @@ int main(int argc, char **argv)
             }
             continue;
         }
-
-        dht_infohash key;
-        dht_infohash_from_hex(&key, arg);
-        if (dht_infohash_is_zero(&key)) {
-            dht_infohash_get_from_string(&key, arg);
-            printf("Using h(%s) = %s\n", arg, dht_infohash_print(&key));
+        else if (!strcmp(cmd, "ll")) {
+            dht_infohash key = dht_runner_get_node_id(runner);
+            printf("DHT node %s running on port %u\n", dht_infohash_print(&key), dht_runner_get_bound_port(runner, AF_INET));
+            continue;
         }
-        if (!strcmp(cmd, "g")) {
+        else if (!strcmp(cmd, "g")) {
+            key = parse_key(arg);
             dht_runner_get(runner, &key, dht_get_callback, dht_get_done_callback, runner);
-        } else if (!strcmp(cmd, "l")) {
+        }
+        else if (!strcmp(cmd, "l")) {
+            key = parse_key(arg);
             struct listen_context* ctx = malloc(sizeof(struct listen_context));
             ctx->runner = runner;
+            ctx->count = 0;
             ctx->token = dht_runner_listen(runner, &key, dht_value_callback, listen_context_free, ctx);
-        } else if (!strcmp(cmd, "p")) {
+        }
+        else if (!strcmp(cmd, "p")) {
+            key = parse_key(arg);
             dht_value* val = dht_value_new_from_string(value);
             dht_runner_put(runner, &key, val, dht_put_done_callback, runner, true);
             dht_value_unref(val);
         }
+        else {
+            printf("Unkown command: %s\n", cmd);
+        }
     }
-    printf("Stopping..\n");
 
+    // Graceful shutdown
+    printf("Stopping…\n");
     struct op_context ctx;
     ctx.runner = runner;
     atomic_init(&ctx.stop, false);
     dht_runner_shutdown(runner, dht_shutdown_callback, &ctx);
+
+    // Wait until shutdown callback is called
     while (!atomic_load(&ctx.stop)) {
-        usleep(250);
+        usleep(10000);
     }
     dht_runner_delete(runner);
-    return 0;
+    return EXIT_SUCCESS;
 }

tools/dhtnode.cpp

@@ -146,6 +146,7 @@ void cmd_loop(std::shared_ptr<DhtRunner>& node, dht_params& params
             node->getNodeInfo([&](const std::shared_ptr<dht::NodeInfo>& nodeInfo) {
                 print_node_info(*nodeInfo);
                 std::cout << nodeInfo->ongoing_ops << " ongoing operations" << std::endl;
+                std::cout << "Storage has " << nodeInfo->storage_values <<  " values, using " << (nodeInfo->storage_size/1024) << " KB" << std::endl;
                 std::cout << "IPv4 stats:" << std::endl;
                 std::cout << nodeInfo->ipv4.toString() << std::endl;
                 std::cout << "IPv6 stats:" << std::endl;