ci/github: add experimental docker build action

Current operator== creates a very weird warning about using an
uninitialized value (no warning without the reinterpret_cast).

.github/workflows/ccpp.yml

@@ -29,6 +29,24 @@ jobs:
     - name: make check
       run: make check
 
+  build-ubuntu-minimal:
+    name: Ubuntu/GCC minimal build
+    runs-on: ubuntu-latest
+    steps:
+    - uses: actions/checkout@v3
+    - name: deps
+      run: |
+        sudo apt install libncurses5-dev libreadline-dev nettle-dev \
+        libgnutls28-dev libcppunit-dev libmsgpack-dev libargon2-0-dev
+    - name: cmake
+      run: |
+        mkdir build && cd build
+        cmake .. -DCMAKE_INSTALL_PREFIX=/usr/local -DCMAKE_BUILD_TYPE=Debug \
+                 -DOPENDHT_C=Off -DOPENDHT_TESTS=On -DOPENDHT_PEER_DISCOVERY=Off -DOPENDHT_PYTHON=Off \
+                 -DOPENDHT_TOOLS=On -DOPENDHT_PROXY_SERVER=Off -DOPENDHT_PROXY_CLIENT=Off
+    - name: make
+      run: cd build && make
+
   build-macos:
     name: macOS/Clang build
     runs-on: macos-11
@@ -37,7 +55,7 @@ jobs:
     - name: deps
       run: |
         brew install msgpack-cxx asio gnutls nettle readline fmt jsoncpp argon2 openssl http-parser cppunit
-    
+
     - name: restinio
       run: |
         mkdir restinio && cd restinio

.github/workflows/release-package.yml

@@ -0,0 +1,64 @@
+# This workflow uses actions that are not certified by GitHub.
+# They are provided by a third-party and are governed by
+# separate terms of service, privacy policy, and support
+# documentation.
+
+name: Create and publish Docker images
+
+on:
+  push
+  #release:
+  #  types: [published]
+
+env:
+  REGISTRY: ghcr.io
+  IMAGE_NAME_DEPS: ${{ github.repository }}/opendht
+  IMAGE_NAME: ${{ github.repository }}/opendht-deps
+  IMAGE_NAME_DEPS_LLVM: ${{ github.repository }}/opendht-llvm
+  IMAGE_NAME_LLVM: ${{ github.repository }}/opendht-deps-llvm
+
+jobs:
+  build-and-push-image:
+    runs-on: ubuntu-latest
+    permissions:
+      contents: read
+      packages: write
+
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v3
+
+      - name: Log in to the Container registry
+        uses: docker/login-action@v2
+        with:
+          registry: ${{ env.REGISTRY }}
+          username: ${{ github.actor }}
+          password: ${{ secrets.GITHUB_TOKEN }}
+
+      - name: Extract metadata (tags, labels) for Docker
+        id: meta
+        uses: docker/metadata-action@v4
+        with:
+          images: |
+            ${{ env.REGISTRY }}/${{ env.IMAGE_NAME_DEPS }}
+            ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}
+            ${{ env.REGISTRY }}/${{ env.IMAGE_NAME_DEPS_LLVM }}
+            ${{ env.REGISTRY }}/${{ env.IMAGE_NAME_LLVM }}
+
+      - name: Build and push Docker image for dependencies
+        uses: docker/build-push-action@v3
+        with:
+          context: .
+          file: docker/DockerfileDeps
+          #push: true
+          tags: ${{ steps.meta.outputs.tags }}
+          labels: ${{ steps.meta.outputs.labels }}
+
+      - name: Build and push Docker image
+        uses: docker/build-push-action@v3
+        with:
+          context: .
+          file: docker/Dockerfile
+          #push: true
+          tags: ${{ steps.meta.outputs.tags }}
+          labels: ${{ steps.meta.outputs.labels }}

.gitignore

@@ -76,4 +76,5 @@ doc/Doxyfile
 
 # build dir
 build
+build_dev
 .DS_Store

.vscode/launch.json

@@ -5,9 +5,9 @@
             "name": "dhtnode",
             "type": "cppdbg",
             "request": "launch",
-            "program": "${workspaceFolder}/build/tools/dhtnode",
+            "program": "${workspaceFolder}/build_dev/tools/dhtnode",
             "args": ["-v", "-p", "4222"],
-            "cwd": "${workspaceFolder}/build",
+            "cwd": "${workspaceFolder}/build_dev",
             "environment": [],
             "setupCommands": [
                 {
@@ -30,8 +30,8 @@
             "name": "tests",
             "type": "cppdbg",
             "request": "launch",
-            "program": "${workspaceFolder}/build/opendht_unit_tests",
-            "cwd": "${workspaceFolder}/build",
+            "program": "${workspaceFolder}/build_dev/opendht_unit_tests",
+            "cwd": "${workspaceFolder}/build_dev",
             "environment": [],
             "setupCommands": [
                 {
@@ -41,14 +41,14 @@
                 }
             ],
             "preLaunchTask": "build",
-                        "linux": {
+            "linux": {
                 "MIMode": "gdb",
                 "externalConsole": false,
-              },
-              "osx": {
+            },
+            "osx": {
                 "MIMode": "lldb",
                 "externalConsole": true,
-              },
+            },
         }
     ]
 }

.vscode/tasks.json

@@ -5,7 +5,7 @@
             "label": "build",
             "type": "shell",
             "options": {
-                "cwd": "${workspaceRoot}/build"
+                "cwd": "${workspaceRoot}/build_dev"
             },
             "command": "make",
             "args": ["-j4"],
@@ -15,11 +15,12 @@
             "label": "cmake",
             "type": "shell",
             "options": {
-                "cwd": "${workspaceRoot}/build"
+                "cwd": "${workspaceRoot}/build_dev"
             },
             "command": "cmake",
             "args": [
                 "${workspaceRoot}",
+                "-DCMAKE_EXPORT_COMPILE_COMMANDS=On",
                 "-DCMAKE_BUILD_TYPE=Debug",
                 "-DOPENDHT_PROXY_CLIENT=On",
                 "-DOPENDHT_PROXY_SERVER=On",
@@ -37,7 +38,7 @@
             "command": "mkdir",
             "args": [
                 "-p",
-                "${workspaceRoot}/build"
+                "${workspaceRoot}/build_dev"
             ]
         },
     ]

CMakeLists.txt

@@ -1,10 +1,4 @@
-cmake_minimum_required (VERSION 3.10)
-if(POLICY CMP0065)
-  cmake_policy(SET CMP0065 NEW)
-endif()
-if(POLICY CMP0069)
-  cmake_policy(SET CMP0069 NEW)
-endif()
+cmake_minimum_required (VERSION 3.10..3.20)
 if(POLICY CMP0073)
   cmake_policy(SET CMP0073 NEW)
 endif()
@@ -15,11 +9,12 @@ project (opendht)
 
 include(CMakePackageConfigHelpers)
 include(CMakeDependentOption)
+include(CheckIncludeFileCXX)
 include(FindPkgConfig)
 include(cmake/CheckAtomic.cmake)
 
 set (opendht_VERSION_MAJOR 2)
-set (opendht_VERSION_MINOR 4.0)
+set (opendht_VERSION_MINOR 4.7)
 set (opendht_VERSION ${opendht_VERSION_MAJOR}.${opendht_VERSION_MINOR})
 set (PACKAGE_VERSION ${opendht_VERSION})
 set (VERSION "${opendht_VERSION}")
@@ -49,6 +44,10 @@ find_package(Doxygen)
 option (OPENDHT_DOCUMENTATION "Create and install the HTML based API documentation (requires Doxygen)" ${DOXYGEN_FOUND})
 
 # Dependencies
+if (NOT HAVE_CXX_ATOMICS_WITHOUT_LIB)
+    link_libraries (atomic)
+endif ()
+
 list (APPEND CMAKE_MODULE_PATH "${PROJECT_SOURCE_DIR}/cmake")
 if (NOT MSVC)
     set (THREADS_PREFER_PTHREAD_FLAG TRUE)
@@ -56,11 +55,15 @@ if (NOT MSVC)
     find_package (PkgConfig REQUIRED)
     find_package (GnuTLS 3.3 REQUIRED)
     pkg_search_module (Nettle REQUIRED nettle)
-    find_package (msgpack REQUIRED)
+    check_include_file_cxx(msgpack.hpp HAVE_MSGPACKCXX)
+    if (NOT HAVE_MSGPACKCXX)
+        find_package (msgpack REQUIRED)
+    endif()
     if (OPENDHT_TOOLS)
         find_package (Readline 6 REQUIRED)
     endif ()
     pkg_search_module(argon2 REQUIRED libargon2)
+    set(argon2_lib ", libargon2")
     pkg_search_module(Jsoncpp jsoncpp)
     if (Jsoncpp_FOUND)
         add_definitions(-DOPENDHT_JSONCPP)
@@ -80,9 +83,11 @@ if (NOT MSVC)
         add_library(fmt SHARED IMPORTED)
         find_library(HTTP_PARSER_LIBRARY http_parser)
         add_library(http_parser SHARED IMPORTED)
+        set(http_parser_lib "-lhttp_parser ")
         if (NOT Jsoncpp_FOUND)
             message(SEND_ERROR "Jsoncpp is required for DHT proxy support")
         endif()
+        set(jsoncpp_lib ", jsoncpp")
         if (OPENDHT_PROXY_OPENSSL)
             # https://cmake.org/cmake/help/latest/module/FindOpenSSL.html
             pkg_search_module(OPENSSL REQUIRED openssl)
@@ -93,6 +98,7 @@ if (NOT MSVC)
             else ()
                 message(SEND_ERROR "OpenSSL is required for DHT proxy as specified")
             endif()
+            set(openssl_lib ", openssl")
         endif()
     else ()
         set(OPENDHT_PROXY_OPENSSL OFF)
@@ -399,15 +405,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
@@ -416,7 +430,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 ()

Makefile.am

@@ -4,6 +4,10 @@ SUBDIRS =
 
 SUBDIRS += src
 
+if ENABLE_C
+SUBDIRS += c
+endif
+
 if ENABLE_TOOLS
 SUBDIRS += tools
 endif

c/Makefile.am

@@ -0,0 +1,15 @@
+lib_LTLIBRARIES = libopendht-c.la
+noinst_HEADERS = opendht_c.h
+
+AM_CPPFLAGS = -DOPENDHT_C_BUILD -isystem @top_srcdir@/include @JsonCpp_CFLAGS@ @MsgPack_CFLAGS@
+if OPENDHT_SHARED
+AM_CPPFLAGS += -Dopendht_c_EXPORTS
+endif
+
+libopendht_c_la_LIBADD = $(LIBOBJS) ../src/.libs/libopendht.la
+
+libopendht_c_la_SOURCES  = \
+        opendht.cpp
+
+nobase_include_HEADERS = \
+        opendht_c.h

c/opendht.cpp

@@ -17,7 +17,9 @@
  */
 
 #include "opendht_c.h"
-#include "opendht.h"
+
+#include <opendht.h>
+#include <opendht/log.h>
 
 using ValueSp = std::shared_ptr<dht::Value>;
 using PrivkeySp = std::shared_ptr<dht::crypto::PrivateKey>;
