update version to 2.4.3

CMakeLists.txt

@@ -19,7 +19,7 @@ include(FindPkgConfig)
 include(cmake/CheckAtomic.cmake)
 
 set (opendht_VERSION_MAJOR 2)
-set (opendht_VERSION_MINOR 4.2)
+set (opendht_VERSION_MINOR 4.3)
 set (opendht_VERSION ${opendht_VERSION_MAJOR}.${opendht_VERSION_MINOR})
 set (PACKAGE_VERSION ${opendht_VERSION})
 set (VERSION "${opendht_VERSION}")

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();
+
 // 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], 2)
+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};

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/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();
     }

src/dht.cpp

@@ -1793,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),

src/dhtrunner.cpp

@@ -572,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();

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));
       }

tests/dhtrunnertester.cpp

@@ -35,6 +35,8 @@ 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(0, config);
     node2.run(0, config);
@@ -105,23 +107,35 @@ DhtRunnerTester::testListen() {
     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)
+            valueCount -= values.size();
+        else
+            valueCount += values.size();
         return true;
     });
 
-    auto ftokenb = node1.listen(b, [&](const std::shared_ptr<dht::Value>&) {
-        valueCount++;
+    auto ftokenb = node1.listen(b, [&](const std::vector<std::shared_ptr<dht::Value>>& values, bool expired) {
+        if (expired)
+            valueCount -= values.size();
+        else
+            valueCount += values.size();
         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)
+            valueCount -= values.size();
+        else
+            valueCount += 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)
+            valueCount -= values.size();
+        else
+            valueCount += values.size();
         return true;
     });
 
@@ -145,7 +159,7 @@ DhtRunnerTester::testListen() {
         });
         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) putOkCount3++;

tools/dhtcnode.c

@@ -18,17 +18,17 @@
 
 #include <opendht_c.h>
 
-#include <getopt.h>
-#include <readline/readline.h>
-#include <readline/history.h>
-#include <arpa/inet.h>
-
 #include <stdio.h>
 #include <string.h>
 #include <unistd.h>
 #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;
@@ -36,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;
 }
 
@@ -114,25 +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;
-    bzero(&params, sizeof 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) {
@@ -163,6 +170,9 @@ parse_args(int argc, char **argv) {
         case 's':
             params.service = true;
             break;
+        case 'V':
+            params.version = true;
+            break;
         default:
             break;
         }
@@ -170,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);
@@ -191,18 +216,18 @@ 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);
 
         if (!strcmp(cmd, "la")) {
@@ -219,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(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;