@@ -30,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) {
@@ -69,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) {
@@ -333,7 +344,12 @@ int dht_runner_run_config(dht_runner* r, in_port_t port, const dht_runner_config
         config.push_token = conf->push_token ? std::string(conf->push_token) : std::string{};
         config.peer_discovery = conf->peer_discovery;
         config.peer_publish = conf->peer_publish;
-        runner->run(port, config);
+
+        dht::DhtRunner::Context context;
+        if (conf->log) {
+            context.logger = dht::log::getStdLogger();
+        }
+        runner->run(port, config, std::move(context));
     } catch(...) {
         return ENOTCONN;
     }
@@ -488,14 +504,8 @@ struct sockaddr** dht_runner_get_public_address(const dht_runner* r) {
     if (addrs.empty())
         return nullptr;
     auto ret = (struct sockaddr**)malloc(sizeof(struct sockaddr*) * (addrs.size() + 1));
-    for (size_t i=0; i<addrs.size(); i++) {
-        if (const auto& addr = addrs[i]) {
-            ret[i] = (struct sockaddr*)malloc(addr.getLength());
-            memcpy((struct sockaddr*)ret[i], addr.get(), addr.getLength());
-        } else {
-            ret[i] = nullptr;
-        }
-    }
+    for (size_t i=0; i<addrs.size(); i++)
+        ret[i] = addrs[i].release();
     ret[addrs.size()] = nullptr;
     return ret;
 }

c/opendht_c.h

@@ -22,7 +22,7 @@
 extern "C" {
 #endif
 
-#include "def.h"
+#include <opendht/def.h>
 #include <sys/socket.h>
 #include <netinet/in.h>
 #include <stdbool.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);
@@ -147,6 +150,7 @@ struct OPENDHT_PUBLIC dht_runner_config {
     bool peer_publish;
     dht_certificate* server_ca;
     dht_identity client_identity;
+    bool log;
 };
 typedef struct dht_runner_config dht_runner_config;
 OPENDHT_C_PUBLIC void dht_runner_config_default(dht_runner_config* config);

cmake/CheckAtomic.cmake

@@ -2,6 +2,7 @@
 
 INCLUDE(CheckCXXSourceCompiles)
 INCLUDE(CheckLibraryExists)
+INCLUDE("${CMAKE_CURRENT_LIST_DIR}/DetermineGCCCompatible.cmake")
 
 # Sometimes linking against libatomic is required for atomic ops, if
 # the platform doesn't support lock-free atomics.
@@ -32,6 +33,7 @@ function(check_working_cxx_atomics64 varname)
 std::atomic<uint64_t> x (0);
 int main() {
   uint64_t i = x.load(std::memory_order_relaxed);
+  (void)i;
   return 0;
 }
 " ${varname})
@@ -39,9 +41,10 @@ int main() {
 endfunction(check_working_cxx_atomics64)
 
 
-# This isn't necessary on MSVC, so avoid command-line switch annoyance
-# by only running on GCC-like hosts.
-if (LLVM_COMPILER_IS_GCC_COMPATIBLE)
+# Check for (non-64-bit) atomic operations.
+if(MSVC)
+  set(HAVE_CXX_ATOMICS_WITHOUT_LIB True)
+elseif(LLVM_COMPILER_IS_GCC_COMPATIBLE OR CMAKE_CXX_COMPILER_ID MATCHES "XL")
   # First check if atomics work without the library.
   check_working_cxx_atomics(HAVE_CXX_ATOMICS_WITHOUT_LIB)
   # If not, check if the library exists, and atomics work with it.
@@ -57,16 +60,16 @@ endif()
 # Check for 64 bit atomic operations.
 if(MSVC)
   set(HAVE_CXX_ATOMICS64_WITHOUT_LIB True)
-else()
+elseif(LLVM_COMPILER_IS_GCC_COMPATIBLE OR CMAKE_CXX_COMPILER_ID MATCHES "XL")
+  # First check if atomics work without the library.
   check_working_cxx_atomics64(HAVE_CXX_ATOMICS64_WITHOUT_LIB)
-endif()
-
-# If not, check if the library exists, and atomics work with it.
-if(NOT HAVE_CXX_ATOMICS64_WITHOUT_LIB)
-  list(APPEND CMAKE_REQUIRED_LIBRARIES "atomic")
-  check_working_cxx_atomics64(HAVE_CXX_ATOMICS64_WITH_LIB)
-  if (NOT HAVE_CXX_ATOMICS64_WITH_LIB)
-    message(FATAL_ERROR "Host compiler must support 64-bit std::atomic!")
+  # If not, check if the library exists, and atomics work with it.
+  if(NOT HAVE_CXX_ATOMICS64_WITHOUT_LIB)
+    list(APPEND CMAKE_REQUIRED_LIBRARIES "atomic")
+    check_working_cxx_atomics64(HAVE_CXX_ATOMICS64_WITH_LIB)
+    if (NOT HAVE_CXX_ATOMICS64_WITH_LIB)
+      message(FATAL_ERROR "Host compiler must support 64-bit std::atomic!")
+    endif()
   endif()
 endif()
 

cmake/DetermineGCCCompatible.cmake

@@ -0,0 +1,13 @@
+# Determine if the compiler has GCC-compatible command-line syntax.
+
+if(NOT DEFINED LLVM_COMPILER_IS_GCC_COMPATIBLE)
+  if(CMAKE_COMPILER_IS_GNUCXX)
+    set(LLVM_COMPILER_IS_GCC_COMPATIBLE ON)
+  elseif( MSVC )
+    set(LLVM_COMPILER_IS_GCC_COMPATIBLE OFF)
+  elseif( "${CMAKE_CXX_COMPILER_ID}" MATCHES "Clang" )
+    set(LLVM_COMPILER_IS_GCC_COMPATIBLE ON)
+  elseif( "${CMAKE_CXX_COMPILER_ID}" MATCHES "Intel" )
+    set(LLVM_COMPILER_IS_GCC_COMPATIBLE ON)
+  endif()
+endif()

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], 7)
 m4_define([opendht_version],
 		  [opendht_major_version.opendht_minor_version.opendht_patch_version])
 
@@ -19,17 +19,32 @@ AC_SUBST(OPENDHT_PATCH_VERSION, opendht_patch_version)
 AC_ARG_ENABLE([debug], AS_HELP_STRING([--enable-debug], [Build in debug mode, adds stricter warnings, disables optimization]))
 AS_IF([test "x$enable_debug" = "xyes"],
       [CXXFLAGS="${CXXFLAGS} -g -Wno-return-type -Wall -Wextra -Wnon-virtual-dtor -O0 -pedantic-errors"],
-      [CXXFLAGS="${CXXFLAGS} -O3 -pedantic-errors"])
+      [CXXFLAGS="${CXXFLAGS} -O3 -pedantic-errors -fvisibility=hidden"])
+
+CPPFLAGS+=" -DOPENDHT_BUILD"
+AM_CONDITIONAL([OPENDHT_SHARED], [test "x$enable_shared" != xno])
+AM_COND_IF(OPENDHT_SHARED, [
+  CPPFLAGS+=" -Dopendht_EXPORTS"
+])
 
-AC_PROG_CXX
 AM_PROG_AR
+LT_INIT()
+LT_LANG(C++)
+AC_LANG(C++)
+AC_PROG_CXX
+AX_CXX_COMPILE_STDCXX(17,[noext],[mandatory])
 
 dnl Check for logs
 AC_ARG_ENABLE([logs], [AS_HELP_STRING([--disable-logs], [Disable DHT logs])])
-AS_IF([test "x$enable_logs" != "xno"], [
-	AC_DEFINE([OPENDHT_LOG], [true], [Define if DHT logs are enabled])
-], [
-	AC_DEFINE([OPENDHT_LOG], [false], [Define if DHT logs are enabled])
+AS_IF([test "x$enable_logs" != "xno"], [], [
+	AC_DEFINE([OPENDHT_LOG], [false], [Define if DHT logs are disabled])
+])
+
+dnl Check for C binding
+AC_ARG_ENABLE([c], [AS_HELP_STRING([--disable-c], [Disable DHT C binding])])
+AM_CONDITIONAL(ENABLE_C, test x$enable_c != "xno")
+AM_COND_IF(ENABLE_C, [
+	AC_DEFINE([OPENDHT_C], [], [Define if DHT C biding is enabled])
 ])
 
 dnl Check for indexation
@@ -99,11 +114,6 @@ AM_CONDITIONAL(WIN32, [test "x$SYS" = "xmingw32"])
 AS_IF([test "x$SYS" = "xandroid"],
       [], [LDFLAGS="${LDFLAGS} -lpthread"])
 
-LT_INIT()
-LT_LANG(C++)
-
-AX_CXX_COMPILE_STDCXX(17,[noext],[mandatory])
-
 PKG_PROG_PKG_CONFIG()
 
 AC_ARG_ENABLE([proxy_server], AS_HELP_STRING([--enable-proxy-server], [Enable proxy server ability]), proxy_server=yes, proxy_server=no)
@@ -122,8 +132,8 @@ AC_ARG_ENABLE([proxy_client], AS_HELP_STRING([--enable-proxy-client], [Enable pr
 AM_CONDITIONAL(ENABLE_PROXY_CLIENT, test x$proxy_client == xyes)
 
 AC_ARG_ENABLE([tests], AS_HELP_STRING([--enable-tests], [Enable tests]), build_tests=yes, build_tests=no)
-AM_CONDITIONAL(ENABLE_TESTS, test x$build_tests == xyes)
-AM_COND_IF([ENABLE_TESTS], [
+AM_CONDITIONAL(ENABLE_TESTS, test "x$build_tests" == xyes)
+AS_IF([test "x$build_tests" == xyes], [
 	PKG_CHECK_MODULES([CppUnit], [cppunit >= 1.12])
 ])
 
@@ -131,7 +141,9 @@ AM_CONDITIONAL(PROXY_CLIENT_OR_SERVER, test x$proxy_client == xyes || test x$pro
 
 PKG_CHECK_MODULES([Nettle], [nettle >= 2.4])
 PKG_CHECK_MODULES([GnuTLS], [gnutls >= 3.3])
-PKG_CHECK_MODULES([MsgPack], [msgpack >= 1.2])
+AC_CHECK_HEADERS([msgpack.hpp], [], [
+  PKG_CHECK_MODULES([MsgPack], [msgpack >= 1.2])
+])
 PKG_CHECK_MODULES([Argon2], [libargon2])
 AC_SUBST(argon2_lib, [", libargon2"])
 
@@ -160,7 +172,7 @@ AS_IF([test "x$have_openssl" = "xyes"], [
 ])
 
 AM_COND_IF([PROXY_CLIENT_OR_SERVER], [
-    AC_CHECK_HEADERS([asio.hpp], exit,, AC_MSG_ERROR([Missing Asio headers files]))
+    AC_CHECK_HEADERS([asio.hpp],, AC_MSG_ERROR([Missing Asio headers files]))
     CXXFLAGS="${CXXFLAGS} -DASIO_STANDALONE"
     PKG_CHECK_MODULES([Fmt], [fmt >= 5.3.0], [have_fmt=yes], [have_fmt=no])
     AS_IF([test "x$have_fmt" = "xyes"], [
@@ -198,6 +210,7 @@ AC_SUBST(PROJECT_BINARY_DIR, "../src/.libs")
 
 AC_CONFIG_FILES([Makefile
                  src/Makefile
+                 c/Makefile
                  tools/Makefile
                  python/Makefile
                  python/setup.py

docker/DockerfileDeps

@@ -9,7 +9,7 @@ RUN apt-get update && apt-get install -y \
 
 RUN apt-get update && apt-get install -y \
         build-essential pkg-config cmake git wget \
-        autotools-dev autoconf \
+        libtool autotools-dev autoconf \
         cython3 python3-dev python3-setuptools \
         libncurses5-dev libreadline-dev nettle-dev libcppunit-dev \
         libgnutls28-dev libuv1-dev libjsoncpp-dev libargon2-dev \

docker/DockerfileDepsLlvm

@@ -8,7 +8,8 @@ RUN apt-get update && apt-get install -y \
     && echo 'debconf debconf/frontend select Noninteractive' | debconf-set-selections
 
 RUN apt-get update \
-    && apt-get install -y llvm llvm-dev lldb clang gdb make cmake pkg-config git wget libncurses5-dev libreadline-dev \
+    && apt-get install -y llvm llvm-dev lldb clang gdb make cmake pkg-config \
+       libtool git wget libncurses5-dev libreadline-dev \
        nettle-dev libgnutls28-dev libuv1-dev libmsgpack-dev libjsoncpp-dev cython3 python3-dev \
        python3-setuptools libcppunit-dev python3-pip \
        autotools-dev autoconf libssl-dev libargon2-dev \

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/crypto.h

@@ -32,6 +32,8 @@ extern "C" {
 
 #include <vector>
 #include <memory>
+#include <atomic>
+#include <mutex>
 
 #ifdef _WIN32
 #include <iso646.h>
@@ -100,12 +102,12 @@ struct OPENDHT_PUBLIC PublicKey
     /**
      * Get public key fingerprint
      */
-    InfoHash getId() const;
+   const InfoHash& getId() const;
 
     /**
      * Get public key long fingerprint
      */
-    PkId getLongId() const;
+    const PkId& getLongId() const;
 
     bool checkSignature(const uint8_t* data, size_t data_len, const uint8_t* signature, size_t signature_len) const;
     inline bool checkSignature(const Blob& data, const Blob& signature) const {
@@ -138,6 +140,11 @@ struct OPENDHT_PUBLIC PublicKey
 
     gnutls_pubkey_t pk {nullptr};
 private:
+    mutable InfoHash cachedId_ {};
+    mutable PkId cachedLongId_ {};
+    mutable std::atomic_bool idCached_ {false};
+    mutable std::atomic_bool longIdCached_ {false};
+
     PublicKey(const PublicKey&) = delete;
     PublicKey& operator=(const PublicKey&) = delete;
     void encryptBloc(const uint8_t* src, size_t src_size, uint8_t* dst, size_t dst_size) const;
@@ -201,6 +208,7 @@ private:
     PrivateKey& operator=(const PrivateKey&) = delete;
     Blob decryptBloc(const uint8_t* src, size_t src_size) const;
 
+    mutable std::mutex publicKeyMutex_ {};
     mutable std::shared_ptr<PublicKey> publicKey_ {};
 };
 
@@ -486,9 +494,9 @@ struct OPENDHT_PUBLIC Certificate {
     PublicKey getPublicKey() const;
 
     /** Same as getPublicKey().getId() */
-    InfoHash getId() const;
+    const InfoHash& getId() const;
     /** Same as getPublicKey().getLongId() */
-    PkId getLongId() const;
+    const PkId& getLongId() const;
 
     Blob getSerialNumber() const;
 
@@ -606,6 +614,8 @@ private:
     Certificate& operator=(const Certificate&) = delete;
     mutable InfoHash cachedId_ {};
     mutable PkId cachedLongId_ {};
+    mutable std::atomic_bool idCached_ {false};
+    mutable std::atomic_bool longIdCached_ {false};
 
     struct crlNumberCmp {
         bool operator() (const std::shared_ptr<RevocationList>& lhs, const std::shared_ptr<RevocationList>& rhs) const {

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 override { return 0; }
     void connectivityChanged(sa_family_t) override {
         getProxyInfos();
     }

include/opendht/dhtrunner.h

@@ -273,13 +273,13 @@ public:
      * Usefull to restart a node and get things running fast without putting load on the network.
      */
     void bootstrap(std::vector<SockAddr> nodes, DoneCallbackSimple cb={});
-    void bootstrap(const SockAddr& addr, DoneCallbackSimple cb={});
+    void bootstrap(SockAddr addr, DoneCallbackSimple cb={});
 
     /**
      * Insert known nodes to the routing table, without necessarly ping them.
      * Usefull to restart a node and get things running fast without putting load on the network.
      */
-    void bootstrap(const std::vector<NodeExport>& nodes);
+    void bootstrap(std::vector<NodeExport> nodes);
 
     /**
      * Add host:service to bootstrap nodes, and ping this node.
@@ -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;
@@ -371,8 +372,8 @@ public:
     std::string getRoutingTablesLog(sa_family_t af) const;
     std::string getSearchesLog(sa_family_t af = AF_UNSPEC) const;
     std::string getSearchLog(const InfoHash&, sa_family_t af = AF_UNSPEC) const;
-    std::vector<SockAddr> getPublicAddress(sa_family_t af = AF_UNSPEC);
-    std::vector<std::string> getPublicAddressStr(sa_family_t af = AF_UNSPEC);
+    std::vector<SockAddr> getPublicAddress(sa_family_t af = AF_UNSPEC) const;
+    std::vector<std::string> getPublicAddressStr(sa_family_t af = AF_UNSPEC) const;
     void getPublicAddress(std::function<void(std::vector<SockAddr>&&)>, sa_family_t af = AF_UNSPEC);
 
     // securedht methods
@@ -436,7 +437,6 @@ public:
     /**
      * Start or stop the proxy
      * @param proxify if we want to use the proxy
-     * @param deviceKey non empty to enable push notifications
      */
     void enableProxy(bool proxify);
 
@@ -473,12 +473,9 @@ private:
     DoneCallback bindOpDoneCallback(DoneCallback&& cb);
     DoneCallbackSimple bindOpDoneCallback(DoneCallbackSimple&& cb);
 
-    /** Local DHT instance */
+    /** DHT instance */
     std::unique_ptr<SecureDht> dht_;
 
-    /** Proxy client instance */
-    std::unique_ptr<SecureDht> dht_via_proxy_;
-
     /** true if we are currently using a proxy */
     std::atomic_bool use_proxy {false};
 
@@ -490,17 +487,6 @@ private:
      * reset dht clients
      */
     void resetDht();
-    /**
-     * @return the current active DHT
-     */
-    SecureDht* activeDht() const;
-
-    /**
-     * Store current listeners and translates global tokens for each client.
-     */
-    struct Listener;
-    std::map<size_t, Listener> listeners_;
-    size_t listener_token_ {1};
 
     mutable std::mutex dht_mtx {};
     std::thread dht_thread {};

include/opendht/infohash.h

@@ -103,13 +103,7 @@ public:
     static constexpr inline Hash zero() noexcept { return Hash{}; }
 
     bool operator==(const Hash& h) const {
-        auto a = reinterpret_cast<const uint32_t*>(data_.data());
-        auto b = reinterpret_cast<const uint32_t*>(h.data_.data());
-        constexpr unsigned n = N / sizeof(uint32_t);
-        for (unsigned i=0; i < n; i++)
-            if (a[i] != b[i])
-                return false;
-        return true;
+        return data_ == h.data_;
     }
     bool operator!=(const Hash& h) const { return !(*this == h); }
 

include/opendht/network_engine.h

@@ -252,12 +252,12 @@ public:
      * @param values      The values to send.
      * @param version     If version = 1, a request will be used to answer to the listener
      */
-    void tellListener(Sp<Node> n, Tid socket_id, const InfoHash& hash, want_t want, const Blob& ntoken,
+    void tellListener(const Sp<Node>& n, Tid socket_id, const InfoHash& hash, want_t want, const Blob& ntoken,
             std::vector<Sp<Node>>&& nodes, std::vector<Sp<Node>>&& nodes6,
             std::vector<Sp<Value>>&& values, const Query& q, int version);
 
-    void tellListenerRefreshed(Sp<Node> n, Tid socket_id, const InfoHash& hash, const Blob& ntoken, const std::vector<Value::Id>& values, int version);
-    void tellListenerExpired(Sp<Node> n, Tid socket_id, const InfoHash& hash, const Blob& ntoken, const std::vector<Value::Id>& values, int version);
+    void tellListenerRefreshed(const Sp<Node>& n, Tid socket_id, const InfoHash& hash, const Blob& ntoken, const std::vector<Value::Id>& values, int version);
+    void tellListenerExpired(const Sp<Node>& n, Tid socket_id, const InfoHash& hash, const Blob& ntoken, const std::vector<Value::Id>& values, int version);
 
     bool isRunning(sa_family_t af) const;
     inline want_t want () const { return dht_socket->hasIPv4() and dht_socket->hasIPv6() ? (WANT4 | WANT6) : -1; }
@@ -278,7 +278,7 @@ public:
      * @return the request with information concerning its success.
      */
     Sp<Request>
-    sendPing(Sp<Node> n, RequestCb&& on_done, RequestExpiredCb&& on_expired);
+    sendPing(const Sp<Node>& n, RequestCb&& on_done, RequestExpiredCb&& on_expired);
 
     /**
      * Send a "ping" request to a given node.
@@ -309,7 +309,7 @@ public:
      *
      * @return the request with information concerning its success.
      */
-    Sp<Request> sendFindNode(Sp<Node> n,
+    Sp<Request> sendFindNode(const Sp<Node>& n,
                              const InfoHash& hash,
                              want_t want = -1,
                              RequestCb&& on_done = {},
@@ -328,7 +328,7 @@ public:
      *
      * @return the request with information concerning its success.
      */
-    Sp<Request> sendGetValues(Sp<Node> n,
+    Sp<Request> sendGetValues(const Sp<Node>& n,
                               const InfoHash& hash,
                               const Query& query,
                               want_t want,
@@ -357,7 +357,7 @@ public:
      *
      * @return the request with information concerning its success.
      */
-    Sp<Request> sendListen(Sp<Node> n,
+    Sp<Request> sendListen(const Sp<Node>& n,
                            const InfoHash& hash,
                            const Query& query,
                            const Blob& token,
@@ -377,7 +377,7 @@ public:
      *
      * @return the request with information concerning its success.
      */
-    Sp<Request> sendAnnounceValue(Sp<Node> n,
+    Sp<Request> sendAnnounceValue(const Sp<Node>& n,
                                   const InfoHash& hash,
                                   const Sp<Value>& v,
                                   time_point created,
@@ -397,7 +397,7 @@ public:
      *
      * @return the request with information concerning its success.
      */
-    Sp<Request> sendRefreshValue(Sp<Node> n,
+    Sp<Request> sendRefreshValue(const Sp<Node>& n,
                                  const InfoHash& hash,
                                  const Value::Id& vid,
                                  const Blob& token,
@@ -416,7 +416,7 @@ public:
      *
      * @return the request with information concerning its success.
      */
-    Sp<Request> sendUpdateValues(Sp<Node> n,
+    Sp<Request> sendUpdateValues(const Sp<Node>& n,
                                  const InfoHash& infohash,
                                  const std::vector<Sp<Value>>& values,
                                  time_point created,

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/sockaddr.h

@@ -236,6 +236,15 @@ public:
         return *reinterpret_cast<sockaddr_in6*>(get());
     }
 
+    /** 
+     * Releases the ownership of the managed object, if any.
+     * The caller is responsible for deleting the object with free().
+     */
+    inline sockaddr* release() {
+        len = 0;
+        return addr.release();
+    }
+
     /**
      * Return true if address is a loopback IP address.
      */

include/opendht/value.h

@@ -213,7 +213,7 @@ struct OPENDHT_PUBLIC Value
         }
         static Filter notFilter(Filter&& f) {
             if (not f) return [](const Value&) { return false; };
-            return [f](const Value& v) { return not f(v); };
+            return [f = std::move(f)](const Value& v) { return not f(v); };
         }
         std::vector<Sp<Value>> filter(const std::vector<Sp<Value>>& values) {
             if (not (*this))
@@ -343,10 +343,10 @@ struct OPENDHT_PUBLIC Value
         return unpack<T>(*this);
     }
 
-    bool isEncrypted() const {
+    inline bool isEncrypted() const {
         return not cypher.empty();
     }
-    bool isSigned() const {
+    inline bool isSigned() const {
         return owner and not signature.empty();
     }
 
@@ -355,37 +355,24 @@ struct OPENDHT_PUBLIC Value
      * Afterward, checkSignature() will return true and owner will
      * be set to the corresponding public key.
      */
-    void sign(const crypto::PrivateKey& key) {
-        if (isEncrypted())
-            throw DhtException("Can't sign encrypted data.");
-        owner = key.getSharedPublicKey();
-        signature = key.sign(getToSign());
-    }
+    void sign(const crypto::PrivateKey& key);
 
     /**
      * Check that the value is signed and that the signature matches.
      * If true, the owner field will contain the signer public key.
      */
-    bool checkSignature() const {
+    inline bool checkSignature() const {
         return isSigned() and owner->checkSignature(getToSign(), signature);
     }
 
-    std::shared_ptr<crypto::PublicKey> getOwner() const {
-        return std::static_pointer_cast<crypto::PublicKey>(owner);
+    inline std::shared_ptr<crypto::PublicKey> getOwner() const {
+        return owner;
     }
 
     /**
      * Sign the value with from and returns the encrypted version for to.
      */
-    Value encrypt(const crypto::PrivateKey& from, const crypto::PublicKey& to) {
-        if (isEncrypted())
-            throw DhtException("Data is already encrypted.");
-        setRecipient(to.getId());
-        sign(from);
-        Value nv {id};
-        nv.setCypher(to.encrypt(getToEncrypt()));
-        return nv;
-    }
+    Value encrypt(const crypto::PrivateKey& from, const crypto::PublicKey& to);
 
     Value() {}
 
@@ -453,18 +440,18 @@ struct OPENDHT_PUBLIC Value
         return id == o.id and contentEquals(o);
     }
 
-    void setRecipient(const InfoHash& r) {
+    inline void setRecipient(const InfoHash& r) {
         recipient = r;
     }
 
-    void setCypher(Blob&& c) {
+    inline void setCypher(Blob&& c) {
         cypher = std::move(c);
     }
 
     /**
      * Pack part of the data to be signed (must always be done the same way)
      */
-    Blob getToSign() const {
+    inline Blob getToSign() const {
         msgpack::sbuffer buffer;
         msgpack::packer<msgpack::sbuffer> pk(&buffer);
         msgpack_pack_to_sign(pk);
@@ -474,7 +461,7 @@ struct OPENDHT_PUBLIC Value
     /**
      * Pack part of the data to be encrypted
      */
-    Blob getToEncrypt() const {
+    inline Blob getToEncrypt() const {
         msgpack::sbuffer buffer;
         msgpack::packer<msgpack::sbuffer> pk(&buffer);
         msgpack_pack_to_encrypt(pk);
@@ -484,7 +471,7 @@ struct OPENDHT_PUBLIC Value
     /** print value for debugging */
     OPENDHT_PUBLIC friend std::ostream& operator<< (std::ostream& s, const Value& v);
 
-    std::string toString() const {
+    inline std::string toString() const {
         std::stringstream ss;
         ss << *this;
         return ss.str();
@@ -638,8 +625,16 @@ struct OPENDHT_PUBLIC Value
      */
     unsigned priority {0};
 
+    inline bool isSignatureChecked() const {
+        return signatureChecked;
+    }
+    inline bool isDecrypted() const {
+        return decrypted;
+    }
+    bool checkSignature();
+    Sp<Value> decrypt(const crypto::PrivateKey& key);
+
 private:
-    friend class SecureDht;
     /* Cache for crypto ops */
     bool signatureChecked {false};
     bool signatureValid {false};

opendht.pc.in

@@ -3,7 +3,7 @@ exec_prefix=@exec_prefix@
 libdir=@libdir@
 includedir=@includedir@
 Name: OpenDHT
-Description: C++14 Distributed Hash Table library
+Description: C++17 Distributed Hash Table library
 Version: @VERSION@
 Libs: -L${libdir} -lopendht
 Libs.private: @http_parser_lib@ -pthread

python/tools/README.md

@@ -20,7 +20,7 @@ help).
 
 ## Python dependencies
 
-- pyroute2 >=0.3.14
+- pyroute2 >=0.6.9
 - matplotlib
 - GeoIP (used by `scanner.py` for drawing map of the world)
 - ipaddress

python/tools/dht/network.py

@@ -30,7 +30,7 @@ import traceback
 import ipaddress
 import netifaces
 import numpy as np
-from pyroute2.netns.process.proxy import NSPopen
+from pyroute2 import NSPopen
 import msgpack
 
 from opendht import *

python/tools/dht/virtual_network_builder.py

@@ -15,107 +15,253 @@
 #
 # You should have received a copy of the GNU General Public License
 # along with this program; If not, see <http://www.gnu.org/licenses/>.
+import argparse
+import os
+import subprocess
 
-import argparse, subprocess
+from pyroute2 import NDB, NSPopen
 
-from pyroute2 import IPDB, NetNS
-from pyroute2.netns.process.proxy import NSPopen
 
-if __name__ == "__main__":
-    parser = argparse.ArgumentParser(description='Creates a virtual network topology for testing')
-    parser.add_argument('-i', '--ifname', help='interface name', default='ethdht')
-    parser.add_argument('-n', '--ifnum', type=int, help='number of isolated interfaces to create', default=1)
-    parser.add_argument('-r', '--remove', help='remove instead of adding network interfaces', action="store_true")
-    parser.add_argument('-l', '--loss', help='simulated packet loss (percent)', type=int, default=0)
-    parser.add_argument('-d', '--delay', help='simulated latency (ms)', type=int, default=0)
-    parser.add_argument('-4', '--ipv4', help='Enable IPv4', action="store_true")
-    parser.add_argument('-6', '--ipv6', help='Enable IPv6', action="store_true")
+def int_range(mini, maxi):
+    def check_ifnum(arg):
+        try:
+            ret = int(arg)
+        except ValueError:
+            raise argparse.ArgumentTypeError('must be an integer')
+        if ret > maxi or ret < mini:
+            raise argparse.ArgumentTypeError(
+                f'must be {mini} <= int <= {maxi}'
+            )
+        return ret
+
+    return check_ifnum
+
+
+if __name__ == '__main__':
+    parser = argparse.ArgumentParser(
+        description='Creates a virtual network topology for testing'
+    )
+    parser.add_argument(
+        '-i', '--ifname', help='interface name', default='ethdht'
+    )
+    parser.add_argument(
+        '-n',
+        '--ifnum',
+        type=int_range(1, 245),
+        help='number of isolated interfaces to create',
+        default=1,
+    )
+    parser.add_argument(
+        '-r',
+        '--remove',
+        help='remove instead of adding network interfaces',
+        action='store_true',
+    )
+    parser.add_argument(
+        '-l',
+        '--loss',
+        help='simulated packet loss (percent)',
+        type=int,
+        default=0,
+    )
+    parser.add_argument(
+        '-d', '--delay', help='simulated latency (ms)', type=int, default=0
+    )
+    parser.add_argument(
+        '-4', '--ipv4', help='Enable IPv4', action='store_true'
+    )
+    parser.add_argument(
+        '-6', '--ipv6', help='Enable IPv6', action='store_true'
+    )
+    parser.add_argument(
+        '-b',
+        '--debug',
+        help='Turn on debug logging and dump topology databases',
+        action='store_true',
+    )
+    parser.add_argument(
+        '-v',
+        '--verbose',
+        help='Turn on verbose output on netns and interfaces operations',
+        action='store_true',
+    )
 
     args = parser.parse_args()
 
     local_addr4 = '10.0.42.'
     local_addr6 = '2001:db9::'
-    brige_name = 'br'+args.ifname
+    bripv4 = f'{local_addr4}1/24'
+    bripv6 = f'{local_addr6}1/64'
+    bridge_name = f'br{args.ifname}'
+    tap_name = f'tap{args.ifname}'
+    veth_names = []
+    namespaces = []
+    ipv4addrs = []
+    ipv6addrs = []
+    for ifn in range(args.ifnum):
+        namespaces.append(f'node{ifn}')
+        veth_names.append(f'{args.ifname}{ifn}')
+        ipv4addrs.append(f'{local_addr4}{ifn+8}/24' if args.ipv4 else None)
+        ipv6addrs.append(f'{local_addr6}{ifn+8}/64' if args.ipv6 else None)
 
-    ip = None
-    try:
-        ip = IPDB()
+    with NDB(log='debug' if args.debug else None) as ndb:
         if args.remove:
-            # cleanup interfaces
-            for ifn in range(args.ifnum):
-                iface = args.ifname+str(ifn)
-                if iface in ip.interfaces:
-                    with ip.interfaces[iface] as i:
-                        i.remove()
-            if 'tap'+args.ifname in ip.interfaces:
-                with ip.interfaces['tap'+args.ifname] as i:
-                    i.remove()
-            if brige_name in ip.interfaces:
-                with ip.interfaces[brige_name] as i:
-                    i.remove()
-            for ifn in range(args.ifnum):
-                netns = NetNS('node'+str(ifn))
-                netns.close()
-                netns.remove()
-        else:
-            for ifn in range(args.ifnum):
-                iface = args.ifname+str(ifn)
-                if not iface in ip.interfaces:
-                    ip.create(kind='veth', ifname=iface, peer=iface+'.1').commit()
+            # cleanup interfaces in the main namespace
+            for iface in veth_names + [bridge_name] + [tap_name]:
+                if iface in ndb.interfaces:
+                    ndb.interfaces[iface].remove().commit()
+                    if args.verbose:
+                        print(f'link: del main/{iface}')
 
-            ip.create(kind='tuntap', ifname='tap'+args.ifname, mode='tap').commit()
-
-            with ip.create(kind='bridge', ifname=brige_name) as i:
-                for ifn in range(args.ifnum):
-                    iface = args.ifname+str(ifn)
-                    i.add_port(ip.interfaces[iface])
-                i.add_port(ip.interfaces['tap'+args.ifname])
-                if args.ipv4:
-                    i.add_ip(local_addr4+'1/24')
-                if args.ipv6:
-                    i.add_ip(local_addr6+'1/64')
-                i.up()
-
-            with ip.interfaces['tap'+args.ifname] as tap:
-                tap.up()
-
-            for ifn in range(args.ifnum):
-                iface = args.ifname+str(ifn)
-
-                nns = NetNS('node'+str(ifn))
-                iface1 = iface+'.1'
-                with ip.interfaces[iface1] as i:
-                    i.net_ns_fd = nns.netns
-
-                with ip.interfaces[iface] as i:
-                    i.up()
-
-                ip_ns = IPDB(nl=nns)
+            # cleanup namespaces
+            for nsname in namespaces:
                 try:
-                    with ip_ns.interfaces.lo as lo:
-                        lo.up()
-                    with ip_ns.interfaces[iface1] as i:
-                        if args.ipv4:
-                            i.add_ip(local_addr4+str(ifn+8)+'/24')
-                        if args.ipv6:
-                            i.add_ip(local_addr6+str(ifn+8)+'/64')
-                        i.up()
-                finally:
-                    ip_ns.release()
-
-                nsp = NSPopen(nns.netns, ["tc", "qdisc", "add", "dev", iface1, "root", "netem", "delay", str(args.delay)+"ms", str(int(args.delay/2))+"ms", "loss", str(args.loss)+"%", "25%"], stdout=subprocess.PIPE)
-                #print(nsp.communicate()[0].decode())
+                    ndb.netns[nsname].remove().commit()
+                    if args.verbose:
+                        print(f'netns: del {nsname}')
+                except KeyError:
+                    pass
+        else:
+            # create ports
+            for veth, nsname, ipv4addr, ipv6addr in zip(
+                veth_names, namespaces, ipv4addrs, ipv6addrs
+            ):
+                # create a network namespace and launch NDB for it
+                #
+                # another possible solution could be simply to attach
+                # the namespace to the main NDB instance, but it can
+                # take a lot of memory in case of many interfaces, thus
+                # launch and discard netns NDB instances
+                netns = NDB(
+                    log='debug' if args.debug else None,
+                    sources=[
+                        {
+                            'target': 'localhost',
+                            'netns': nsname,
+                            'kind': 'netns',
+                        }
+                    ],
+                )
+                if args.verbose:
+                    print(f'netns: add {nsname}')
+                # create the port and push the peer into the namespace
+                (
+                    ndb.interfaces.create(
+                        **{
+                            'ifname': veth,
+                            'kind': 'veth',
+                            'state': 'up',
+                            'peer': {'ifname': veth, 'net_ns_fd': nsname},
+                        }
+                    ).commit()
+                )
+                if args.verbose:
+                    print(f'link: add main/{veth} <-> {nsname}/{veth}')
+                # bring up namespace's loopback
+                (
+                    netns.interfaces.wait(ifname='lo', timeout=3)
+                    .set('state', 'up')
+                    .commit()
+                )
+                if args.verbose:
+                    print(f'link: set {nsname}/lo')
+                # bring up the peer
+                with netns.interfaces.wait(ifname=veth, timeout=3) as i:
+                    i.set('state', 'up')
+                    if args.ipv4:
+                        i.add_ip(ipv4addr)
+                    if args.ipv6:
+                        i.add_ip(ipv6addr)
+                if args.verbose:
+                    print(f'link: set {nsname}/{veth}, {ipv4addr}, {ipv6addr}')
+                # disconnect the namespace NDB agent, not removing the NS
+                if args.debug:
+                    fname = f'{nsname}-ndb.db'
+                    print(f'dump: netns topology database {fname}')
+                    netns.schema.backup(fname)
+                netns.close()
+                # set up the emulation QDisc
+                nsp = NSPopen(
+                    nsname,
+                    [
+                        'tc',
+                        'qdisc',
+                        'add',
+                        'dev',
+                        veth,
+                        'root',
+                        'netem',
+                        'delay',
+                        f'{args.delay}ms',
+                        f'{int(args.delay)/2}ms',
+                        'loss',
+                        f'{args.loss}%',
+                        '25%',
+                    ],
+                    stdout=subprocess.PIPE,
+                )
                 nsp.communicate()
                 nsp.wait()
                 nsp.release()
+                if args.verbose:
+                    print(
+                        f'netem: add {nsname}/{veth}, '
+                        f'{args.delay}, {args.loss}'
+                    )
 
-            if args.ipv4:
-                subprocess.call(["sysctl", "-w", "net.ipv4.conf."+brige_name+".forwarding=1"])
-            if args.ipv6:
-                subprocess.call(["sysctl", "-w", "net.ipv6.conf."+brige_name+".forwarding=1"])
+            # create the tap
+            #
+            # for some reason we should create the tap inteface first,
+            # and only then bring it up, thus two commit() calls
+            (
+                ndb.interfaces.create(
+                    kind='tuntap', ifname=tap_name, mode='tap'
+                )
+                .commit()
+                .set('state', 'up')
+                .commit()
+            )
+            if args.verbose:
+                print(f'link: add main/{tap_name}')
 
-    except Exception as e:
-          print('Error',e)
-    finally:
-        if ip:
-            ip.release()
+            # create the bridge and add all the ports
+            with ndb.interfaces.create(
+                ifname=bridge_name, kind='bridge', state='up'
+            ) as i:
+                if args.ipv4:
+                    i.add_ip(bripv4)
+                if args.ipv6:
+                    i.add_ip(bripv6)
+                for iface in veth_names + [tap_name]:
+                    i.add_port(iface)
+            if args.verbose:
+                print(f'link: add main/{bridge_name}, {bripv4}, {bripv6}')
+
+            with open(os.devnull, 'w') as fnull:
+                if args.ipv4:
+                    subprocess.call(
+                        [
+                            'sysctl',
+                            '-w',
+                            f'net.ipv4.conf.{bridge_name}.forwarding=1',
+                        ],
+                        stdout=fnull,
+                    )
+                if args.verbose:
+                    print(f'sysctl: set {bridge_name} ipv4 forwarding')
+                if args.ipv6:
+                    subprocess.call(
+                        [
+                            'sysctl',
+                            '-w',
+                            f'net.ipv6.conf.{bridge_name}.forwarding=1',
+                        ],
+                        stdout=fnull,
+                    )
+                if args.verbose:
+                    print(f'sysctl: set {bridge_name} ipv4 forwarding')
+
+            if args.debug:
+                fname = 'main-ndb.db'
+                print('dump: the main netns topology database')
+                ndb.schema.backup(fname)

src/Makefile.am

@@ -84,6 +84,3 @@ if ENABLE_INDEXATION
 libopendht_la_SOURCES += indexation/pht.cpp
 nobase_include_HEADERS += ../include/opendht/indexation/pht.h
 endif
-
-clean-local:
-	rm -rf libargon2.la

src/crypto.cpp

@@ -43,18 +43,6 @@ static std::uniform_int_distribution<int> rand_byte{ 0, std::numeric_limits<uint
 static std::uniform_int_distribution<uint8_t> rand_byte;
 #endif
 
-// support for GnuTLS < 3.4.
-#if GNUTLS_VERSION_NUMBER < 0x030400
-#define GNUTLS_PKCS_PKCS12_3DES GNUTLS_PKCS_USE_PKCS12_3DES
-#define GNUTLS_PKCS_PKCS12_ARCFOUR GNUTLS_PKCS_USE_PKCS12_ARCFOUR
-#define GNUTLS_PKCS_PKCS12_RC2_40 GNUTLS_PKCS_USE_PKCS12_RC2_40
-#define GNUTLS_PKCS_PBES2_3DES GNUTLS_PKCS_USE_PBES2_3DES
-#define GNUTLS_PKCS_PBES2_AES_128 GNUTLS_PKCS_USE_PBES2_AES_128
-#define GNUTLS_PKCS_PBES2_AES_192 GNUTLS_PKCS_USE_PBES2_AES_192
-#define GNUTLS_PKCS_PBES2_AES_256 GNUTLS_PKCS_USE_PBES2_AES_256
-#endif
-
-#define DHT_AES_LEGACY_ENCRYPT 0
 #define DHT_AES_LEGACY_DECRYPT 1
 
 namespace dht {
@@ -110,10 +98,6 @@ Blob aesEncrypt(const uint8_t* data, size_t data_length, const Blob& key)
     struct gcm_aes_ctx aes;
     gcm_aes_set_key(&aes, key.size(), key.data());
     gcm_aes_set_iv(&aes, GCM_IV_SIZE, ret.data());
-#if DHT_AES_LEGACY_ENCRYPT
-    gcm_aes_update(&aes, data_length, data);
-#endif
-
     gcm_aes_encrypt(&aes, data_length, ret.data() + GCM_IV_SIZE, data);
     gcm_aes_digest(&aes, GCM_DIGEST_SIZE, ret.data() + GCM_IV_SIZE + data_length);
     return ret;
@@ -377,12 +361,13 @@ PrivateKey::getPublicKey() const
 const std::shared_ptr<PublicKey>&
 PrivateKey::getSharedPublicKey() const
 {
-    if (publicKey_)
-        return publicKey_;
-    auto pk = std::make_shared<PublicKey>();
-    if (auto err = gnutls_pubkey_import_privkey(pk->pk, key, GNUTLS_KEY_KEY_CERT_SIGN | GNUTLS_KEY_CRL_SIGN, 0))
-        throw CryptoException(std::string("Can't retreive public key: ") + gnutls_strerror(err));
-    publicKey_ = pk;
+    std::lock_guard<std::mutex> lock(publicKeyMutex_);
+    if (not publicKey_) {
+        auto pk = std::make_shared<PublicKey>();
+        if (auto err = gnutls_pubkey_import_privkey(pk->pk, key, GNUTLS_KEY_KEY_CERT_SIGN | GNUTLS_KEY_CRL_SIGN, 0))
+            throw CryptoException(std::string("Can't retreive public key: ") + gnutls_strerror(err));
+        publicKey_ = pk;
+    }
     return publicKey_;
 }
 
@@ -542,36 +527,36 @@ PublicKey::encrypt(const uint8_t* data, size_t data_len) const
     return ret;
 }
 
-InfoHash
+const InfoHash&
 PublicKey::getId() const
 {
-    if (not pk)
-        return {};
-    InfoHash id;
-    size_t sz = id.size();
-    if (auto err = gnutls_pubkey_get_key_id(pk, 0, id.data(), &sz))
-        throw CryptoException(std::string("Can't get public key ID: ") + gnutls_strerror(err));
-    if (sz != id.size())
-        throw CryptoException("Can't get public key ID: wrong output length.");
-    return id;
+    if (pk and not idCached_.load()) {
+        InfoHash id;
+        size_t sz = id.size();
+        if (auto err = gnutls_pubkey_get_key_id(pk, 0, id.data(), &sz))
+            throw CryptoException(std::string("Can't get public key ID: ") + gnutls_strerror(err));
+        if (sz != id.size())
+            throw CryptoException("Can't get public key ID: wrong output length.");
+        cachedId_ = id;
+        idCached_.store(true);
+    }
+    return cachedId_;
 }
 
-PkId
+const PkId&
 PublicKey::getLongId() const
 {
-    if (not pk)
-        return {};
-#if GNUTLS_VERSION_NUMBER < 0x030401
-    throw CryptoException("Can't get 256 bits public key ID: GnuTLS 3.4.1 or higher required.");
-#else
-    PkId h;
-    size_t sz = h.size();
-    if (auto err = gnutls_pubkey_get_key_id(pk, GNUTLS_KEYID_USE_SHA256, h.data(), &sz))
-        throw CryptoException(std::string("Can't get 256 bits public key ID: ") + gnutls_strerror(err));
-    if (sz != h.size())
-        throw CryptoException("Can't get 256 bits public key ID: wrong output length.");
-    return h;
-#endif
+    if (pk and not longIdCached_.load()) {
+        PkId h;
+        size_t sz = h.size();
+        if (auto err = gnutls_pubkey_get_key_id(pk, GNUTLS_KEYID_USE_SHA256, h.data(), &sz))
+            throw CryptoException(std::string("Can't get 256 bits public key ID: ") + gnutls_strerror(err));
+        if (sz != h.size())
+            throw CryptoException("Can't get 256 bits public key ID: wrong output length.");
+        cachedLongId_ = h;
+        longIdCached_.store(true);
+    }
+    return cachedLongId_;
 }
 
 gnutls_digest_algorithm_t
@@ -848,42 +833,36 @@ Certificate::getPublicKey() const
     return pk_ret;
 }
 
-InfoHash
+const InfoHash&
 Certificate::getId() const
 {
-    if (not cert)
-        return {};
-    if (cachedId_)
-        return cachedId_;
-    InfoHash id;
-    size_t sz = id.size();
-    if (auto err = gnutls_x509_crt_get_key_id(cert, 0, id.data(), &sz))
-        throw CryptoException(std::string("Can't get certificate public key ID: ") + gnutls_strerror(err));
-    if (sz != id.size())
-        throw CryptoException("Can't get certificate public key ID: wrong output length.");
-    cachedId_ = id;
-    return id;
+    if (cert and not idCached_.load()) {
+        InfoHash id;
+        size_t sz = id.size();
+        if (auto err = gnutls_x509_crt_get_key_id(cert, 0, id.data(), &sz))
+            throw CryptoException(std::string("Can't get certificate public key ID: ") + gnutls_strerror(err));
+        if (sz != id.size())
+            throw CryptoException("Can't get certificate public key ID: wrong output length.");
+        cachedId_ = id;
+        idCached_.store(true);
+    }
+    return cachedId_;
 }
 
-PkId
+const PkId&
 Certificate::getLongId() const
 {
-    if (not cert)
-        return {};
-    if (cachedLongId_)
-        return cachedLongId_;
-#if GNUTLS_VERSION_NUMBER < 0x030401
-    throw CryptoException("Can't get certificate 256 bits public key ID: GnuTLS 3.4.1 or higher required.");
-#else
-    PkId id;
-    size_t sz = id.size();
-    if (auto err = gnutls_x509_crt_get_key_id(cert, GNUTLS_KEYID_USE_SHA256, id.data(), &sz))
-        throw CryptoException(std::string("Can't get certificate 256 bits public key ID: ") + gnutls_strerror(err));
-    if (sz != id.size())
-        throw CryptoException("Can't get certificate 256 bits public key ID: wrong output length.");
-    cachedLongId_ = id;
-    return id;
-#endif
+    if (cert and not longIdCached_.load()) {
+        PkId id;
+        size_t sz = id.size();
+        if (auto err = gnutls_x509_crt_get_key_id(cert, GNUTLS_KEYID_USE_SHA256, id.data(), &sz))
+            throw CryptoException(std::string("Can't get certificate 256 bits public key ID: ") + gnutls_strerror(err));
+        if (sz != id.size())
+            throw CryptoException("Can't get certificate 256 bits public key ID: wrong output length.");
+        cachedLongId_ = id;
+        longIdCached_.store(true);
+    }
+    return cachedLongId_;
 }
 
 Blob
@@ -1771,10 +1750,8 @@ std::ostream& operator<< (std::ostream& o, const TrustList::VerifyResult& h)
             o << "* Certificate has expired" << std::endl;
         if (h.result & GNUTLS_CERT_UNEXPECTED_OWNER)
             o << "* The owner is not the expected one" << std::endl;
-#if GNUTLS_VERSION_NUMBER >= 0x030401
         if (h.result & GNUTLS_CERT_PURPOSE_MISMATCH)
             o << "* Certificate or an intermediate does not match the intended purpose" << std::endl;
-#endif
         if (h.result & GNUTLS_CERT_MISMATCH)
             o << "* Certificate presented isn't the expected one" << std::endl;
     } else {

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

@@ -38,15 +38,6 @@ namespace dht {
 
 static const std::string PEER_DISCOVERY_DHT_SERVICE = "dht";
 
-struct DhtRunner::Listener {
-    size_t tokenClassicDht {0};
-    size_t tokenProxyDht {0};
-    ValueCallback gcb;
-    InfoHash hash {};
-    Value::Filter f;
-    Where w;
-};
-
 struct NodeInsertionPack {
     dht::InfoHash nodeId;
     in_port_t port;
@@ -55,9 +46,6 @@ struct NodeInsertionPack {
 };
 
 DhtRunner::DhtRunner() : dht_()
-#ifdef OPENDHT_PROXY_CLIENT
-, dht_via_proxy_()
-#endif //OPENDHT_PROXY_CLIENT
 {
 #ifdef _WIN32
     WSADATA wsd;
@@ -119,8 +107,9 @@ DhtRunner::run(const Config& config, Context&& context)
             local6.setFamily(AF_INET6);
         }
         auto state_path = config.dht_config.node_config.persist_path;
-        if (not state_path.empty() && (local4.getPort() == 0 || local6.getPort() == 0)) {
+        if (not state_path.empty())
             state_path += "_port.txt";
+        if (not state_path.empty() && (local4.getPort() == 0 || local6.getPort() == 0)) {
             std::ifstream inConfig(state_path);
             if (inConfig.is_open()) {
                 in_port_t port;
@@ -143,68 +132,59 @@ DhtRunner::run(const Config& config, Context&& context)
 
         if (context.logger) {
             logger_ = context.logger;
-        }
-
-        if (not context.sock) {
-            context.sock.reset(new net::UdpSocket(local4, local6, context.logger));
-        }
-
-        if (not state_path.empty()) {
-            std::ofstream outConfig(state_path);
-            outConfig << context.sock->getBoundRef(AF_INET).getPort() << std::endl;
-            outConfig << context.sock->getBoundRef(AF_INET6).getPort() << std::endl;
-        }
-
-        if (context.logger) {
             logger_->d("[runner %p] state changed to Running", this);
         }
 
-        context.sock->setOnReceive([&] (net::PacketList&& pkts) {
-            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));
-                ret = std::move(rcv_free);
-            }
-            cv.notify_all();
-            return ret;
-        });
-
 #ifdef OPENDHT_PROXY_CLIENT
         config_ = config;
         identityAnnouncedCb_ = context.identityAnnouncedCb;
 #endif
-        auto dht = std::make_unique<Dht>(std::move(context.sock), SecureDht::getConfig(config.dht_config), context.logger);
-        dht_ = std::make_unique<SecureDht>(std::move(dht), config.dht_config, std::move(context.identityAnnouncedCb), context.logger);
-        enableProxy(not config.proxy_server.empty());
+
+        if (config.proxy_server.empty()) {
+            if (not context.sock) {
+                context.sock.reset(new net::UdpSocket(local4, local6, context.logger));
+            }
+            context.sock->setOnReceive([&] (net::PacketList&& pkts) {
+                net::PacketList ret;
+                {
+                    std::lock_guard<std::mutex> lck(sock_mtx);
+                    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();
+                return ret;
+            });
+            if (not state_path.empty()) {
+                std::ofstream outConfig(state_path);
+                outConfig << context.sock->getBoundRef(AF_INET).getPort() << std::endl;
+                outConfig << context.sock->getBoundRef(AF_INET6).getPort() << std::endl;
+            }
+            auto dht = std::make_unique<Dht>(std::move(context.sock), SecureDht::getConfig(config.dht_config), context.logger);
+            dht_ = std::make_unique<SecureDht>(std::move(dht), config.dht_config, std::move(context.identityAnnouncedCb), context.logger);
+        } else {
+            enableProxy(true);
+        }
     } catch(const std::exception& e) {
         config_ = {};
         identityAnnouncedCb_ = {};
         dht_.reset();
-#ifdef OPENDHT_PROXY_CLIENT
-        dht_via_proxy_.reset();
-#endif
         running = State::Idle;
         throw;
     }
 
-    if (context.logger and dht_via_proxy_) {
-        dht_via_proxy_->setLogger(context.logger);
-    }
     if (context.statusChangedCallback) {
         statusCb = std::move(context.statusChangedCallback);
     }
     if (context.certificateStore) {
         dht_->setLocalCertificateStore(std::move(context.certificateStore));
-        if (dht_via_proxy_)
-            dht_via_proxy_->setLocalCertificateStore(std::move(context.certificateStore));
     }
 
     if (not config.threaded)
@@ -287,7 +267,8 @@ DhtRunner::shutdown(ShutdownCallback cb, bool stop) {
     auto expected = State::Running;
     if (not running.compare_exchange_strong(expected, State::Stopping)) {
         if (expected == State::Stopping and ongoing_ops) {
-            shutdownCallbacks_.emplace_back(std::move(cb));
+            if (cb)
+                shutdownCallbacks_.emplace_back(std::move(cb));
         }
         else if (cb) {
             lck.unlock();
@@ -301,12 +282,10 @@ DhtRunner::shutdown(ShutdownCallback cb, bool stop) {
     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);
-#endif
         if (dht_)
             dht_->shutdown(onShutdown, stop);
+        else
+            opEnded();
     });
     cv.notify_all();
 }
@@ -409,31 +388,25 @@ void
 DhtRunner::dumpTables() const
 {
     std::lock_guard<std::mutex> lck(dht_mtx);
-    activeDht()->dumpTables();
+    dht_->dumpTables();
 }
 
 InfoHash
 DhtRunner::getId() const
 {
-    if (auto dht = activeDht())
-        return dht->getId();
-    return {};
+    return dht_ ? dht_->getId() : InfoHash{};
 }
 
 std::shared_ptr<crypto::PublicKey>
 DhtRunner::getPublicKey() const
 {
-    if (auto dht = activeDht())
-        return dht->getPublicKey();
-    return {};
+    return dht_ ? dht_->getPublicKey() : std::shared_ptr<crypto::PublicKey>{};
 }
 
 InfoHash
 DhtRunner::getNodeId() const
 {
-    if (auto dht = activeDht())
-        return dht->getNodeId();
-    return {};
+    return dht_ ? dht_->getNodeId() : InfoHash{};
 }
 
 
@@ -475,10 +448,6 @@ DhtRunner::setLogger(const Sp<Logger>& logger) {
     logger_ = logger;
     if (dht_)
         dht_->setLogger(logger);
-#ifdef OPENDHT_PROXY_CLIENT
-    if (dht_via_proxy_)
-        dht_via_proxy_->setLogger(logger);
-#endif
 }
 
 void
@@ -492,16 +461,12 @@ DhtRunner::setLogFilter(const InfoHash& f) {
     std::lock_guard<std::mutex> lck(dht_mtx);
     if (dht_)
         dht_->setLogFilter(f);
-#ifdef OPENDHT_PROXY_CLIENT
-    if (dht_via_proxy_)
-        dht_via_proxy_->setLogFilter(f);
-#endif
 }
 
 void
 DhtRunner::registerType(const ValueType& type) {
     std::lock_guard<std::mutex> lck(dht_mtx);
-    activeDht()->registerType(type);
+    dht_->registerType(type);
 }
 
 void
@@ -514,7 +479,7 @@ unsigned
 DhtRunner::getNodesStats(sa_family_t af, unsigned *good_return, unsigned *dubious_return, unsigned *cached_return, unsigned *incoming_return) const
 {
     std::lock_guard<std::mutex> lck(dht_mtx);
-    const auto stats = activeDht()->getNodesStats(af);
+    const auto stats = dht_->getNodesStats(af);
     if (good_return)
         *good_return = stats.good_nodes;
     if (dubious_return)
@@ -530,19 +495,19 @@ NodeStats
 DhtRunner::getNodesStats(sa_family_t af) const
 {
     std::lock_guard<std::mutex> lck(dht_mtx);
-    return activeDht()->getNodesStats(af);
+    return dht_->getNodesStats(af);
 }
 
 NodeInfo
 DhtRunner::getNodeInfo() const {
     std::lock_guard<std::mutex> lck(dht_mtx);
     NodeInfo info {};
-    if (auto dht = activeDht()) {
-        info.id = dht->getId();
-        info.node_id = dht->getNodeId();
-        info.ipv4 = dht->getNodesStats(AF_INET);
-        info.ipv6 = dht->getNodesStats(AF_INET6);
-        if (auto sock = dht->getSocket()) {
+    if (dht_) {
+        info.id = dht_->getId();
+        info.node_id = dht_->getNodeId();
+        info.ipv4 = dht_->getNodesStats(AF_INET);
+        info.ipv6 = dht_->getNodesStats(AF_INET6);
+        if (auto sock = dht_->getSocket()) {
             info.bound4 = sock->getBoundRef(AF_INET).getPort();
             info.bound6 = sock->getBoundRef(AF_INET6).getPort();
         }
@@ -563,6 +528,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();
@@ -578,49 +544,49 @@ std::vector<unsigned>
 DhtRunner::getNodeMessageStats(bool in) const
 {
     std::lock_guard<std::mutex> lck(dht_mtx);
-    return activeDht()->getNodeMessageStats(in);
+    return dht_->getNodeMessageStats(in);
 }
 
 std::string
 DhtRunner::getStorageLog() const
 {
     std::lock_guard<std::mutex> lck(dht_mtx);
-    return activeDht()->getStorageLog();
+    return dht_->getStorageLog();
 }
 std::string
 DhtRunner::getStorageLog(const InfoHash& f) const
 {
     std::lock_guard<std::mutex> lck(dht_mtx);
-    return activeDht()->getStorageLog(f);
+    return dht_->getStorageLog(f);
 }
 std::string
 DhtRunner::getRoutingTablesLog(sa_family_t af) const
 {
     std::lock_guard<std::mutex> lck(dht_mtx);
-    return activeDht()->getRoutingTablesLog(af);
+    return dht_->getRoutingTablesLog(af);
 }
 std::string
 DhtRunner::getSearchesLog(sa_family_t af) const
 {
     std::lock_guard<std::mutex> lck(dht_mtx);
-    return activeDht()->getSearchesLog(af);
+    return dht_->getSearchesLog(af);
 }
 std::string
 DhtRunner::getSearchLog(const InfoHash& f, sa_family_t af) const
 {
     std::lock_guard<std::mutex> lck(dht_mtx);
-    return activeDht()->getSearchLog(f, af);
+    return dht_->getSearchLog(f, af);
 }
 std::vector<SockAddr>
-DhtRunner::getPublicAddress(sa_family_t af)
+DhtRunner::getPublicAddress(sa_family_t af) const
 {
     std::lock_guard<std::mutex> lck(dht_mtx);
-    if (auto dht = activeDht())
-        return dht->getPublicAddress(af);
+    if (dht_)
+        return dht_->getPublicAddress(af);
     return {};
 }
 std::vector<std::string>
-DhtRunner::getPublicAddressStr(sa_family_t af)
+DhtRunner::getPublicAddressStr(sa_family_t af) const
 {
     auto addrs = getPublicAddress(af);
     std::vector<std::string> ret(addrs.size());
@@ -643,15 +609,12 @@ DhtRunner::getPublicAddress(std::function<void(std::vector<SockAddr>&&)> cb, sa_
 void
 DhtRunner::registerCertificate(std::shared_ptr<crypto::Certificate> cert) {
     std::lock_guard<std::mutex> lck(dht_mtx);
-    activeDht()->registerCertificate(cert);
+    dht_->registerCertificate(cert);
 }
+
 void
 DhtRunner::setLocalCertificateStore(CertificateStoreQuery&& query_method) {
     std::lock_guard<std::mutex> lck(dht_mtx);
-#ifdef OPENDHT_PROXY_CLIENT
-    if (dht_via_proxy_)
-        dht_via_proxy_->setLocalCertificateStore(std::forward<CertificateStoreQuery>(query_method));
-#endif
     if (dht_)
         dht_->setLocalCertificateStore(std::forward<CertificateStoreQuery>(query_method));
 }
@@ -659,8 +622,7 @@ DhtRunner::setLocalCertificateStore(CertificateStoreQuery&& query_method) {
 time_point
 DhtRunner::loop_()
 {
-    auto dht = activeDht();
-    if (not dht)
+    if (not dht_)
         return {};
 
     decltype(pending_ops) ops {};
@@ -671,7 +633,7 @@ DhtRunner::loop_()
                std::move(pending_ops) : std::move(pending_ops_prio);
     }
     while (not ops.empty()) {
-        ops.front()(*dht);
+        ops.front()(*dht_);
         ops.pop();
     }
 
@@ -704,13 +666,13 @@ DhtRunner::loop_()
             if (now - pkt.received > net::RX_QUEUE_MAX_DELAY)
                 dropped++;
             else
-                wakeup = dht->periodic(pkt.data.data(), pkt.data.size(), std::move(pkt.from), now);
+                wakeup = dht_->periodic(pkt.data.data(), pkt.data.size(), std::move(pkt.from), now);
             pkt.data.clear();
         }
         received_treated.splice(received_treated.end(), std::move(received));
     } else {
         // Or just run the scheduler
-        wakeup = dht->periodic(nullptr, 0, nullptr, 0, clock::now());
+        wakeup = dht_->periodic(nullptr, 0, nullptr, 0, clock::now());
     }
 
     if (not received_treated.empty()) {
@@ -722,8 +684,8 @@ DhtRunner::loop_()
     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);
+    NodeStatus nstatus4 = dht_->updateStatus(AF_INET);
+    NodeStatus nstatus6 = dht_->updateStatus(AF_INET6);
     if (nstatus4 != status4 || nstatus6 != status6) {
         status4 = nstatus4;
         status6 = nstatus6;
@@ -781,27 +743,7 @@ DhtRunner::listen(InfoHash hash, ValueCallback vcb, Value::Filter f, Where w)
         return ret_token->get_future();
     }
     pending_ops.emplace([=](SecureDht& dht) mutable {
-#ifdef OPENDHT_PROXY_CLIENT
-        auto tokenbGlobal = listener_token_++;
-        auto& listener = listeners_[tokenbGlobal];
-        listener.hash = hash;
-        listener.f = std::move(f);
-        listener.w = std::move(w);
-        listener.gcb = [hash,vcb,tokenbGlobal,this](const std::vector<Sp<Value>>& vals, bool expired) {
-            if (not vcb(vals, expired)) {
-                cancelListen(hash, tokenbGlobal);
-                return false;
-            }
-            return true;
-        };
-        if (auto token = dht.listen(hash, listener.gcb, listener.f, listener.w)) {
-            if (use_proxy)  listener.tokenProxyDht = token;
-            else            listener.tokenClassicDht = token;
-        }
-        ret_token->set_value(tokenbGlobal);
-#else
         ret_token->set_value(dht.listen(hash, std::move(vcb), std::move(f), std::move(w)));
-#endif
     });
     cv.notify_all();
     return ret_token->get_future();
@@ -817,21 +759,13 @@ void
 DhtRunner::cancelListen(InfoHash h, size_t token)
 {
     std::lock_guard<std::mutex> lck(storage_mtx);
-#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);
-    });
-#else
+    if (running != State::Running)
+        return;
+    ongoing_ops++;
     pending_ops.emplace([=](SecureDht& dht) {
         dht.cancelListen(h, token);
+        opEnded();
     });
-#endif // OPENDHT_PROXY_CLIENT
     cv.notify_all();
 }
 
@@ -839,21 +773,13 @@ void
 DhtRunner::cancelListen(InfoHash h, std::shared_future<size_t> ftoken)
 {
     std::lock_guard<std::mutex> lck(storage_mtx);
-#ifdef OPENDHT_PROXY_CLIENT
-    pending_ops.emplace([=](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);
-    });
-#else
-    pending_ops.emplace([=](SecureDht& dht) {
+    if (running != State::Running)
+        return;
+    ongoing_ops++;
+    pending_ops.emplace([this, h, ftoken = std::move(ftoken)](SecureDht& dht) {
         dht.cancelListen(h, ftoken.get());
+        opEnded();
     });
-#endif // OPENDHT_PROXY_CLIENT
     cv.notify_all();
 }
 
@@ -886,7 +812,7 @@ DhtRunner::put(InfoHash hash, std::shared_ptr<Value> value, DoneCallback cb, tim
         return;
     }
     ongoing_ops++;
-    pending_ops.emplace([=, cb = std::move(cb)](SecureDht& dht) mutable {
+    pending_ops.emplace([=, value = std::move(value), cb = std::move(cb)](SecureDht& dht) mutable {
         dht.put(hash, value, bindOpDoneCallback(std::move(cb)), created, permanent);
     });
     cv.notify_all();
@@ -1066,7 +992,7 @@ DhtRunner::bootstrap(std::vector<SockAddr> nodes, DoneCallbackSimple cb)
 }
 
 void
-DhtRunner::bootstrap(const SockAddr& addr, DoneCallbackSimple cb)
+DhtRunner::bootstrap(SockAddr addr, DoneCallbackSimple cb)
 {
     std::unique_lock<std::mutex> lck(storage_mtx);
     if (running != State::Running) {
@@ -1075,7 +1001,7 @@ DhtRunner::bootstrap(const SockAddr& addr, DoneCallbackSimple cb)
         return;
     }
     ongoing_ops++;
-    pending_ops_prio.emplace([addr, cb = bindOpDoneCallback(std::move(cb))](SecureDht& dht) mutable {
+    pending_ops_prio.emplace([addr = std::move(addr), cb = bindOpDoneCallback(std::move(cb))](SecureDht& dht) mutable {
         dht.pingNode(std::move(addr), std::move(cb));
     });
     cv.notify_all();
@@ -1094,12 +1020,12 @@ DhtRunner::bootstrap(const InfoHash& id, const SockAddr& address)
 }
 
 void
-DhtRunner::bootstrap(const std::vector<NodeExport>& nodes)
+DhtRunner::bootstrap(std::vector<NodeExport> nodes)
 {
     std::lock_guard<std::mutex> lck(storage_mtx);
     if (running != State::Running)
         return;
-    pending_ops_prio.emplace([=](SecureDht& dht) {
+    pending_ops_prio.emplace([nodes = std::move(nodes)](SecureDht& dht) {
         for (auto& node : nodes)
             dht.insertNode(node);
     });
@@ -1142,23 +1068,9 @@ void
 DhtRunner::resetDht()
 {
     peerDiscovery_.reset();
-#ifdef OPENDHT_PROXY_CLIENT
-    listeners_.clear();
-    dht_via_proxy_.reset();
-#endif // OPENDHT_PROXY_CLIENT
     dht_.reset();
 }
 
-SecureDht*
-DhtRunner::activeDht() const
-{
-#ifdef OPENDHT_PROXY_CLIENT
-    return use_proxy? dht_via_proxy_.get() : dht_.get();
-#else
-    return dht_.get();
-#endif // OPENDHT_PROXY_CLIENT
-}
-
 void
 DhtRunner::setProxyServer(const std::string& proxy, const std::string& pushNodeId)
 {
@@ -1171,7 +1083,7 @@ DhtRunner::setProxyServer(const std::string& proxy, const std::string& pushNodeI
     enableProxy(use_proxy and not config_.proxy_server.empty());
 #else
     if (not proxy.empty())
-        std::cerr << "DHT proxy requested but OpenDHT built without proxy support." << std::endl;
+        throw std::runtime_error("DHT proxy requested but OpenDHT built without proxy support.");
 #endif
 }
 
@@ -1179,54 +1091,33 @@ void
 DhtRunner::enableProxy(bool proxify)
 {
 #ifdef OPENDHT_PROXY_CLIENT
-    if (dht_via_proxy_) {
-        dht_via_proxy_->shutdown({});
+    if (dht_) {
+        dht_->shutdown({});
     }
     if (proxify) {
         // Init the proxy client
-        auto dht_via_proxy = std::unique_ptr<DhtInterface>(
-            new DhtProxyClient(
+        auto dht_via_proxy = std::make_unique<DhtProxyClient>(
                 config_.server_ca,
                 config_.client_identity,
                 [this]{
                     if (config_.threaded) {
-                        {
-                            std::lock_guard<std::mutex> lck(storage_mtx);
-                            pending_ops_prio.emplace([=](SecureDht&) mutable {});
-                        }
+                        std::lock_guard<std::mutex> lck(storage_mtx);
+                        pending_ops_prio.emplace([=](SecureDht&) mutable {});
                         cv.notify_all();
                     }
                 },
-                config_.proxy_server, config_.push_node_id, logger_)
-        );
-#ifdef OPENDHT_PUSH_NOTIFICATIONS
+                config_.proxy_server, config_.push_node_id, logger_);
         if (not config_.push_token.empty())
             dht_via_proxy->setPushNotificationToken(config_.push_token);
-#endif
-        dht_via_proxy_ = std::unique_ptr<SecureDht>(new SecureDht(std::move(dht_via_proxy), config_.dht_config, identityAnnouncedCb_, logger_));
-        // add current listeners
-        for (auto& l: listeners_)
-            l.second.tokenProxyDht = dht_via_proxy_->listen(l.second.hash, l.second.gcb, l.second.f, l.second.w);
+        dht_ = std::make_unique<SecureDht>(std::move(dht_via_proxy), config_.dht_config, identityAnnouncedCb_, logger_);
         // and use it
         use_proxy = proxify;
     } else {
         use_proxy = proxify;
-        std::lock_guard<std::mutex> lck(storage_mtx);
-        if (not listeners_.empty()) {
-            pending_ops.emplace([this](SecureDht& /*dht*/) mutable {
-                if (not dht_)
-                    return;
-                for (auto& l : listeners_) {
-                    if (not l.second.tokenClassicDht) {
-                        l.second.tokenClassicDht = dht_->listen(l.second.hash, l.second.gcb, l.second.f, l.second.w);
-                    }
-                }
-            });
-        }
     }
 #else
     if (proxify)
-        std::cerr << "DHT proxy requested but OpenDHT built without proxy support." << std::endl;
+        throw std::runtime_error("DHT proxy requested but OpenDHT built without proxy support.");
 #endif
 }
 
@@ -1234,16 +1125,8 @@ void
 DhtRunner::forwardAllMessages(bool forward)
 {
     std::lock_guard<std::mutex> lck(dht_mtx);
-#ifdef OPENDHT_PROXY_SERVER
-#ifdef OPENDHT_PROXY_CLIENT
-    if (dht_via_proxy_)
-        dht_via_proxy_->forwardAllMessages(forward);
-#endif // OPENDHT_PROXY_CLIENT
     if (dht_)
         dht_->forwardAllMessages(forward);
-#else
-    (void) forward;
-#endif // OPENDHT_PROXY_SERVER
 }
 
 /**
@@ -1254,8 +1137,8 @@ DhtRunner::setPushNotificationToken(const std::string& token) {
     std::lock_guard<std::mutex> lck(dht_mtx);
 #if defined(OPENDHT_PROXY_CLIENT) && defined(OPENDHT_PUSH_NOTIFICATIONS)
     config_.push_token = token;
-    if (dht_via_proxy_)
-        dht_via_proxy_->setPushNotificationToken(token);
+    if (dht_)
+        dht_->setPushNotificationToken(token);
 #else
     (void) token;
 #endif
@@ -1265,13 +1148,11 @@ void
 DhtRunner::pushNotificationReceived(const std::map<std::string, std::string>& data)
 {
 #if defined(OPENDHT_PROXY_CLIENT) && defined(OPENDHT_PUSH_NOTIFICATIONS)
-    {
-        std::lock_guard<std::mutex> lck(storage_mtx);
-        pending_ops_prio.emplace([=](SecureDht&) {
-            if (dht_via_proxy_)
-                dht_via_proxy_->pushNotificationReceived(data);
-        });
-    }
+    std::lock_guard<std::mutex> lck(storage_mtx);
+    pending_ops_prio.emplace([=](SecureDht&) {
+        if (dht_)
+            dht_->pushNotificationReceived(data);
+    });
     cv.notify_all();
 #else
     (void) data;

src/network_engine.cpp

@@ -116,7 +116,7 @@ NetworkEngine::~NetworkEngine() {
 }
 
 void
-NetworkEngine::tellListener(Sp<Node> node, Tid socket_id, const InfoHash& hash, want_t want,
+NetworkEngine::tellListener(const Sp<Node>& node, Tid socket_id, const InfoHash& hash, want_t want,
         const Blob& ntoken, std::vector<Sp<Node>>&& nodes,
         std::vector<Sp<Node>>&& nodes6, std::vector<Sp<Value>>&& values,
         const Query& query, int version)
@@ -135,7 +135,7 @@ NetworkEngine::tellListener(Sp<Node> node, Tid socket_id, const InfoHash& hash,
 }
 
 void
-NetworkEngine::tellListenerRefreshed(Sp<Node> n, Tid socket_id, const InfoHash&, const Blob& token, const std::vector<Value::Id>& values, int version)
+NetworkEngine::tellListenerRefreshed(const Sp<Node>& n, Tid socket_id, const InfoHash&, const Blob& token, const std::vector<Value::Id>& values, int version)
 {
     msgpack::sbuffer buffer;
     msgpack::packer<msgpack::sbuffer> pk(&buffer);
@@ -186,7 +186,7 @@ NetworkEngine::tellListenerRefreshed(Sp<Node> n, Tid socket_id, const InfoHash&,
 }
 
 void
-NetworkEngine::tellListenerExpired(Sp<Node> n, Tid socket_id, const InfoHash&, const Blob& token, const std::vector<Value::Id>& values, int version)
+NetworkEngine::tellListenerExpired(const Sp<Node>& n, Tid socket_id, const InfoHash&, const Blob& token, const std::vector<Value::Id>& values, int version)
 {
     msgpack::sbuffer buffer;
     msgpack::packer<msgpack::sbuffer> pk(&buffer);
@@ -415,7 +415,7 @@ NetworkEngine::processMessage(const uint8_t *buf, size_t buflen, SockAddr f)
         return;
     }
 
-    std::unique_ptr<ParsedMessage> msg {new ParsedMessage};
+    auto msg = std::make_unique<ParsedMessage>();
     try {
         msgpack::unpacked msg_res = msgpack::unpack((const char*)buf, buflen);
         msg->msgpack_unpack(msg_res.get());
@@ -699,7 +699,7 @@ NetworkEngine::send(const SockAddr& addr, const char *buf, size_t len, bool conf
 }
 
 Sp<Request>
-NetworkEngine::sendPing(Sp<Node> node, RequestCb&& on_done, RequestExpiredCb&& on_expired) {
+NetworkEngine::sendPing(const Sp<Node>& node, RequestCb&& on_done, RequestExpiredCb&& on_expired) {
     Tid tid (node->getNewTid());
     msgpack::sbuffer buffer;
     msgpack::packer<msgpack::sbuffer> pk(&buffer);
@@ -757,7 +757,7 @@ NetworkEngine::sendPong(const SockAddr& addr, Tid tid) {
 }
 
 Sp<Request>
-NetworkEngine::sendFindNode(Sp<Node> n, const InfoHash& target, want_t want,
+NetworkEngine::sendFindNode(const Sp<Node>& n, const InfoHash& target, want_t want,
         RequestCb&& on_done, RequestExpiredCb&& on_expired) {
     Tid tid (n->getNewTid());
     msgpack::sbuffer buffer;
@@ -802,7 +802,7 @@ NetworkEngine::sendFindNode(Sp<Node> n, const InfoHash& target, want_t want,
 
 
 Sp<Request>
-NetworkEngine::sendGetValues(Sp<Node> n, const InfoHash& info_hash, const Query& query, want_t want,
+NetworkEngine::sendGetValues(const Sp<Node>& n, const InfoHash& info_hash, const Query& query, want_t want,
         RequestCb&& on_done, RequestExpiredCb&& on_expired) {
     Tid tid (n->getNewTid());
     msgpack::sbuffer buffer;
@@ -1073,7 +1073,7 @@ NetworkEngine::bufferNodes(sa_family_t af, const InfoHash& id, want_t want,
 }
 
 Sp<Request>
-NetworkEngine::sendListen(Sp<Node> n,
+NetworkEngine::sendListen(const Sp<Node>& n,
         const InfoHash& hash,
         const Query& query,
         const Blob& token,
@@ -1142,7 +1142,7 @@ NetworkEngine::sendListenConfirmation(const SockAddr& addr, Tid tid) {
 }
 
 Sp<Request>
-NetworkEngine::sendAnnounceValue(Sp<Node> n,
+NetworkEngine::sendAnnounceValue(const Sp<Node>& n,
         const InfoHash& infohash,
         const Sp<Value>& value,
         time_point created,
@@ -1155,11 +1155,12 @@ NetworkEngine::sendAnnounceValue(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));
       }
@@ -1200,7 +1201,7 @@ NetworkEngine::sendAnnounceValue(Sp<Node> n,
 }
 
 Sp<Request>
-NetworkEngine::sendUpdateValues(Sp<Node> n,
+NetworkEngine::sendUpdateValues(const Sp<Node>& n,
                                 const InfoHash& infohash,
                                 const std::vector<Sp<Value>>& values,
                                 time_point created,
@@ -1246,7 +1247,7 @@ NetworkEngine::sendUpdateValues(Sp<Node> n,
 }
 
 Sp<Request>
-NetworkEngine::sendRefreshValue(Sp<Node> n,
+NetworkEngine::sendRefreshValue(const Sp<Node>& n,
                 const InfoHash& infohash,
                 const Value::Id& vid,
                 const Blob& token,

src/op_cache.h

@@ -28,7 +28,7 @@ struct OpCacheValueStorage
     Sp<Value> data {};
     unsigned refCount {1};
     system_clock::time_point updated {system_clock::time_point::min()};
-    explicit OpCacheValueStorage(Sp<Value> val) : data(val) {}
+    explicit OpCacheValueStorage(Sp<Value> val) : data(std::move(val)) {}
 };
 
 class OpValueCache {

src/request.h

@@ -74,14 +74,14 @@ struct Request {
             Blob&& msg,
             std::function<void(const Request&, ParsedMessage&&)> on_done,
             std::function<void(const Request&, bool)> on_expired) :
-        node(node), tid(tid), type(type), on_done(on_done), on_expired(on_expired), msg(std::move(msg)) { }
+        node(std::move(node)), tid(tid), type(type), on_done(std::move(on_done)), on_expired(std::move(on_expired)), msg(std::move(msg)) { }
     Request(MessageType type, Tid tid,
             Sp<Node> node,
             Blob&& msg,
             std::function<void(const Request&, ParsedMessage&&)> on_done,
             std::function<bool(const Request&, DhtProtocolException&&)> on_error,
             std::function<void(const Request&, bool)> on_expired) :
-        node(node), tid(tid), type(type), on_done(on_done), on_error(on_error), on_expired(on_expired), msg(std::move(msg)) { }
+        node(std::move(node)), tid(tid), type(type), on_done(std::move(on_done)), on_error(std::move(on_error)), on_expired(std::move(on_expired)), msg(std::move(msg)) { }
 
     Tid getTid() const { return tid; }
     MessageType getType() const { return type; }

src/securedht.cpp

@@ -71,37 +71,21 @@ SecureDht::~SecureDht() = default;
 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;
-            }
-        }
+    type.storePolicy = [type](InfoHash id, Sp<Value>& v, const InfoHash& nid, const SockAddr& a) {
+        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,14 @@ 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;
+            auto isDecrypted = v->isDecrypted();
+            if (auto decrypted_val = v->decrypt(*key_)) {
+                auto cacheValue = not isDecrypted and decrypted_val->owner;
+                if (cacheValue)
+                    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 +252,12 @@ 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_)
+        auto cacheValue = not v->isSignatureChecked() and enableCache_ and v->owner;
+        if (v->checkSignature()) {
+            if (cacheValue)
                 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

@@ -254,6 +254,62 @@ unpackId(const Json::Value& json, const std::string& key) {
 }
 #endif
 
+void
+Value::sign(const crypto::PrivateKey& key)
+{
+    if (isEncrypted())
+        throw DhtException("Can't sign encrypted data.");
+    owner = key.getSharedPublicKey();
+    signature = key.sign(getToSign());
+}
+
+Value
+Value::encrypt(const crypto::PrivateKey& from, const crypto::PublicKey& to)
+{
+    if (isEncrypted())
+        throw DhtException("Data is already encrypted.");
+    setRecipient(to.getId());
+    sign(from);
+    Value nv {id};
+    nv.setCypher(to.encrypt(getToEncrypt()));
+    return nv;
+}
+
+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);
+            auto msg = msgpack::unpack((const char*)decryptedBlob.data(), decryptedBlob.size());
+            auto v = std::make_unique<Value>(id);
+            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;
+}
+
 bool
 FieldValue::operator==(const FieldValue& vfd) const
 {

src/value_cache.h

@@ -53,7 +53,7 @@ public:
     CallbackQueue clear() {
         std::vector<Sp<Value>> expired_values;
         expired_values.reserve(values.size());
-        for (const auto& v : values)
+        for (auto& v : values)
             expired_values.emplace_back(std::move(v.second.data));
         values.clear();
         CallbackQueue ret;

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));
@@ -199,6 +228,11 @@ DhtRunnerTester::testIdOps() {
     node2.run(42232, config2, std::move(context2));
     node2.bootstrap(node1.getBound());
 
+    {
+        std::unique_lock<std::mutex> lk(mutex);
+        CPPUNIT_ASSERT(cv.wait_for(lk, 20s, [&]{ return valueCount == 1; }));
+    }
+
     node1.findCertificate(node2.getId(), [&](const std::shared_ptr<dht::crypto::Certificate>& crt){
         CPPUNIT_ASSERT(crt);
         std::lock_guard<std::mutex> lk(mutex);

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/Makefile.am

@@ -11,3 +11,10 @@ dhtchat_LDFLAGS = -lopendht -lreadline -L@top_builddir@/src/.libs @GnuTLS_LIBS@
 
 dhtscanner_SOURCES = dhtscanner.cpp
 dhtscanner_LDFLAGS = -lopendht -lreadline -L@top_builddir@/src/.libs @GnuTLS_LIBS@
+
+if ENABLE_C
+bin_PROGRAMS += dhtcnode
+dhtcnode_CFLAGS = -std=c11 -isystem @top_srcdir@/c -isystem @top_srcdir@/include
+dhtcnode_SOURCES = dhtcnode.c
+dhtcnode_LDFLAGS = -lopendht-c -lreadline -L@top_builddir@/c/.libs
+endif

tools/dhtcnode.c

@@ -1,20 +1,53 @@
+/*
+ *  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 <opendht_c.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;
-    int d;
+    atomic_bool stop;
+};
+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;
 }
 
@@ -26,16 +59,30 @@ bool dht_get_callback(const dht_value* value, void* user_data)
     return true;
 }
 
-void dht_done_callback(bool ok, void* user_data)
+void dht_get_done_callback(bool ok, void* user_data)
 {
     dht_runner* runner = (dht_runner*)user_data;
-    printf("Done callback. %s\n", ok ? "Success !" : "Failure :-(");
+    printf("Get completed: %s\n", ok ? "success !" : "failure :-(");
 }
 
-void op_context_free(void* user_data)
+void dht_put_done_callback(bool ok, void* user_data)
 {
+    dht_runner* runner = (dht_runner*)user_data;
+    printf("Put completed: %s\n", ok ? "success !" : "failure :-(");
+}
+
+void dht_shutdown_callback(void* user_data)
+{
+    printf("Stopped.\n");
     struct op_context* ctx = (struct op_context*)user_data;
-    printf("op_context_free %d.\n", ctx->d);
+    atomic_store(&ctx->stop, true);
+}
+
+void listen_context_free(void* user_data)
+{
+    printf("listen_context_free.\n");
+    struct listen_context* ctx = (struct listen_context*)user_data;
+    dht_op_token_delete(ctx->token);
     free(ctx);
 }
 
@@ -60,66 +107,179 @@ char* print_addr(const struct sockaddr* addr) {
     return s;
 }
 
-int main()
+struct dht_params {
+    bool help;
+    bool version;
+    bool generate_identity;
+    bool service;
+    bool peer_discovery;
+    bool log;
+    const char* bootstrap;
+    unsigned network;
+    in_port_t port;
+};
+
+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'},
+    {"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) {
+        case 'p': {
+                int port_arg = atoi(optarg);
+                if (port_arg >= 0 && port_arg < 0x10000)
+                    params.port = port_arg;
+            }
+            break;
+        case 'D':
+            params.peer_discovery = true;
+            break;
+        case 'n':
+            params.network = strtoul(optarg, NULL, 0);
+            break;
+        case 'b':
+            params.bootstrap = (optarg[0] == '=') ? optarg+1 : optarg;
+            break;
+        case 'h':
+            params.help = true;
+            break;
+        case 'v':
+            params.log = true;
+            break;
+        case 'i':
+            params.generate_identity = true;
+            break;
+        case 's':
+            params.service = true;
+            break;
+        case 'V':
+            params.version = true;
+            break;
+        default:
+            break;
+        }
+    }
+    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)
 {
-    dht_identity id = dht_identity_generate("testNode", NULL);
-    dht_infohash cert_id = dht_certificate_get_id(id.certificate);
-    printf("Cert ID: %s\n", dht_infohash_print(&cert_id));
+    struct dht_params params = parse_args(argc, argv);
 
-    dht_publickey* pk = dht_certificate_get_publickey(id.certificate);
-    dht_infohash pk_id = dht_publickey_get_id(pk);
-    printf("PK ID: %s\n", dht_infohash_print(&pk_id));
-    dht_publickey_delete(pk);
-
-    pk = dht_privatekey_get_publickey(id.privatekey);
-    pk_id = dht_publickey_get_id(pk);
-    printf("Key ID: %s\n", dht_infohash_print(&pk_id));
-    dht_publickey_delete(pk);
-
-    dht_identity_delete(&id);
+    if (params.version) {
+        printf("OpenDHT version %s\n", dht_version());
+        return EXIT_SUCCESS;
+    }
 
     dht_runner* runner = dht_runner_new();
-    dht_runner_run(runner, 4040);
+    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_config.dht_config.node_config.network = params.network;
+    dht_config.log = params.log;
+    dht_runner_run_config(runner, params.port, &dht_config);
 
-    dht_infohash h;
-    dht_infohash_random(&h);
-
-    printf("random hash: %s\n", dht_infohash_print(&h));
-
-    // Put data
-    const char* data_str = "yo, this is some data";
-    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);
-
-    // Get data
-    dht_runner_get(runner, &h, dht_get_callback, dht_done_callback, runner);
-
-    // Listen for data
-    struct op_context* ctx = malloc(sizeof(struct op_context));
-    ctx->runner = runner;
-    ctx->d = 42;
-    dht_op_token* token = dht_runner_listen(runner, &h, dht_value_callback, op_context_free, ctx);
-
-    sleep(1);
-
-    dht_runner_bootstrap(runner, "bootstrap.jami.net", NULL);
-
-    sleep(2);
-
-    struct sockaddr** addrs = dht_runner_get_public_address(runner);
-    for (struct sockaddr** addrIt = addrs; *addrIt; addrIt++) {
-        struct sockaddr* addr = *addrIt;
-        char* addr_str = print_addr(addr);
-        free(addr);
-        printf("Found public address: %s\n", addr_str);
-        free(addr_str);
+    if (params.bootstrap) {
+        printf("Bootstrap using %s\n", params.bootstrap);
+        dht_runner_bootstrap(runner, params.bootstrap, NULL);
     }
-    free(addrs);
 
-    dht_runner_cancel_listen(runner, &h, token);
-    dht_op_token_delete(token);
+    char cmd[64];
+    char arg[64];
+    char value[256];
+    dht_infohash key;
+    while (true) {
+        const char* line_read = readline("> ");
+        if (!line_read)
+            break;
+        if (!*line_read)
+            continue;
+        add_history(line_read);
 
+        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")) {
+            struct sockaddr** addrs = dht_runner_get_public_address(runner);
+            if (addrs) {
+                for (struct sockaddr** addrIt = addrs; *addrIt; addrIt++) {
+                    struct sockaddr* addr = *addrIt;
+                    char* addr_str = print_addr(addr);
+                    free(addr);
+                    printf("Found public address: %s\n", addr_str);
+                    free(addr_str);
+                }
+                free(addrs);
+            }
+        }
+        else if (!strcmp(cmd, "ll")) {
+            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));
+        }
+        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")) {
+            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")) {
+            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);
+        }
+    }
+
+    // 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;