Merge pull request #116 from kaldoran/filter_pht_values

pht : Filter of duplicate values

CMakeLists.txt

@@ -1,10 +1,9 @@
 cmake_minimum_required (VERSION 2.8.6)
 project (opendht)
-set (opendht_VERSION_MAJOR 1)
-set (opendht_VERSION_MINOR 2.0)
+set (opendht_VERSION_MAJOR 0)
+set (opendht_VERSION_MINOR 6.1)
 set (opendht_VERSION ${opendht_VERSION_MAJOR}.${opendht_VERSION_MINOR})
 set (PACKAGE_VERSION ${opendht_VERSION})
-set (VERSION "${opendht_VERSION}")
 
 list(APPEND CMAKE_MODULE_PATH "${PROJECT_SOURCE_DIR}/cmake")
 
@@ -13,29 +12,16 @@ set (exec_prefix "\${prefix}")
 set (libdir "\${exec_prefix}/lib")
 set (includedir "\${prefix}/include")
 
+option (OPENDHT_DEBUG "Build with debug flags" OFF)
 option (OPENDHT_STATIC "Build static library" ON)
 option (OPENDHT_SHARED "Build shared library" ON)
-option (OPENDHT_LOG "Build with logs" ON)
-option (OPENDHT_PYTHON "Build Python bindings" OFF)
-option (OPENDHT_TOOLS "Build DHT tools" ON)
 option (OPENDHT_LTO "Build with LTO" OFF)
 
-find_package (GnuTLS 3.3 REQUIRED)
-find_package (Msgpack 1.2 REQUIRED)
-if (OPENDHT_TOOLS)
-    find_package (Readline 6 REQUIRED)
-endif ()
+option (OPENDHT_PYTHON "Build Python bindings" OFF)
+option (OPENDHT_TOOLS "Build DHT tools" ON)
 
-set (CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -pthread -std=c++11 -Wno-return-type -Wall -Wextra -Wnon-virtual-dtor -pedantic-errors")
+set (CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -pthread -std=c++11 -Wno-return-type -Wall -Wextra -Wnon-virtual-dtor")
 set (CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -DMSGPACK_DISABLE_LEGACY_NIL -DMSGPACK_DISABLE_LEGACY_CONVERT")
-if (NOT CMAKE_BUILD_TYPE)
-    set(CMAKE_BUILD_TYPE Release)
-endif ()
-if (OPENDHT_LOG)
-    add_definitions(-DOPENDHT_LOG=true)
-else ()
-    add_definitions(-DOPENDHT_LOG=false)
-endif()
 if (OPENDHT_LTO)
     set (CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -flto")
     if (CMAKE_COMPILER_IS_GNUCC)
@@ -46,6 +32,12 @@ if (OPENDHT_LTO)
     endif ()
 endif ()
 
+find_package (GnuTLS 3.1 REQUIRED)
+find_package (Msgpack 1.2 REQUIRED)
+if (OPENDHT_TOOLS)
+    find_package (Readline 6 REQUIRED)
+endif ()
+
 list (APPEND opendht_SOURCES
     src/utils.cpp
     src/infohash.cpp
@@ -60,18 +52,17 @@ list (APPEND opendht_SOURCES
     src/network_engine.cpp
     src/securedht.cpp
     src/dhtrunner.cpp
+	src/indexation/pht.cpp
     src/argon2/argon2.c
     src/argon2/core.c
     src/argon2/blake2/blake2b.c
     src/argon2/thread.c
     src/argon2/encoding.c
     src/argon2/ref.c
-    src/indexation/pht.cpp
 )
 
 list (APPEND opendht_HEADERS
     include/opendht/utils.h
-    include/opendht/sockaddr.h
     include/opendht/rng.h
     include/opendht/crypto.h
     include/opendht/infohash.h
@@ -84,12 +75,10 @@ list (APPEND opendht_HEADERS
     include/opendht/node_cache.h
     include/opendht/network_engine.h
     include/opendht/scheduler.h
-    include/opendht/rate_limiter.h
     include/opendht/securedht.h
     include/opendht/log.h
-    include/opendht/log_enable.h
     include/opendht.h
-    include/opendht/indexation/pht.h
+	include/opendht/indexation/pht.h
 )
 
 configure_file (
@@ -105,6 +94,12 @@ include_directories (
     ${CMAKE_CURRENT_BINARY_DIR}/include/
 )
 
+if (OPENDHT_DEBUG)
+    set(CMAKE_BUILD_TYPE Debug)
+else ()
+    set(CMAKE_BUILD_TYPE Release)
+endif ()
+
 if (NOT DEFINED CMAKE_INSTALL_LIBDIR)
     set(CMAKE_INSTALL_LIBDIR lib)
 endif ()
@@ -128,7 +123,7 @@ if (OPENDHT_SHARED)
         ${opendht_HEADERS}
     )
     set_target_properties (opendht PROPERTIES IMPORT_SUFFIX "_import.lib")
-    set_target_properties (opendht PROPERTIES SOVERSION ${opendht_VERSION_MAJOR} VERSION ${opendht_VERSION})
+    #set_target_properties (opendht PROPERTIES SOVERSION 1 VERSION 1.0.0)
     if (OPENDHT_LTO)
         target_link_libraries(opendht -flto -fuse-linker-plugin)
     endif ()
@@ -139,7 +134,6 @@ endif ()
 if (OPENDHT_TOOLS)
     add_subdirectory(tools)
 endif ()
-add_subdirectory(doc)
 
 if (OPENDHT_PYTHON)
     add_subdirectory(python)

Makefile.am

@@ -10,13 +10,15 @@ if USE_CYTHON
 SUBDIRS += python
 endif
 
+if HAVE_DOXYGEN
 SUBDIRS += doc
+endif
 
 ACLOCAL_AMFLAGS = -I m4
 
 DOC_FILES = \
 			README.md \
-			COPYING
+			LICENSE
 
 EXTRA_DIST = \
 			$(DOC_FILES)

README.md

@@ -1,19 +1,16 @@
-<img src="https://raw.githubusercontent.com/savoirfairelinux/opendht/master/resources/opendht_logo_512.png" width="100" align="right">
-<br>
-<h1 style="margin-top:10px">
-    <a id="user-content-opendht-" class="anchor" href="/savoirfairelinux/opendht/blob/master/README.md#opendht-" aria-hidden="true"></a>OpenDHT 
-</h1>
-
+OpenDHT
+===
 A lightweight C++11 Distributed Hash Table implementation originally based on https://github.com/jech/dht by Juliusz Chroboczek.
 
  * Light and fast C++11 Kademlia DHT library.
  * Distributed shared key->value data-store
- * Clean and powerful distributed map API with storage of arbitrary binary values of up to 56 KB.
+ * Clean and powerfull distributed map API with storage of arbitrary binary values of up to 56 KB.
  * Optional public key cryptography layer providing data signature and encryption (using GnuTLS).
  * IPv4 and IPv6 support.
  * Python binding.
 
-## Documentation
+Documentation
+-
 See the wiki: <https://github.com/savoirfairelinux/opendht/wiki>
 
 #### How-to build and install
@@ -25,7 +22,8 @@ Build instructions : <https://github.com/savoirfairelinux/opendht/wiki/Build-the
 g++ main.cpp -std=c++11 -lopendht -lgnutls
 ```
 
-## Examples
+Examples
+-
 ### C++ example
 The `tools` directory includes simple example programs :
 * `dhtnode`, a command line tool, mostly used for debuging, allowing to perform operations supported by the library (get, put etc.) with text values.
@@ -87,27 +85,22 @@ for r in results:
     print(r)
 ```
 
-## Dependencies
+Dependencies
+-
 - msgpack-c 1.2+, used for data serialization.
-- GnuTLS 3.3+, used for cryptographic operations.
+- GnuTLS 3.1+, used for cryptographic operations.
 - Nettle 2.4+, a GnuTLS dependency for crypto.
 - Build tested with GCC 4.8+ (GNU/Linux, Android, Windows with MinGW), Clang/LLVM (Linux, OS X).
 - Build tested with Microsoft Visual Studio 2015
 
-## Contact
-
-IRC: join us on Freenode at [`#opendht`](https://webchat.freenode.net/?channels=%23opendht).
-
-## Licence
+Licence
+-
 Copyright (C) 2014-2016 Savoir-faire Linux Inc.
 
-OpenDHT 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.
-
-See COPYING or https://www.gnu.org/licenses/gpl-3.0.en.html for the full GPLv3 licence.
+Licenced under the GNU General Public License version 3 (see COPYING for the full licence).
 
 This project is independent from another project called OpenDHT (Sean Rhea. Ph.D. Thesis, 2005), now extinct.
 
-## Donations
+Donations
+-
 We gratefully accept Bitcoin donations to support OpenDHT development at: `bitcoin:3EykSd1An888efq4Bq3KaV3hJ3JQ4FPnwm`.

configure.ac

@@ -1,43 +1,24 @@
-dnl define macros
-m4_define([opendht_major_version], 1)
-m4_define([opendht_minor_version], 2)
-m4_define([opendht_patch_version], 0)
-m4_define([opendht_version],
-		  [opendht_major_version.opendht_minor_version.opendht_patch_version])
-
-AC_INIT(opendht, [opendht_version])
+AC_INIT(opendht, 0.6.1)
 AC_CONFIG_AUX_DIR(ac)
 AM_INIT_AUTOMAKE([foreign subdir-objects])
 AC_CONFIG_HEADERS([config.h])
 AC_CONFIG_MACRO_DIR([m4])
 AC_CANONICAL_HOST
 
-AC_SUBST(OPENDHT_MAJOR_VERSION, opendht_major_version)
-AC_SUBST(OPENDHT_MINOR_VERSION, opendht_minor_version)
-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} -g -Wno-return-type -Wall -Wextra -Wnon-virtual-dtor -O0"],
+      [CXXFLAGS="${CXXFLAGS} -O3"])
 
 AC_PROG_CXX
 AM_PROG_AR
 
-dnl Check for logs
-AC_ARG_ENABLE([logs], [AS_HELP_STRING([--disable-logs], [Disble 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])
-])
-
 dnl Check for Doxygen
-AC_ARG_ENABLE([doc], AS_HELP_STRING([--enable-doc], [Enable documentation generation (doxygen)]))
-AS_IF([test "x$enable_doc" = "xyes"], [
+AC_ARG_ENABLE([doc], [AS_HELP_STRING([--disable-doc], [Disable documentation generation (doxygen)])])
+AS_IF([test "x$enable_doc" != "xno"], [
        AC_CHECK_PROGS([DOXYGEN], [doxygen])
-       AS_IF([test -z "$DOXYGEN"], [AC_MSG_WARN([Doxygen not found - continuing without Doxygen support])])
-])
+       AS_IF([test -z "$DOXYGEN"],[AC_MSG_WARN([Doxygen not found - continuing without Doxygen support])])
+       ])
 AM_CONDITIONAL([HAVE_DOXYGEN], [test -n "$DOXYGEN"])
 
 dnl Check for Python
@@ -94,9 +75,9 @@ LT_LANG(C++)
 AX_CXX_COMPILE_STDCXX(11,[noext],[mandatory])
 
 PKG_PROG_PKG_CONFIG()
-PKG_CHECK_MODULES([Nettle], [nettle >= 2.4])
-PKG_CHECK_MODULES([GnuTLS], [gnutls >= 3.3])
-PKG_CHECK_MODULES([MsgPack], [msgpack >= 1.2])
+PKG_CHECK_MODULES([nettle], [nettle >= 2.4])
+PKG_CHECK_MODULES([GNUTLS], [gnutls >= 3.1])
+PKG_CHECK_MODULES([msgpack], [msgpack >= 1.2])
 CXXFLAGS="${CXXFLAGS} -DMSGPACK_DISABLE_LEGACY_NIL -DMSGPACK_DISABLE_LEGACY_CONVERT"
 
 AC_ARG_ENABLE([tools], AS_HELP_STRING([--disable-tools],[Disable tools (CLI DHT node)]),,build_tools=yes)
@@ -107,7 +88,7 @@ AM_COND_IF([ENABLE_TOOLS], [
   ])
 ])
 
-AC_CONFIG_FILES([doc/Doxyfile doc/Makefile])
+AM_COND_IF([HAVE_DOXYGEN], [AC_CONFIG_FILES([doc/Doxyfile doc/Makefile])])
 
 dnl Configure setup.py if we build the python module
 AM_COND_IF([USE_CYTHON], [

doc/CMakeLists.txt

@@ -1,3 +0,0 @@
-if (OPENDHT_TOOLS)
-	INSTALL(FILES ${CMAKE_CURRENT_SOURCE_DIR}/dhtnode.1 DESTINATION ${CMAKE_INSTALL_PREFIX}/share/man/man1)
-endif ()

doc/Makefile.am

@@ -1,6 +1,3 @@
-dist_man1_MANS = dhtnode.1
-
-if HAVE_DOXYGEN
 doxyfile.stamp:
 	$(DOXYGEN) Doxyfile
 	echo stamp > doxyfile.stamp
@@ -10,4 +7,3 @@ CLEANFILES = doxyfile.stamp
 all-local: doxyfile.stamp
 clean-local:
 	rm -rf $(top_srcdir)/doc/man $(top_srcdir)/doc/html
-endif

doc/connectivity-loss-by-search-criteria.tex

@@ -1,142 +0,0 @@
-\documentclass[11pt]{article}
-
-\usepackage[utf8x]{inputenc}
-\usepackage[top=2cm,bottom=2cm]{geometry}
-
-\usepackage{forloop}
-\newcounter{counter}
-
-% maths
-\usepackage{amssymb}
-\usepackage{amsmath}
-\usepackage{mathrsfs}
-\usepackage{shadethm}
-\usepackage{amsthm}
-\newshadetheorem{shadeDef}{Definition}[section]
-\newtheorem{remark}{Remark}[section]
-\renewcommand{\P}{\mathbb{P}}
-
-\usepackage{float}
-\usepackage{booktabs}
-
-\setlength{\parindent}{0ex}
-\setlength{\parskip}{0.5em}
-
-\begin{document}
-    \title{Annex 1: Probabilistic analysis of connectivity changes}
-    \author{Adrien Béraud, Simon Désaulniers, Guillaume Roguez}
-    \maketitle
-    \pagestyle{empty}
-    \begin{shadeDef}
-        A node flagged as \emph{``expired''} by a node $n$ is a node which has not responded to any
-        of $n$'s last three requests.
-    \end{shadeDef}
-    \begin{remark}
-        An expired node will not be contacted before 10 minutes from its expiration time.
-    \end{remark}
-
-    Let $N$ the DHT network, $n_0\in N$, a given node and the following probabilistic events:
-    \begin{itemize}
-        \item $A$: $\forall n \in N$ $n$ is unreachable by $n_0$, \emph{i.e.} $n_0$ lost connection
-            with $N$;
-        \item $B$: $S\subset N$, the nodes unreachable by $n_0$ with $k={|S|\over|N|}$;
-        \item $C$: $m \le |N|$ nodes are flagged as ``expired''.
-    \end{itemize}
-
-    We are interested in knowing $\P(A|C)$, \emph{i.e.} the probability of the event where $A$ occurs
-    prior to $C$. From the above, we immediately get
-    $$\left\{
-        \begin{array}{ll}
-            \P(C|A)       & = 1\\
-            \P(A) + \P(B) & = 1
-        \end{array}
-    \right.$$
-    Also, the event $A|C$ can be abstracted as the urn problem of draw without replacement. Then,
-        $$\P(C|B) = \prod_{i=0}^m \left[k|N| - i \over |N|\right] = \prod_{i=0}^m \left[k - { i \over |N| }\right]$$
-    Furthermore, using Bayes' theroem we have
-    \begin{align*}
-        \P(A|C) & = { \P(C|A)\P(A) \over \P(C|A)\P(A) + \P(C|B)\P(B)}\\
-                & = { \P(A) \over \P(A) + \P(C|B)\P(B) }\\
-                & = { \P(A) \over \P(A) + \P(C|B)\left[1 - \P(A)\right] }\\
-        \Rightarrow \forloop{counter}{0}{\value{counter} < 5}{\qquad}\quad  \P(A)  & =
-            \P(A|C)\left[\P(A) + \P(C|B)\left(1 - \P(A)\right) \right] \\
-        \Rightarrow \forloop{counter}{0}{\value{counter} < 2}{\qquad}\;\:\, \P(A)\left[{ 1 \over \P(A|C)} - 1\right] & =
-            \P(C|B)\left(1 - \P(A)\right) \\
-    \end{align*}
-    Finally,
-    \begin{equation}
-        \label{eq:final}
-        \left[{\P(A) \over 1 - \P(A)}\right]\left[{ 1 \over \P(A|C)} - 1\right] =
-            \prod_{i=0}^m \left[k - { i \over |N| }\right]
-    \end{equation}
-
-    From \eqref{eq:final}, we may set a plausible configuration $\{\P(A),\P(A|C),k,|N|\}$ letting us
-    produce results such as in table \ref{tbl:k_1_2}, \ref{tbl:k_2_3} and \ref{tbl:k_3_4}.
-
-    \begin{table}[H]
-        \centering
-        \caption{The values for $m$ assuming $\P(A|C) \ge 0.95,\, k = {1 \over 2}$}
-        \label{tbl:k_1_2}
-        \begin{tabular}{lcccc}
-            \toprule
-            $|N| \diagdown \P(A)$ & ${1 \over 10 }$ & ${1 \over 100 }$ & ${1 \over 1000 }$ & ${1 \over 10000 }$\\
-            \midrule
-            $2^0$               & 1               & 1                & 1                 & 1\\
-            $2^1$               & 1               & 1                & 1                 & 1\\
-            $2^2$               & 2               & 2                & 2                 & 2\\
-            $2^3$               & 4               & 4                & 4                 & 4\\
-            $2^4$               & 5               & 6                & 7                 & 8\\
-            $2^5$               & 5               & 7                & 9                 & 10\\
-            $2^6$               & 6               & 9                & 11                & 13\\
-            $2^7$               & 6               & 9                & 12                & 14\\
-            $2^8$               & 7               & 10               & 13                & 16\\
-            $2^9$               & 7               & 10               & 13                & 16\\
-            $2^{10}$            & 7               & 10               & 13                & 17\\
-            \bottomrule
-        \end{tabular}
-    \end{table}
-    \begin{table}[H]
-        \centering
-        \caption{The values for $m$ assuming $\P(A|C) \ge 0.95,\, k = {2 \over 3}$}
-        \label{tbl:k_2_3}
-        \begin{tabular}{lcccc}
-            \toprule
-            $|N| \diagdown \P(A)$ & ${1 \over 10 }$ & ${1 \over 100 }$ & ${1 \over 1000 }$ & ${1 \over 10000 }$\\
-            \midrule
-            $2^0$               & 1               & 1                & 1                 & 1\\
-            $2^1$               & 2               & 2                & 2                 & 2\\
-            $2^2$               & 3               & 3                & 4                 & 4\\
-            $2^3$               & 5               & 5                & 6                 & 8\\
-            $2^4$               & 6               & 8                & 9                 & 10\\
-            $2^5$               & 8               & 10               & 12                & 14\\
-            $2^6$               & 9               & 13               & 16                & 18\\
-            $2^7$               & 11              & 15               & 18                & 22\\
-            $2^8$               & 11              & 16               & 21                & 25\\
-            $2^9$               & 12              & 17               & 22                & 27\\
-            $2^{10}$            & 12              & 18               & 23                & 28\\
-            \bottomrule
-        \end{tabular}
-    \end{table}
-    \begin{table}[H]
-        \centering
-        \caption{The values for $m$ assuming $\P(A|C) \ge 0.95,\, k = {3 \over 4}$}
-        \label{tbl:k_3_4}
-        \begin{tabular}{lcccc}
-            \toprule
-            $|N| \diagdown \P(A)$ & ${1 \over 10 }$ & ${1 \over 100 }$ & ${1 \over 1000 }$ & ${1 \over 10000 }$\\
-            \midrule
-            $2^0$               & 1               & 1                & 1                 & 1\\
-            $2^1$               & 2               & 2                & 2                 & 2\\
-            $2^2$               & 3               & 3                & 3                 & 3\\
-            $2^3$               & 5               & 6                & 6                 & 6\\
-            $2^4$               & 7               & 9                & 10                & 11\\
-            $2^5$               & 10              & 12               & 14                & 16\\
-            $2^6$               & 12              & 16               & 19                & 22\\
-            $2^7$               & 14              & 19               & 23                & 27\\
-            $2^8$               & 15              & 21               & 27                & 32\\
-            $2^9$               & 16              & 23               & 30                & 36\\
-            $2^{10}$            & 17              & 24               & 31                & 38\\
-            \bottomrule
-        \end{tabular}
-    \end{table}
-\end{document}

doc/dhtnode.1

@@ -1,87 +0,0 @@
-.TH DHTNODE 1 2016-07-29
-
-.SH NAME
-.B dhtnode
-- a simple OpenDHT command line node runner.
-
-.SH SYNOPSIS
-.B dhtnode [-h]
-
-.B dhtnode [-v [-l \fIlogfile\fP]] [-i] [-d] [-n \fInetwork_id\fP] [-p \fIlocal_port\fP] [-b \fIbootstrap_host\fP[:\fIport\fP]]
-
-.SH DESCRIPTION
-
-This program runs a simple OpenDHT node in an interactive way. If you rather
-want to run the node in daemon mode, option \fB'-d'\fP is provided. When running
-in the interactive shell, you benefit from the readline capabilities for writing
-your commands such as command history. Here are the available commands in the
-interactive shell:
-
-.EE
-    h, help    Print this help message.
-    q, quit    Quit the program.
-    log        Start/stop printing DHT logs.
-
-    Node information:
-    ll         Print basic information and stats about the current node.
-    ls         Print basic information about current searches.
-    ld         Print basic information about currenty stored values on this node.
-    lr         Print the full current routing table of this node
-
-    Operations on the DHT:
-    b ip:port             Ping potential node at given IP address/port.
-    g [key]               Get values at [key].
-    l [key]               Listen for value changes at [key].
-    p [key] [str]         Put string value at [key].
-    s [key] [str]         Put string value at [key], signed with our generated
-                          private key.
-    e [key] [dest] [str]  Put string value at [key], encrypted for [dest] with
-                          its public key (if found).
-
-.SH OPTIONS
-
-.TP
-\fB-h\fP
-Prints some help.
-
-.TP
-\fB-v\fP
-Enable the verbose mode (log to stdout by default)
-
-.TP
-\fB-l\fP \fIlog_file\fP
-Write log to file instead of stdout
-
-.TP
-\fB-i\fP
-Generate cryptographic identity for the node.
-
-.TP
-\fB-d\fP
-Run the program in daemon mode (will fork in the background).
-
-.TP
-\fB-n\fP \fInetwork_id\fP
-Specify the network id. This let you connect to distinct networks and prevents
-the merge of two different networks (available since OpenDHT v0.6.1).
-
-.TP
-\fB-p\fP \fIlocal_port\fP
-Use port \fIlocal_port\fP for the program to bind to.
-
-.TP
-\fB-b\fP \fIbootstrap_host\fP[:\fIport\fP]
-The program needs to be given a node to connect to the network. You use this
-option to provide the ip address of that node.
-
-.SH AUTHORS
-.TP
-Program written by
-.IP \(bu
-.\}
-Adrien Béraud <adrien.beraud@savoirfairelinux.com>
-.TP
-Man page written by
-.IP \(bu
-.\}
-Simon Désaulniers <sim.desaulniers@gmail.com>

include/opendht/callbacks.h

@@ -30,47 +30,25 @@ namespace dht {
 
 struct Node;
 
-/**
- * Current status of a DHT node.
- */
 enum class NodeStatus {
     Disconnected, // 0 nodes
     Connecting,   // 1+ nodes
     Connected     // 1+ good nodes
 };
 
-/**
- * Dht configuration.
- */
 struct Config {
-    /** DHT node ID */
     InfoHash node_id;
-
-    /** 
-     * DHT network ID. A node will only talk with other nodes having
-     * the same network ID.
-     * Network ID 0 (default) represents the main public network.
-     */
-    NetId network;
-
-    /** For testing purposes only, enables bootstrap mode */
     bool is_bootstrap;
 };
 
-/**
- * SecureDht configuration.
- */
 struct SecureDhtConfig
 {
     Config node_config;
     crypto::Identity id;
 };
 
-static constexpr size_t DEFAULT_STORAGE_LIMIT {1024 * 1024 * 64};
-
 using ValuesExport = std::pair<InfoHash, Blob>;
 
-using QueryCallback = std::function<bool(const std::vector<std::shared_ptr<FieldValueIndex>>& fields)>;
 using GetCallback = std::function<bool(const std::vector<std::shared_ptr<Value>>& values)>;
 using GetCallbackSimple = std::function<bool(std::shared_ptr<Value> value)>;
 using ShutdownCallback = std::function<void()>;
@@ -79,6 +57,8 @@ using CertificateStoreQuery = std::function<std::vector<std::shared_ptr<crypto::
 
 typedef bool (*GetCallbackRaw)(std::shared_ptr<Value>, void *user_data);
 
+static constexpr size_t DEFAULT_STORAGE_LIMIT {1024 * 1024 * 64};
+
 GetCallbackSimple bindGetCb(GetCallbackRaw raw_cb, void* user_data);
 GetCallback bindGetCb(GetCallbackSimple cb);
 

include/opendht/crypto.h

@@ -329,14 +329,9 @@ Identity generateIdentity(const std::string& name = "dhtnode", Identity ca = {},
 
 
 /**
- * Performs SHA512, SHA256 or SHA1, depending on hash_length.
- * Attempts to choose an hash function with
- * output size of at least hash_length bytes, Current implementation
- * will use SHA1 for hash_length up to 20 bytes,
- * will use SHA256 for hash_length up to 32 bytes,
- * will use SHA512 for hash_length of 33 bytes and more.
+ * SHA512
  */
-Blob hash(const Blob& data, size_t hash_length = 512/8);
+Blob hash(const Blob& data);
 
 /**
  * Generates an encryption key from a text password,
@@ -345,7 +340,7 @@ Blob hash(const Blob& data, size_t hash_length = 512/8);
  * that can be transmitted in clear, and will be generated if
  * not provided (32 bytes).
  */
-Blob stretchKey(const std::string& password, Blob& salt, size_t key_length = 512/8);
+Blob stretchKey(const std::string& password, Blob& salt);
 
 /**
  * AES-GCM encryption. Key must be 128, 192 or 256 bits long (16, 24 or 32 bytes).

include/opendht/default_types.h

@@ -21,8 +21,6 @@
 
 #include "value.h"
 
-MSGPACK_ADD_ENUM(dht::Value::Field)
-
 namespace dht {
 enum class ImStatus : uint8_t {
     NONE = 0,
@@ -31,7 +29,7 @@ enum class ImStatus : uint8_t {
     READ
 };
 }
-MSGPACK_ADD_ENUM(dht::ImStatus)
+MSGPACK_ADD_ENUM(dht::ImStatus);
 
 namespace dht {
 
@@ -57,7 +55,7 @@ public:
 
     std::string service;
     Blob data;
-    MSGPACK_DEFINE(service, data)
+    MSGPACK_DEFINE(service, data);
 };
 
 template <typename T>
@@ -128,7 +126,7 @@ public:
     long date {0};
     ImStatus status {ImStatus::NONE};
 
-    MSGPACK_DEFINE_MAP(id, msg, date, status)
+    MSGPACK_DEFINE_MAP(id, msg, date, status);
 };
 
 class TrustRequest : public EncryptedValue<TrustRequest>
@@ -149,7 +147,7 @@ public:
 
     std::string service;
     Blob payload;
-    MSGPACK_DEFINE(service, payload)
+    MSGPACK_DEFINE(service, payload);
 };
 
 class IceCandidates : public EncryptedValue<IceCandidates>
@@ -177,7 +175,7 @@ public:
         pk.pack_bin_body((const char*)ice_data.data(), ice_data.size());
 #else
         // hack for backward compatibility with old opendht compiled with msgpack 1.0
-        // remove when enough people have moved to new versions
+        // remove when enough people have moved to new versions 
         pk.pack_array(ice_data.size());
         for (uint8_t b : ice_data)
             pk.pack(b);

include/opendht/dht.h

@@ -14,9 +14,11 @@
  *  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 <http://www.gnu.org/licenses/>.
+ *  along with this program; if not, write to the Free Software
+ *  Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301 USA.
  */
 
+
 #pragma once
 
 #include "infohash.h"
@@ -26,13 +28,15 @@
 #include "scheduler.h"
 #include "routing_table.h"
 #include "callbacks.h"
-#include "log_enable.h"
 
 #include <string>
 #include <array>
 #include <vector>
 #include <map>
+#include <list>
+#include <queue>
 #include <functional>
+#include <algorithm>
 #include <memory>
 
 #ifdef _WIN32
@@ -100,13 +104,6 @@ public:
      */
     void setLoggers(LogMethod error = NOLOG, LogMethod warn = NOLOG, LogMethod debug = NOLOG);
 
-    /**
-     * Only print logs related to the given InfoHash (if given), or disable filter (if zeroes).
-     */
-    void setLogFilter(const InfoHash& f) {
-        DHT_LOG.setFilter(f);
-    }
-
     virtual void registerType(const ValueType& type) {
         types[type.id] = type;
     }
@@ -120,20 +117,14 @@ public:
      * The node is not pinged, so this should be
      * used to bootstrap efficiently from previously known nodes.
      */
-    void insertNode(const InfoHash& id, const SockAddr&);
-    void insertNode(const InfoHash& id, const sockaddr* sa, socklen_t salen) {
-        insertNode(id, SockAddr(sa, salen));
-    }
-    void insertNode(const NodeExport& n) {
-        insertNode(n.id, SockAddr(n.ss, n.sslen));
+    bool insertNode(const InfoHash& id, const sockaddr*, socklen_t);
+    bool insertNode(const NodeExport& n) {
+        return insertNode(n.id, reinterpret_cast<const sockaddr*>(&n.ss), n.sslen);
     }
 
-    void pingNode(const sockaddr*, socklen_t, DoneCallbackSimple&& cb={});
+    int pingNode(const sockaddr*, socklen_t);
 
-    time_point periodic(const uint8_t *buf, size_t buflen, const SockAddr&);
-    time_point periodic(const uint8_t *buf, size_t buflen, const sockaddr* from, socklen_t fromlen) {
-        return periodic(buf, buflen, SockAddr(from, fromlen));
-    }
+    time_point periodic(const uint8_t *buf, size_t buflen, const sockaddr *from, socklen_t fromlen);
 
     /**
      * Get a value by searching on all available protocols (IPv4, IPv6),
@@ -142,32 +133,18 @@ public:
      * @param cb a function called when new values are found on the network.
      *           It should return false to stop the operation.
      * @param donecb a function called when the operation is complete.
-                     cb and donecb won't be called again afterward.
+                        cb and donecb won't be called again afterward.
      * @param f a filter function used to prefilter values.
      */
-    virtual void get(const InfoHash& key, GetCallback cb, DoneCallback donecb={}, Value::Filter&& f={}, Where&& w = {});
-    virtual void get(const InfoHash& key, GetCallback cb, DoneCallbackSimple donecb={}, Value::Filter&& f={}, Where&& w = {}) {
-        get(key, cb, bindDoneCb(donecb), std::forward<Value::Filter>(f), std::forward<Where>(w));
+    virtual void get(const InfoHash& key, GetCallback cb, DoneCallback donecb={}, Value::Filter&& f={});
+    virtual void get(const InfoHash& key, GetCallback cb, DoneCallbackSimple donecb={}, Value::Filter&& f={}) {
+        get(key, cb, bindDoneCb(donecb), std::forward<Value::Filter>(f));
     }
-    virtual void get(const InfoHash& key, GetCallbackSimple cb, DoneCallback donecb={}, Value::Filter&& f={}, Where&& w = {}) {
-        get(key, bindGetCb(cb), donecb, std::forward<Value::Filter>(f), std::forward<Where>(w));
+    virtual void get(const InfoHash& key, GetCallbackSimple cb, DoneCallback donecb={}, Value::Filter&& f={}) {
+        get(key, bindGetCb(cb), donecb, std::forward<Value::Filter>(f));
     }
-    virtual void get(const InfoHash& key, GetCallbackSimple cb, DoneCallbackSimple donecb, Value::Filter&& f={}, Where&& w = {}) {
-        get(key, bindGetCb(cb), bindDoneCb(donecb), std::forward<Value::Filter>(f), std::forward<Where>(w));
-    }
-    /**
-     * Similar to Dht::get, but sends a Query to filter data remotely.
-     * @param key the key for which to query data for.
-     * @param cb a function called when new values are found on the network.
-     *           It should return false to stop the operation.
-     * @param donecb a function called when the operation is complete.
-                     cb and donecb won't be called again afterward.
-     * @param q a query used to filter values on the remotes before they send a
-     *          response.
-     */
-    virtual void query(const InfoHash& key, QueryCallback cb, DoneCallback done_cb = {}, Query&& q = {});
-    virtual void query(const InfoHash& key, QueryCallback cb, DoneCallbackSimple done_cb = {}, Query&& q = {}) {
-        query(key, cb, bindDoneCb(done_cb), std::forward<Query>(q));
+    virtual void get(const InfoHash& key, GetCallbackSimple cb, DoneCallbackSimple donecb, Value::Filter&& f={}) {
+        get(key, bindGetCb(cb), bindDoneCb(donecb), std::forward<Value::Filter>(f));
     }
 
     /**
@@ -181,10 +158,14 @@ public:
     std::shared_ptr<Value> getLocalById(const InfoHash& key, Value::Id vid) const;
 
     /**
-     * Announce a value on all available protocols (IPv4, IPv6).
-     *
+     * Announce a value on all available protocols (IPv4, IPv6), and
+     * automatically re-announce when it's about to expire.
      * The operation will start as soon as the node is connected to the network.
      * The done callback will be called once, when the first announce succeeds, or fails.
+     *
+     * A "put" operation will never end by itself because the value will need to be
+     * reannounced on a regular basis.
+     * User can call #cancelPut(InfoHash, Value::Id) to cancel a put operation.
      */
     void put(const InfoHash& key,
             std::shared_ptr<Value>,
@@ -240,9 +221,9 @@ public:
      *
      * @return a token to cancel the listener later.
      */
-    virtual size_t listen(const InfoHash&, GetCallback, Value::Filter&&={}, Where&& w = {});
-    virtual size_t listen(const InfoHash& key, GetCallbackSimple cb, Value::Filter f={}, Where w = {}) {
-        return listen(key, bindGetCb(cb), std::forward<Value::Filter>(f), std::forward<Where>(w));
+    virtual size_t listen(const InfoHash&, GetCallback, Value::Filter&&={});
+    virtual size_t listen(const InfoHash& key, GetCallbackSimple cb, Value::Filter f={}) {
+        return listen(key, bindGetCb(cb), std::forward<Value::Filter>(f));
     }
 
     virtual bool cancelListen(const InfoHash&, size_t token);
@@ -252,12 +233,7 @@ public:
      * This will cause the DHT to assume a public IP address change.
      * The DHT will recontact neighbor nodes, re-register for listen ops etc.
      */
-    void connectivityChanged(sa_family_t);
-    void connectivityChanged() {
-        reported_addr.clear();
-        connectivityChanged(AF_INET);
-        connectivityChanged(AF_INET6);
-    }
+    void connectivityChanged();
 
     /**
      * Get the list of good nodes for local storage saving purposes
@@ -268,11 +244,9 @@ public:
     std::vector<ValuesExport> exportValues() const;
     void importValues(const std::vector<ValuesExport>&);
 
-    unsigned getNodesStats(sa_family_t af, unsigned *good_return, unsigned *dubious_return, unsigned *cached_return,
+    int getNodesStats(sa_family_t af, unsigned *good_return, unsigned *dubious_return, unsigned *cached_return,
             unsigned *incoming_return) const;
     std::string getStorageLog() const;
-    std::string getStorageLog(const InfoHash&) const;
-
     std::string getRoutingTablesLog(sa_family_t) const;
     std::string getSearchesLog(sa_family_t) const;
 
@@ -296,12 +270,10 @@ public:
         return {total_store_size, total_values};
     }
 
-    std::vector<SockAddr> getPublicAddress(sa_family_t family = 0);
+    std::vector<Address> getPublicAddress(sa_family_t family = 0);
 
 protected:
     Logger DHT_LOG;
-    bool logFilerEnable_ {};
-    InfoHash logFiler_ {};
 
 private:
 
@@ -310,18 +282,8 @@ private:
        the target 8 turn out to be dead. */
     static constexpr unsigned SEARCH_NODES {14};
 
-    /* The number of bad nodes is limited in order to help determine
-     * presence of connectivity changes. See
-     * https://github.com/savoirfairelinux/opendht/issues/137 for details.
-     *
-     * According to the tables, 25 is a good average value for big networks. If
-     * the network is small, normal search expiration process will handle the
-     * situation.
-     * */
-    static constexpr unsigned SEARCH_MAX_BAD_NODES {25};
-
-    /* Concurrent search nodes requested count */
-    static constexpr unsigned MAX_REQUESTED_SEARCH_NODES {4};
+    /* Concurrent requests during a search */
+    static constexpr unsigned SEARCH_REQUESTS {4};
 
     /* Number of listening nodes */
     static constexpr unsigned LISTEN_NODES {4};
@@ -347,14 +309,65 @@ private:
 
     static constexpr size_t TOKEN_SIZE {64};
 
-    // internal structures
     struct SearchNode;
-    struct Get;
-    struct Announce;
-    struct LocalListener;
+
+    /**
+     * A single "get" operation data
+     */
+    struct Get {
+        time_point start;
+        Value::Filter filter;
+        GetCallback get_cb;
+        DoneCallback done_cb;
+    };
+
+    /**
+     * A single "put" operation data
+     */
+    struct Announce {
+        bool permanent;
+        std::shared_ptr<Value> value;
+        time_point created;
+        DoneCallback callback;
+    };
+
+    /**
+     * A single "listen" operation data
+     */
+    struct LocalListener {
+        Value::Filter filter;
+        GetCallback get_cb;
+    };
+
+    /**
+     * A search is a list of the nodes we think are responsible
+     * for storing values for a given hash.
+     */
     struct Search;
-    struct ValueStorage;
-    struct Listener;
+
+    struct ValueStorage {
+        std::shared_ptr<Value> data {};
+        time_point time {};
+
+        ValueStorage() {}
+        ValueStorage(const std::shared_ptr<Value>& v, time_point t) : data(v), time(t) {}
+    };
+
+    /**
+     * Foreign nodes asking for updates about an InfoHash.
+     */
+    struct Listener {
+        size_t rid {};
+        time_point time {};
+
+        /*constexpr*/ Listener(size_t rid, time_point t) : rid(rid), time(t) {}
+
+        void refresh(size_t tid, time_point t) {
+            rid = tid;
+            time = t;
+        }
+    };
+
     struct Storage;
 
     // prevent copy
@@ -376,10 +389,10 @@ private:
     const bool is_bootstrap {false};
 
     // the stuff
-    RoutingTable buckets4 {};
+    RoutingTable buckets {};
     RoutingTable buckets6 {};
 
-    std::map<InfoHash, Storage> store;
+    std::vector<Storage> store;
     size_t total_values {0};
     size_t total_store_size {0};
     size_t max_store_size {DEFAULT_STORAGE_LIMIT};
@@ -394,16 +407,13 @@ private:
     size_t listener_token {1};
 
     // timing
-    Scheduler scheduler;
+    Scheduler scheduler {};
     std::shared_ptr<Scheduler::Job> nextNodesConfirmation {};
-    std::shared_ptr<Scheduler::Job> nextStorageMaintenance {};
     time_point mybucket_grow_time {time_point::min()}, mybucket6_grow_time {time_point::min()};
 
     NetworkEngine network_engine;
-    unsigned pending_pings4 {0};
-    unsigned pending_pings6 {0};
 
-    using ReportedAddr = std::pair<unsigned, SockAddr>;
+    using ReportedAddr = std::pair<unsigned, Address>;
     std::vector<ReportedAddr> reported_addr;
 
     void rotateSecrets();
@@ -411,14 +421,16 @@ private:
     Blob makeToken(const sockaddr *sa, bool old) const;
     bool tokenMatch(const Blob& token, const sockaddr *sa) const;
 
-    void reportedAddr(const SockAddr&);
+    void reportedAddr(const sockaddr *sa, socklen_t sa_len);
 
     // Storage
-    void storageAddListener(const InfoHash& id, const std::shared_ptr<Node>& node, size_t tid, Query&& = {});
+    decltype(Dht::store)::iterator findStorage(const InfoHash& id);
+    decltype(Dht::store)::const_iterator findStorage(const InfoHash& id) const;
+
+    void storageAddListener(const InfoHash& id, const std::shared_ptr<Node>& node, size_t tid);
     bool storageStore(const InfoHash& id, const std::shared_ptr<Value>& value, time_point created);
     void expireStorage();
-    void storageChanged(const InfoHash& id, Storage& st, ValueStorage&);
-    std::string printStorageLog(const decltype(store)::value_type&) const;
+    void storageChanged(Storage& st, ValueStorage&);
 
     /**
      * For a given storage, if values don't belong there anymore because this
@@ -429,14 +441,12 @@ private:
     size_t maintainStorage(InfoHash id, bool force=false, DoneCallback donecb=nullptr);
 
     // Buckets
-    RoutingTable& buckets(sa_family_t af) { return af == AF_INET ? buckets4 : buckets6; }
-    const RoutingTable& buckets(sa_family_t af) const { return af == AF_INET ? buckets4 : buckets6; }
     Bucket* findBucket(const InfoHash& id, sa_family_t af) {
         RoutingTable::iterator b;
         switch (af) {
         case AF_INET:
-            b = buckets4.findBucket(id);
-            return b == buckets4.end() ? nullptr : &(*b);
+            b = buckets.findBucket(id);
+            return b == buckets.end() ? nullptr : &(*b);
         case AF_INET6:
             b = buckets6.findBucket(id);
             return b == buckets6.end() ? nullptr : &(*b);
@@ -462,126 +472,50 @@ private:
     // Searches
 
     /**
-     * Low-level method that will perform a search on the DHT for the specified
-     * infohash (id), using the specified IP version (IPv4 or IPv6).
+     * Low-level method that will perform a search on the DHT for the
+     * specified infohash (id), using the specified IP version (IPv4 or IPv6).
+     * The values can be filtered by an arbitrary provided filter.
      */
-    std::shared_ptr<Search> search(const InfoHash& id, sa_family_t af, GetCallback = {}, QueryCallback = {}, DoneCallback = {}, Value::Filter = {}, Query q = {});
-
+    std::shared_ptr<Search> search(const InfoHash& id, sa_family_t af, GetCallback = nullptr, DoneCallback = nullptr, Value::Filter = Value::AllFilter());
     void announce(const InfoHash& id, sa_family_t af, std::shared_ptr<Value> value, DoneCallback callback, time_point created=time_point::max(), bool permanent = false);
-    size_t listenTo(const InfoHash& id, sa_family_t af, GetCallback cb, Value::Filter f = Value::AllFilter(), const std::shared_ptr<Query>& q = {});
-
-    /**
-     * Refill the search with good nodes if possible.
-     *
-     * @param sr  The search to refill.
-     *
-     * @return the number inserted nodes.
-     */
-    unsigned refill(Search& sr);
+    size_t listenTo(const InfoHash& id, sa_family_t af, GetCallback cb, Value::Filter f = Value::AllFilter());
 
+    void bootstrapSearch(Search& sr);
     Search *findSearch(unsigned short tid, sa_family_t af);
     void expireSearches();
 
     void confirmNodes();
     void expire();
 
-    /**
-     * Generic function to execute when a 'get' request has completed.
-     *
-     * @param status  The request passed by the network engine.
-     * @param answer  The answer from the network engine.
-     * @param ws      A weak pointer to the search concerned by the request.
-     * @param query   The query sent to the node.
-     */
-    void searchNodeGetDone(const Request& status,
-            NetworkEngine::RequestAnswer&& answer,
-            std::weak_ptr<Search> ws,
-            std::shared_ptr<Query> query);
-
-    /**
-     * Generic function to execute when a 'get' request expires.
-     *
-     * @param status  The request passed by the network engine.
-     * @param over    Whether we're done to try sending the request to the node
-     *                or not. This lets us mark a node as candidate.
-     * @param ws      A weak pointer to the search concerned by the request.
-     * @param query   The query sent to the node.
-     */
-    void searchNodeGetExpired(const Request& status, bool over, std::weak_ptr<Search> ws, std::shared_ptr<Query> query);
-
-    /**
-     * This method recovers sends individual request for values per id.
-     *
-     * @param ws     A weak pointer to the Search.
-     * @param query  The initial query passed through the API.
-     * @param n      The node to which send the requests.
-     */
-    void paginate(std::weak_ptr<Search> ws, std::shared_ptr<Query> query, SearchNode* n);
-
     /**
      * If update is true, this method will also send message to synced but non-updated search nodes.
      */
     SearchNode* searchSendGetValues(std::shared_ptr<Search> sr, SearchNode *n = nullptr, bool update = true);
 
-    /**
-     * Forwards an 'announce' request for a list of nodes to the network engine.
-     *
-     * @param sr  The search for which we want to announce a value.
-     * @param announce  The 'announce' structure.
-     */
-    void searchSendAnnounceValue(const std::shared_ptr<Search>& sr);
-
-    /**
-     * Main process of a Search's operations. This function will demand the
-     * network engine to send requests packets for all pending operations
-     * ('get', 'put' and 'listen').
-     *
-     * @param sr  The search to execute its operations.
-     */
     void searchStep(std::shared_ptr<Search> sr);
     void dumpSearch(const Search& sr, std::ostream& out) const;
 
     bool neighbourhoodMaintenance(RoutingTable&);
 
-    void processMessage(const uint8_t *buf, size_t buflen, const SockAddr&);
+    void processMessage(const uint8_t *buf, size_t buflen, const sockaddr *from, socklen_t fromlen);
 
     void onError(std::shared_ptr<Request> node, DhtProtocolException e);
     /* when our address is reported by a distant peer. */
-    void onReportedAddr(const InfoHash& id, const SockAddr&);
+    void onReportedAddr(const InfoHash& id, sockaddr* sa , socklen_t salen);
     /* when we receive a ping request */
     NetworkEngine::RequestAnswer onPing(std::shared_ptr<Node> node);
     /* when we receive a "find node" request */
-    NetworkEngine::RequestAnswer onFindNode(std::shared_ptr<Node> node, const InfoHash& hash, want_t want);
+    NetworkEngine::RequestAnswer onFindNode(std::shared_ptr<Node> node, InfoHash& hash, want_t want);
     void onFindNodeDone(const Request& status, NetworkEngine::RequestAnswer& a, std::shared_ptr<Search> sr);
     /* when we receive a "get values" request */
-    NetworkEngine::RequestAnswer onGetValues(std::shared_ptr<Node> node,
-            const InfoHash& hash,
-            want_t want,
-            const Query& q);
-    void onGetValuesDone(const Request& status,
-            NetworkEngine::RequestAnswer& a,
-            std::shared_ptr<Search>& sr,
-            const std::shared_ptr<Query>& orig_query);
+    NetworkEngine::RequestAnswer onGetValues(std::shared_ptr<Node> node, InfoHash& hash, want_t want);
+    void onGetValuesDone(const Request& status, NetworkEngine::RequestAnswer& a, std::shared_ptr<Search> sr);
     /* when we receive a listen request */
-    NetworkEngine::RequestAnswer onListen(std::shared_ptr<Node> node,
-            const InfoHash& hash,
-            const Blob& token,
-            size_t rid,
-            const Query& query);
-    void onListenDone(const Request& status,
-            NetworkEngine::RequestAnswer& a,
-            std::shared_ptr<Search>& sr,
-            const std::shared_ptr<Query>& orig_query);
+    NetworkEngine::RequestAnswer onListen(std::shared_ptr<Node> node, InfoHash& hash, Blob& token, size_t rid);
+    void onListenDone(const Request& status, NetworkEngine::RequestAnswer& a, std::shared_ptr<Search>& sr);
     /* when we receive an announce request */
     NetworkEngine::RequestAnswer onAnnounce(std::shared_ptr<Node> node,
-            const InfoHash& hash,
-            const Blob& token,
-            const std::vector<std::shared_ptr<Value>>& v,
-            const time_point& created);
-    NetworkEngine::RequestAnswer onRefresh(std::shared_ptr<Node> node,
-            const InfoHash& hash,
-            const Blob& token,
-            const Value::Id& vid);
+            InfoHash& hash, Blob& token, std::vector<std::shared_ptr<Value>> v, time_point created);
     void onAnnounceDone(const Request& status, NetworkEngine::RequestAnswer& a, std::shared_ptr<Search>& sr);
 };
 

include/opendht/dhtrunner.h

@@ -24,8 +24,6 @@
 #include "infohash.h"
 #include "value.h"
 #include "callbacks.h"
-#include "sockaddr.h"
-#include "log_enable.h"
 
 #include <thread>
 #include <mutex>
@@ -56,20 +54,20 @@ public:
     DhtRunner();
     virtual ~DhtRunner();
 
-    void get(InfoHash id, GetCallbackSimple cb, DoneCallback donecb={}, Value::Filter f = Value::AllFilter(), Where w = {}) {
-        get(id, bindGetCb(cb), donecb, f, w);
+    void get(InfoHash id, GetCallbackSimple cb, DoneCallback donecb={}, Value::Filter f = Value::AllFilter()) {
+        get(id, bindGetCb(cb), donecb, f);
     }
 
-    void get(InfoHash id, GetCallbackSimple cb, DoneCallbackSimple donecb={}, Value::Filter f = Value::AllFilter(), Where w = {}) {
-        get(id, bindGetCb(cb), donecb, f, w);
+    void get(InfoHash id, GetCallbackSimple cb, DoneCallbackSimple donecb={}, Value::Filter f = Value::AllFilter()) {
+        get(id, bindGetCb(cb), donecb, f);
     }
 
-    void get(InfoHash hash, GetCallback vcb, DoneCallback dcb, Value::Filter f={}, Where w = {});
+    void get(InfoHash hash, GetCallback vcb, DoneCallback dcb, Value::Filter f={});
 
-    void get(InfoHash id, GetCallback cb, DoneCallbackSimple donecb={}, Value::Filter f = Value::AllFilter(), Where w = {}) {
-        get(id, cb, bindDoneCb(donecb), f, w);
+    void get(InfoHash id, GetCallback cb, DoneCallbackSimple donecb={}, Value::Filter f = Value::AllFilter()) {
+        get(id, cb, bindDoneCb(donecb), f);
     }
-    void get(const std::string& key, GetCallback vcb, DoneCallbackSimple dcb={}, Value::Filter f = Value::AllFilter(), Where w = {});
+    void get(const std::string& key, GetCallback vcb, DoneCallbackSimple dcb={}, Value::Filter f = Value::AllFilter());
 
     template <class T>
     void get(InfoHash hash, std::function<bool(std::vector<T>&&)> cb, DoneCallbackSimple dcb={})
@@ -98,7 +96,7 @@ public:
         getFilterSet<T>());
     }
 
-    std::future<std::vector<std::shared_ptr<dht::Value>>> get(InfoHash key, Value::Filter f = Value::AllFilter(), Where w = {}) {
+    std::future<std::vector<std::shared_ptr<dht::Value>>> get(InfoHash key, Value::Filter f = Value::AllFilter()) {
         auto p = std::make_shared<std::promise<std::vector<std::shared_ptr< dht::Value >>>>();
         auto values = std::make_shared<std::vector<std::shared_ptr< dht::Value >>>();
         get(key, [=](const std::vector<std::shared_ptr<dht::Value>>& vlist) {
@@ -107,7 +105,7 @@ public:
         }, [=](bool) {
             p->set_value(std::move(*values));
         },
-        f, w);
+        f);
         return p->get_future();
     }
 
@@ -124,15 +122,10 @@ public:
         return p->get_future();
     }
 
-    void query(const InfoHash& hash, QueryCallback cb, DoneCallback done_cb = {}, Query q = {});
-    void query(const InfoHash& hash, QueryCallback cb, DoneCallbackSimple done_cb = {}, Query q = {}) {
-        query(hash, cb, bindDoneCb(done_cb), q);
-    }
-
-    std::future<size_t> listen(InfoHash key, GetCallback vcb, Value::Filter f = Value::AllFilter(), Where w = {});
-    std::future<size_t> listen(const std::string& key, GetCallback vcb, Value::Filter f = Value::AllFilter(), Where w = {});
-    std::future<size_t> listen(InfoHash key, GetCallbackSimple cb, Value::Filter f = Value::AllFilter(), Where w = {}) {
-        return listen(key, bindGetCb(cb), f, w);
+    std::future<size_t> listen(InfoHash key, GetCallback vcb, Value::Filter f = Value::AllFilter());
+    std::future<size_t> listen(const std::string& key, GetCallback vcb, Value::Filter f = Value::AllFilter());
+    std::future<size_t> listen(InfoHash key, GetCallbackSimple cb, Value::Filter f = Value::AllFilter()) {
+        return listen(key, bindGetCb(cb), f);
     }
 
     template <class T>
@@ -144,7 +137,7 @@ public:
         getFilterSet<T>());
     }
     template <typename T>
-    std::future<size_t> listen(InfoHash hash, std::function<bool(T&&)> cb, Value::Filter f = Value::AllFilter(), Where w = {})
+    std::future<size_t> listen(InfoHash hash, std::function<bool(T&&)> cb, Value::Filter f = Value::AllFilter())
     {
         return listen(hash, [=](const std::vector<std::shared_ptr<Value>>& vals) {
             for (const auto& v : vals) {
@@ -157,7 +150,7 @@ public:
             }
             return true;
         },
-        getFilterSet<T>(f), w);
+        getFilterSet<T>(f));
     }
 
     void cancelListen(InfoHash h, size_t token);
@@ -198,31 +191,10 @@ public:
     }
     void putEncrypted(const std::string& key, InfoHash to, Value&& value, DoneCallback 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<std::pair<sockaddr_storage, socklen_t>>& nodes, 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 char* host, const char* service);
+    void bootstrap(const std::vector<std::pair<sockaddr_storage, socklen_t>>& nodes);
     void bootstrap(const std::vector<NodeExport>& nodes);
 
-    /**
-     * Add host:service to bootstrap nodes, and ping this node.
-     * DNS resolution is performed asynchronously.
-     * When disconnected, all bootstrap nodes added with this method will be tried regularly until connection
-     * to the DHT network is established.
-     */
-    void bootstrap(const std::string& host, const std::string& service);
-
-    /**
-     * Clear the list of bootstrap added using bootstrap(const std::string&, const std::string&).
-     */
-    void clearBootstrap();
-
     /**
      * Inform the DHT of lower-layer connectivity changes.
      * This will cause the DHT to assume an IP address change.
@@ -240,7 +212,7 @@ public:
      * Returns the currently bound address.
      * @param f: address family of the bound address to retreive.
      */
-    const SockAddr& getBound(sa_family_t f = AF_INET) const {
+    const Address& getBound(sa_family_t f = AF_INET) const {
         return (f == AF_INET) ? bound4 : bound6;
     }
 
@@ -262,11 +234,6 @@ public:
 
     void setLoggers(LogMethod err = NOLOG, LogMethod warn = NOLOG, LogMethod debug = NOLOG);
 
-    /**
-     * Only print logs related to the given InfoHash (if given), or disable filter (if zeroes).
-     */
-    void setLogFilter(const InfoHash& f = {});
-
     void registerType(const ValueType& type);
 
     void importValues(const std::vector<ValuesExport>& values);
@@ -279,10 +246,9 @@ public:
 
     std::vector<unsigned> getNodeMessageStats(bool in = false) const;
     std::string getStorageLog() const;
-    std::string getStorageLog(const InfoHash&) const;
     std::string getRoutingTablesLog(sa_family_t af) const;
     std::string getSearchesLog(sa_family_t af = 0) const;
-    std::vector<SockAddr> getPublicAddress(sa_family_t af = 0);
+    std::vector<Address> getPublicAddress(sa_family_t af = 0);
     std::vector<std::string> getPublicAddressStr(sa_family_t af = 0);
 
     // securedht methods
@@ -302,13 +268,12 @@ public:
      * @param threaded: If false, ::loop() must be called periodically. Otherwise a thread is launched.
      * @param cb: Optional callback to receive general state information.
      */
-    void run(in_port_t port, const crypto::Identity identity, bool threaded = false, NetId network = 0) {
+    void run(in_port_t port, const crypto::Identity identity, bool threaded = false, bool is_bootstrap = false) {
         run(port, {
             /*.dht_config = */{
                 /*.node_config = */{
                     /*.node_id = */{},
-                    /*.network = */network,
-                    /*.is_bootstrap = */false
+                    /*.is_bootstrap = */is_bootstrap
                 },
                 /*.id = */identity
             },
@@ -359,20 +324,11 @@ public:
     void join();
 
 private:
-    static constexpr std::chrono::seconds BOOTSTRAP_PERIOD {10};
-
-    /**
-     * Will try to resolve the list of hostnames `bootstrap_nodes` on seperate
-     * thread and then queue ping requests. This list should contain reliable
-     * nodes so that the DHT node can recover quickly from losing connection
-     * with the network.
-     */
-    void tryBootstrapCoutinuously();
 
     void doRun(const sockaddr_in* sin4, const sockaddr_in6* sin6, SecureDhtConfig config);
     time_point loop_();
 
-    static std::vector<std::pair<sockaddr_storage, socklen_t>> getAddrInfo(const std::string& host, const std::string& service);
+    static std::vector<std::pair<sockaddr_storage, socklen_t>> getAddrInfo(const char* host, const char* service);
 
     NodeStatus getStatus() const {
         return std::max(status4, status6);
@@ -385,17 +341,7 @@ private:
 
     std::thread rcv_thread {};
     std::mutex sock_mtx {};
-    std::vector<std::pair<Blob, SockAddr>> rcv {};
-
-    /** true if currently actively boostraping */
-    std::atomic_bool bootstraping {false};
-    /* bootstrap nodes given as (host, service) pairs */
-    std::vector<std::pair<std::string,std::string>> bootstrap_nodes_all {};
-    std::vector<std::pair<std::string,std::string>> bootstrap_nodes {};
-    std::thread bootstrap_thread {};
-    /** protects bootstrap_nodes, bootstrap_thread */
-    std::mutex bootstrap_mtx {};
-    std::condition_variable bootstrap_cv {};
+    std::vector<std::pair<Blob, std::pair<sockaddr_storage, socklen_t>>> rcv {};
 
     std::queue<std::function<void(SecureDht&)>> pending_ops_prio {};
     std::queue<std::function<void(SecureDht&)>> pending_ops {};
@@ -407,8 +353,8 @@ private:
                status6 {NodeStatus::Disconnected};
     StatusCallback statusCb {nullptr};
 
-    SockAddr bound4 {};
-    SockAddr bound6 {};
+    Address bound4 {};
+    Address bound6 {};
 };
 
 }

include/opendht/indexation/pht.h

@@ -262,7 +262,7 @@ struct IndexEntry : public dht::Value::Serializable<IndexEntry> {
     Blob prefix;
     Value value;
     std::string name;
-    MSGPACK_DEFINE_MAP(prefix, value)
+    MSGPACK_DEFINE_MAP(prefix, value);
 };
 
 class Pht {

include/opendht/infohash.h

@@ -197,7 +197,6 @@ public:
     static InfoHash getRandom();
 
     friend std::ostream& operator<< (std::ostream& s, const InfoHash& h);
-    friend std::istream& operator>> (std::istream& s, InfoHash& h);
 
     std::string toString() const;
 

include/opendht/log_enable.h

@@ -1,175 +0,0 @@
-/*
- *  Copyright (C) 2016 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 <http://www.gnu.org/licenses/>.
- */
-
-#pragma once
-
-#ifdef HAVE_CONFIG_H
-#include "config.h"
-#endif
-
-#include "infohash.h"
-
-#ifndef OPENDHT_LOG
-#define OPENDHT_LOG true
-#endif
-
-namespace dht {
-
-// Logging related utility functions
-
-/**
- * Dummy function used to disable logging
- */
-inline void NOLOG(char const*, va_list) {}
-
-/**
- * Wrapper for logging methods
- */
-struct LogMethod {
-    LogMethod() = default;
-
-    template<typename T>
-    LogMethod(T&& t) : func(std::forward<T>(t)) {}
-
-    void operator()(char const* format, ...) const {
-        va_list args;
-        va_start(args, format);
-        func(format, args);
-        va_end(args);
-    }
-    void log(char const* format, va_list args) const {
-        func(format, args);
-    }
-    explicit operator bool() const {
-        return (bool)func;
-    }
-
-    void logPrintable(const uint8_t *buf, size_t buflen) const {
-        std::string buf_clean(buflen, '\0');
-        for (size_t i=0; i<buflen; i++)
-            buf_clean[i] = isprint(buf[i]) ? buf[i] : '.';
-        (*this)("%s", buf_clean.c_str());
-    }
-private:
-    std::function<void(char const*, va_list)> func;
-};
-
-struct Logger {
-    LogMethod DEBUG = NOLOG;
-    LogMethod WARN = NOLOG;
-    LogMethod ERR = NOLOG;
-    void setFilter(const InfoHash& f) {
-        filter_ = f;
-        filterEnable_ = filter_ != InfoHash{};
-    }
-    inline void log0(const LogMethod& logger, char const* format, va_list args) const {
-#if OPENDHT_LOG
-        if (logger and not filterEnable_)
-            logger.log(format, args);
-#endif
-    }
-    inline void log1(const LogMethod& logger, const InfoHash& f, char const* format, va_list args) const {
-#if OPENDHT_LOG
-        if (logger and (not filterEnable_ or f == filter_))
-            logger.log(format, args);
-#endif
-    }
-    inline void log2(const LogMethod& logger, const InfoHash& f1, const InfoHash& f2, char const* format, va_list args) const {
-#if OPENDHT_LOG
-        if (logger and (not filterEnable_ or f1 == filter_ or f2 == filter_))
-            logger.log(format, args);
-#endif
-    }
-    inline void d(char const* format, ...) const {
-#if OPENDHT_LOG
-        va_list args;
-        va_start(args, format);
-        log0(DEBUG, format, args);
-        va_end(args);
-#endif
-    }
-    inline void d(const InfoHash& f, char const* format, ...) const {
-#if OPENDHT_LOG
-        va_list args;
-        va_start(args, format);
-        log1(DEBUG, f, format, args);
-        va_end(args);
-#endif
-    }
-    inline void d(const InfoHash& f1, const InfoHash& f2, char const* format, ...) const {
-#if OPENDHT_LOG
-        va_list args;
-        va_start(args, format);
-        log2(DEBUG, f1, f2, format, args);
-        va_end(args);
-#endif
-    }
-    inline void w(char const* format, ...) const {
-#if OPENDHT_LOG
-        va_list args;
-        va_start(args, format);
-        log0(WARN, format, args);
-        va_end(args);
-#endif
-    }
-    inline void w(const InfoHash& f, char const* format, ...) const {
-#if OPENDHT_LOG
-        va_list args;
-        va_start(args, format);
-        log1(WARN, f, format, args);
-        va_end(args);
-#endif
-    }
-    inline void w(const InfoHash& f1, const InfoHash& f2, char const* format, ...) const {
-#if OPENDHT_LOG
-        va_list args;
-        va_start(args, format);
-        log2(WARN, f1, f2, format, args);
-        va_end(args);
-#endif
-    }
-    inline void e(char const* format, ...) const {
-#if OPENDHT_LOG
-        va_list args;
-        va_start(args, format);
-        log0(ERR, format, args);
-        va_end(args);
-#endif
-    }
-    inline void e(const InfoHash& f, char const* format, ...) const {
-#if OPENDHT_LOG
-        va_list args;
-        va_start(args, format);
-        log1(ERR, f, format, args);
-        va_end(args);
-#endif
-    }
-    inline void e(const InfoHash& f1, const InfoHash& f2, char const* format, ...) const {
-#if OPENDHT_LOG
-        va_list args;
-        va_start(args, format);
-        log2(ERR, f1, f2, format, args);
-        va_end(args);
-#endif
-    }
-private:
-    bool filterEnable_ {false};
-    InfoHash filter_ {};
-};
-
-}

include/opendht/network_engine.h

@@ -14,9 +14,11 @@
  *  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 <http://www.gnu.org/licenses/>.
+ *  along with this program; if not, write to the Free Software
+ *  Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301 USA.
  */
 
+
 #pragma once
 
 #include "node_cache.h"
@@ -26,7 +28,6 @@
 #include "scheduler.h"
 #include "utils.h"
 #include "rng.h"
-#include "rate_limiter.h"
 
 #include <vector>
 #include <string>
@@ -47,7 +48,6 @@ public:
     // sent to another peer (http-like).
     static const constexpr uint16_t NON_AUTHORITATIVE_INFORMATION {203}; /* incomplete request packet. */
     static const constexpr uint16_t UNAUTHORIZED {401};                  /* wrong tokens. */
-    static const constexpr uint16_t NOT_FOUND {404};                     /* storage not found */
     // for internal use (custom).
     static const constexpr uint16_t INVALID_TID_SIZE {421};              /* id was truncated. */
     static const constexpr uint16_t UNKNOWN_TID {422};                   /* unknown tid */
@@ -58,7 +58,6 @@ public:
     static const std::string LISTEN_WRONG_TOKEN; /* wrong token in "listen" request */
     static const std::string PUT_NO_INFOHASH;    /* no infohash in "put" request */
     static const std::string PUT_WRONG_TOKEN;    /* got "put" request with wrong token */
-    static const std::string STORAGE_NOT_FOUND;  /* got access request for an unknown storage */
     static const std::string PUT_INVALID_ID;     /* invalid id in "put" request */
 
     DhtProtocolException(uint16_t code, const std::string& msg="", InfoHash failing_node_id={})
@@ -92,17 +91,14 @@ struct ParsedMessage;
  * @param onGetValues    callback for "get values" request.
  * @param onListen       callback for "listen" request.
  * @param onAnnounce     callback for "announce" request.
- * @param onRefresh      callback for "refresh" request.
  */
-class NetworkEngine final
-{
+class NetworkEngine final {
     struct TransPrefix : public  std::array<uint8_t, 2>  {
         TransPrefix(const std::string& str) : std::array<uint8_t, 2>({{(uint8_t)str[0], (uint8_t)str[1]}}) {}
         static const TransPrefix PING;
         static const TransPrefix FIND_NODE;
         static const TransPrefix GET_VALUES;
         static const TransPrefix ANNOUNCE_VALUES;
-        static const TransPrefix REFRESH;
         static const TransPrefix LISTEN;
     };
 public:
@@ -157,7 +153,6 @@ public:
         Blob ntoken {};
         Value::Id vid {};
         std::vector<std::shared_ptr<Value>> values {};
-        std::vector<std::shared_ptr<FieldValueIndex>> fields {};
         std::vector<std::shared_ptr<Node>> nodes4 {};
         std::vector<std::shared_ptr<Node>> nodes6 {};
         RequestAnswer() {}
@@ -171,7 +166,7 @@ public:
      */
     void cancelRequest(std::shared_ptr<Request>& req);
 
-    void connectivityChanged(sa_family_t);
+    void connectivityChanged();
 
 private:
 
@@ -200,7 +195,7 @@ private:
      * @param saddr (type: sockaddr*) sockaddr* pointer containing address ip information.
      * @param saddr_len (type: socklen_t) lenght of the sockaddr struct.
      */
-    std::function<void(const InfoHash&, const SockAddr&)> onReportedAddr;
+    std::function<void(const InfoHash&, sockaddr*, socklen_t)> onReportedAddr;
     /**
      * @brief on ping request callback.
      *
@@ -216,7 +211,7 @@ private:
      *             or ipv6.
      */
     std::function<RequestAnswer(std::shared_ptr<Node>,
-            const InfoHash&,
+            InfoHash&,
             want_t)> onFindNode {};
     /**
      * @brief on "get values" request callback.
@@ -227,9 +222,8 @@ private:
      *             or ipv6.
      */
     std::function<RequestAnswer(std::shared_ptr<Node>,
-            const InfoHash&,
-            want_t,
-            const Query&)> onGetValues {};
+            InfoHash&,
+            want_t)> onGetValues {};
     /**
      * @brief on listen request callback.
      *
@@ -239,10 +233,9 @@ private:
      * @param rid (type: uint16_t) request id.
      */
     std::function<RequestAnswer(std::shared_ptr<Node>,
-            const InfoHash&,
-            const Blob&,
-            uint16_t,
-            const Query&)> onListen {};
+            InfoHash&,
+            Blob&,
+            uint16_t)> onListen {};
     /**
      * @brief on announce request callback.
      *
@@ -253,29 +246,17 @@ private:
      * @param created (type: time_point) time when the value was created.
      */
     std::function<RequestAnswer(std::shared_ptr<Node>,
-            const InfoHash&,
-            const Blob&,
-            const std::vector<std::shared_ptr<Value>>&,
-            const time_point&)> onAnnounce {};
-    /**
-     * @brief on refresh request callback.
-     *
-     * @param node (type: std::shared_ptr<Node>) the requesting node.
-     * @param vhash (type: InfoHash) hash of the value of interest.
-     * @param token (type: Blob) security token.
-     * @param vid (type: Value::id) the value id.
-     */
-    std::function<RequestAnswer(std::shared_ptr<Node>,
-            const InfoHash&,
-            const Blob&,
-            const Value::Id&)> onRefresh {};
+            InfoHash&,
+            Blob&,
+            std::vector<std::shared_ptr<Value>>,
+            time_point)> onAnnounce {};
 
 public:
     using RequestCb = std::function<void(const Request&, RequestAnswer&&)>;
     using RequestExpiredCb = std::function<void(const Request&, bool)>;
 
-    NetworkEngine(Logger& log, Scheduler& scheduler);
-    NetworkEngine(InfoHash& myid, NetId net, int s, int s6, Logger& log, Scheduler& scheduler,
+    NetworkEngine(Logger& log, Scheduler& scheduler) : myid(zeroes), DHT_LOG(log), scheduler(scheduler) {}
+    NetworkEngine(InfoHash& myid, int s, int s6, Logger& log, Scheduler& scheduler,
             decltype(NetworkEngine::onError) onError,
             decltype(NetworkEngine::onNewNode) onNewNode,
             decltype(NetworkEngine::onReportedAddr) onReportedAddr,
@@ -283,10 +264,16 @@ public:
             decltype(NetworkEngine::onFindNode) onFindNode,
             decltype(NetworkEngine::onGetValues) onGetValues,
             decltype(NetworkEngine::onListen) onListen,
-            decltype(NetworkEngine::onAnnounce) onAnnounce,
-            decltype(NetworkEngine::onRefresh) onRefresh);
-
-    virtual ~NetworkEngine();
+            decltype(NetworkEngine::onAnnounce) onAnnounce) :
+        onError(onError), onNewNode(onNewNode), onReportedAddr(onReportedAddr), onPing(onPing), onFindNode(onFindNode),
+        onGetValues(onGetValues), onListen(onListen), onAnnounce(onAnnounce), myid(myid),
+        dht_socket(s), dht_socket6(s6), DHT_LOG(log), scheduler(scheduler)
+    {
+        transaction_id = std::uniform_int_distribution<decltype(transaction_id)>{1}(rd_device);
+    }
+    virtual ~NetworkEngine() {
+        clear();
+    };
 
     void clear();
 
@@ -303,9 +290,9 @@ public:
      * @param nodes6  The ipv6 closest nodes.
      * @param values  The values to send.
      */
-    void tellListener(std::shared_ptr<Node> n, uint16_t rid, const InfoHash& hash, want_t want, const Blob& ntoken,
-            std::vector<std::shared_ptr<Node>>&& nodes, std::vector<std::shared_ptr<Node>>&& nodes6,
-            std::vector<std::shared_ptr<Value>>&& values, const Query& q);
+    void tellListener(std::shared_ptr<Node> n, uint16_t rid, InfoHash hash, want_t want, Blob ntoken,
+            std::vector<std::shared_ptr<Node>> nodes, std::vector<std::shared_ptr<Node>> nodes6,
+            std::vector<std::shared_ptr<Value>> values);
 
     bool isRunning(sa_family_t af) const;
     inline want_t want () const { return dht_socket >= 0 && dht_socket6 >= 0 ? (WANT4 | WANT6) : -1; }
@@ -327,33 +314,24 @@ public:
                 RequestExpiredCb on_expired);
     std::shared_ptr<Request>
         sendGetValues(std::shared_ptr<Node> n,
-                const InfoHash& info_hash,
-                const Query& query,
+                const InfoHash& target,
                 want_t want,
                 RequestCb on_done,
                 RequestExpiredCb on_expired);
     std::shared_ptr<Request>
         sendListen(std::shared_ptr<Node> n,
                 const InfoHash& infohash,
-                const Query& query,
                 const Blob& token,
                 RequestCb on_done,
                 RequestExpiredCb on_expired);
     std::shared_ptr<Request>
         sendAnnounceValue(std::shared_ptr<Node> n,
                 const InfoHash& infohash,
-                const std::shared_ptr<Value>& v,
+                const Value& v,
                 time_point created,
                 const Blob& token,
                 RequestCb on_done,
                 RequestExpiredCb on_expired);
-    std::shared_ptr<Request>
-        sendRefreshValue(std::shared_ptr<Node> n,
-                const InfoHash& infohash,
-                const Value::Id& vid,
-                const Blob& token,
-                RequestCb on_done,
-                RequestExpiredCb on_expired);
 
     /**
      * Parses a message and calls appropriate callbacks.
@@ -364,12 +342,10 @@ public:
      * @param fromlen  The length of the corresponding sockaddr structure.
      * @param now  The time to adjust the clock in the network engine.
      */
-    void processMessage(const uint8_t *buf, size_t buflen, const SockAddr& addr);
+    void processMessage(const uint8_t *buf, size_t buflen, const sockaddr* from, socklen_t fromlen);
 
-    std::shared_ptr<Node> insertNode(const InfoHash& myid, const SockAddr& addr) {
-        auto n = cache.getNode(myid, addr, scheduler.time(), 0);
-        onNewNode(n, 0);
-        return n;
+    std::shared_ptr<Node> insertNode(const InfoHash& myid, const sockaddr* from, socklen_t fromlen) {
+        return cache.getNode(myid, from, fromlen, scheduler.time(), 0);
     }
 
     std::vector<unsigned> getNodeMessageStats(bool in) {
@@ -381,14 +357,7 @@ public:
 
     void blacklistNode(const std::shared_ptr<Node>& n);
 
-    std::vector<std::shared_ptr<Node>> getCachedNodes(const InfoHash& id, sa_family_t sa_f, size_t count) {
-        return cache.getCachedNodes(id, sa_f, count);
-    }
-
 private:
-
-    struct PartialMessage;
-
     /***************
      *  Constants  *
      ***************/
@@ -397,29 +366,28 @@ private:
     static const constexpr size_t NODE4_INFO_BUF_LEN {26};
     /* the length of a node info buffer in ipv6 format */
     static const constexpr size_t NODE6_INFO_BUF_LEN {38};
-    /* after a UDP reply, the period during which we tell the link layer about it */
+    /* TODO */
     static constexpr std::chrono::seconds UDP_REPLY_TIME {15};
-
-    /* Max. time to receive a full fragmented packet */
-    static constexpr std::chrono::seconds RX_MAX_PACKET_TIME {10};
-    /* Max. time between packet fragments */
-    static constexpr std::chrono::seconds RX_TIMEOUT {3};
     /* The maximum number of nodes that we snub.  There is probably little
         reason to increase this value. */
     static constexpr unsigned BLACKLISTED_MAX {10};
-
-    static constexpr size_t MTU {1280};
-    static constexpr size_t MAX_PACKET_VALUE_SIZE {8 * 1024};
-
+    /* TODO */
     static const std::string my_v;
-    static std::mt19937 rd_device;
 
-    void process(std::unique_ptr<ParsedMessage>&&, const SockAddr& from);
+    /* DHT info */
+    const InfoHash& myid;
+    const int dht_socket {-1};
+    const int dht_socket6 {-1};
+    const Logger& DHT_LOG;
 
-    bool rateLimit(const SockAddr& addr);
+    NodeCache cache {};
+    sockaddr_storage blacklist[BLACKLISTED_MAX] {};
+    unsigned next_blacklisted = 0;
 
-    static bool isMartian(const SockAddr& addr);
-    bool isNodeBlacklisted(const SockAddr& addr) const;
+    bool rateLimit();
+
+    static bool isMartian(const sockaddr* sa, socklen_t len);
+    bool isNodeBlacklisted(const sockaddr*, socklen_t) const;
 
     void requestStep(std::shared_ptr<Request> req);
 
@@ -445,29 +413,24 @@ private:
         unsigned get {0};
         unsigned put {0};
         unsigned listen {0};
-        unsigned refresh {0};
     };
 
 
     // basic wrapper for socket sendto function
-    int send(const char *buf, size_t len, int flags, const SockAddr& addr);
-
-    void sendValueParts(TransId tid, const std::vector<Blob>& svals, const SockAddr& addr);
-    std::vector<Blob> packValueHeader(msgpack::sbuffer&, const std::vector<std::shared_ptr<Value>>&);
-    void maintainRxBuffer(const TransId& tid);
+    int send(const char *buf, size_t len, int flags, const sockaddr *sa, socklen_t salen);
 
     /*************
      *  Answers  *
      *************/
     /* answer to a ping  request */
-    void sendPong(const SockAddr& addr, TransId tid);
+    void sendPong(const sockaddr* sa, socklen_t salen, TransId tid);
     /* answer to findnodes/getvalues request */
-    void sendNodesValues(const SockAddr& addr,
+    void sendNodesValues(const sockaddr* sa,
+            socklen_t salen,
             TransId tid,
             const Blob& nodes,
             const Blob& nodes6,
             const std::vector<std::shared_ptr<Value>>& st,
-            const Query& query,
             const Blob& token);
     Blob bufferNodes(sa_family_t af, const InfoHash& id, std::vector<std::shared_ptr<Node>>& nodes);
 
@@ -477,67 +440,26 @@ private:
             std::vector<std::shared_ptr<Node>>& nodes,
             std::vector<std::shared_ptr<Node>>& nodes6);
     /* answer to a listen request */
-    void sendListenConfirmation(const SockAddr& addr, TransId tid);
+    void sendListenConfirmation(const sockaddr* sa, socklen_t salen, TransId tid);
     /* answer to put request */
-    void sendValueAnnounced(const SockAddr& addr, TransId, Value::Id);
+    void sendValueAnnounced(const sockaddr* sa, socklen_t salen, TransId, Value::Id);
     /* answer in case of error */
-    void sendError(const SockAddr& addr,
+    void sendError(const sockaddr* sa,
+            socklen_t salen,
             TransId tid,
             uint16_t code,
             const std::string& message,
             bool include_id=false);
 
-    void deserializeNodes(ParsedMessage& msg);
+    void deserializeNodesValues(ParsedMessage& msg);
 
-    /* DHT info */
-    const InfoHash& myid;
-    const NetId network {0};
-    const int dht_socket {-1};
-    const int dht_socket6 {-1};
-    const Logger& DHT_LOG;
-
-    NodeCache cache {};
-
-    /**
-     * A comparator to classify IP addresses, only considering the
-     * first 64 bits in IPv6.
-     */
-    struct cmpSockAddr {
-        bool operator()(const SockAddr& a, const SockAddr& b) const {
-            if (a.second != b.second)
-                return a.second < b.second;
-            socklen_t start, len;
-            switch(a.getFamily()) {
-                case AF_INET:
-                    start = offsetof(sockaddr_in, sin_addr);
-                    len = sizeof(in_addr);
-                    break;
-                case AF_INET6:
-                    start = offsetof(sockaddr_in6, sin6_addr);
-                    // don't consider more than 64 bits (IPv6)
-                    len = 8;
-                    break;
-                default:
-                    start = 0;
-                    len = a.second;
-                    break;
-            }
-
-            return std::memcmp((uint8_t*)&a.first+start, (uint8_t*)&b.first+start, len) < 0;
-        }
-    };
-    // global limiting should be triggered by at least 8 different IPs
-    using IpLimiter = RateLimiter<MAX_REQUESTS_PER_SEC/8>;
-    using IpLimiterMap = std::map<SockAddr, IpLimiter, cmpSockAddr>;
-    IpLimiterMap address_rate_limiter {};
-    RateLimiter<MAX_REQUESTS_PER_SEC> rate_limiter {};
+    std::queue<time_point> rate_limit_time {};
+    static std::mt19937 rd_device;
 
     // requests handling
     uint16_t transaction_id {1};
     std::map<uint16_t, std::shared_ptr<Request>> requests {};
-    std::map<TransId, PartialMessage> partial_messages;
     MessageStats in_stats {}, out_stats {};
-    std::set<SockAddr> blacklist {};
 
     Scheduler& scheduler;
 };

include/opendht/node.h

@@ -22,7 +22,6 @@
 
 #include "infohash.h" // includes socket structures
 #include "utils.h"
-#include "sockaddr.h"
 
 #include <list>
 
@@ -31,46 +30,40 @@ namespace dht {
 struct Request;
 
 struct Node {
+    friend class NetworkEngine;
+
     InfoHash id;
-    SockAddr addr;
+
+    socklen_t sslen {0};
+    sockaddr_storage ss;
 
     time_point time {time_point::min()};            /* last time eared about */
     time_point reply_time {time_point::min()};      /* time of last correct reply received */
 
     Node(const InfoHash& id, const sockaddr* sa, socklen_t salen)
-        : id(id), addr(sa, salen) {}
-    Node(const InfoHash& id, const SockAddr& addr) : id(id), addr(addr) {}
-
+        : id(id), sslen(salen), ss() {
+        std::copy_n((const uint8_t*)sa, salen, (uint8_t*)&ss);
+        if ((unsigned)salen < sizeof(ss))
+            std::fill_n((uint8_t*)&ss+salen, sizeof(ss)-salen, 0);
+    }
     InfoHash getId() const {
         return id;
     }
     std::pair<const sockaddr*, socklen_t> getAddr() const {
-        return {(const sockaddr*)&addr.first, addr.second};
+        return {(const sockaddr*)&ss, sslen};
     }
     std::string getAddrStr() const {
-        return addr.toString();
+        return print_addr(ss, sslen);
     }
-
-    /**
-     * Makes notice about an additionnal authentication error with this node. Up
-     * to MAX_AUTH_ERRORS errors are accepted in order to let the node recover.
-     * Upon this limit, the node expires.
-     */
-    void authError() {
-        if (++auth_errors > MAX_AUTH_ERRORS)
-            setExpired();
-    }
-    void authSuccess() { auth_errors = 0; }
-
     bool isExpired() const { return expired_; }
     bool isGood(time_point now) const;
     bool isPendingMessage() const;
     size_t getPendingMessageCount() const;
 
-    NodeExport exportNode() const { return NodeExport {id, addr.first, addr.second}; }
-    sa_family_t getFamily() const { return addr.getFamily(); }
+    NodeExport exportNode() const { return NodeExport {id, ss, sslen}; }
+    sa_family_t getFamily() const { return ss.ss_family; }
 
-    void update(const SockAddr&);
+    void update(const sockaddr* sa, socklen_t salen);
 
     void requested(std::shared_ptr<Request>& req);
     void received(time_point now, std::shared_ptr<Request> req);
@@ -80,7 +73,7 @@ struct Node {
     /**
      * Resets the state of the node so it's not expired anymore.
      */
-    void reset() { expired_ = false; reply_time = time_point::min(); }
+    void reset() { expired_ = false; }
 
     std::string toString() const;
 
@@ -95,11 +88,8 @@ struct Node {
     static constexpr const std::chrono::seconds MAX_RESPONSE_TIME {1};
 
 private:
-    /* Number of times we accept authentication errors from this node. */
-    static const constexpr unsigned MAX_AUTH_ERRORS {3};
 
     std::list<std::weak_ptr<Request>> requests_ {};
-    unsigned auth_errors {0};
     bool expired_ {false};
 
     void clearPendingQueue() {
@@ -107,6 +97,7 @@ private:
             return w.expired();
         });
     }
+
 };
 
 }

include/opendht/node_cache.h

@@ -28,8 +28,7 @@ namespace dht {
 
 struct NodeCache {
     std::shared_ptr<Node> getNode(const InfoHash& id, sa_family_t family);
-    std::shared_ptr<Node> getNode(const InfoHash& id, const SockAddr&, time_point now, bool confirmed);
-    std::vector<std::shared_ptr<Node>> getCachedNodes(const InfoHash& id, sa_family_t sa_f, size_t count);
+    std::shared_ptr<Node> getNode(const InfoHash& id, const sockaddr* sa, socklen_t sa_len, time_point now, int confirmed);
 
     /**
      * Reset the connectivity state of every node,
@@ -42,7 +41,7 @@ private:
     class NodeMap : public std::map<InfoHash, std::weak_ptr<Node>> {
     public:
         std::shared_ptr<Node> getNode(const InfoHash& id);
-        std::shared_ptr<Node> getNode(const InfoHash& id, const SockAddr&, time_point now, bool confirmed);
+        std::shared_ptr<Node> getNode(const InfoHash& id, const sockaddr* sa, socklen_t sa_len, time_point now, int confirmed);
         void clearBadNodes();
     };
 

include/opendht/rate_limiter.h

@@ -1,50 +0,0 @@
-/*
- *  Copyright (C) 2016 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 <http://www.gnu.org/licenses/>.
- */
-
-#pragma once
-
-#include "utils.h"
-#include <queue>
-
-namespace dht {
-
-template<size_t Quota, unsigned long Perdiod=1>
-class RateLimiter {
-public:
-    /** Clear outdated records and return current quota usage */ 
-    size_t maintain(const time_point& now) {
-        using namespace std::chrono;
-        while (not records.empty() and duration_cast<seconds>(now - records.front()) >= seconds(Perdiod))
-            records.pop();
-        return records.size();
-    }
-    /** Return false if quota is reached, insert record and return true otherwise. */
-    bool limit(const time_point& now) {
-        if (maintain(now) >= Quota)
-            return false;
-        records.emplace(now);
-        return true;
-    }
-    bool empty() const {
-        return records.empty();
-    }
-private:
-    std::queue<time_point> records {};
-};
-
-}

include/opendht/routing_table.h

@@ -23,8 +23,6 @@
 
 namespace dht {
 
-static constexpr unsigned TARGET_NODES {8};
-
 struct Bucket {
     Bucket() : cached() {}
     Bucket(sa_family_t af, const InfoHash& f = {}, time_point t = time_point::min())

include/opendht/scheduler.h

@@ -22,7 +22,6 @@
 #pragma once
 
 #include "utils.h"
-#include "log_enable.h"
 
 #include <functional>
 #include <map>
@@ -37,10 +36,8 @@ namespace dht {
  */
 class Scheduler {
 public:
-    Scheduler(const Logger& l) : DHT_LOG(l) {}
-
     struct Job {
-        Job(std::function<void()>&& f) : do_(std::move(f)) {}
+        Job(std::function<void()>&& f) : do_(std::forward<std::function<void()>>(f)) {}
         std::function<void()> do_;
     };
 
@@ -53,7 +50,7 @@ public:
      * @return pointer to the newly scheduled job.
      */
     std::shared_ptr<Scheduler::Job> add(time_point t, std::function<void()>&& job_func) {
-        auto job = std::make_shared<Job>(std::move(job_func));
+        auto job = std::make_shared<Job>(std::forward<std::function<void()>>(job_func));
         if (t != time_point::max())
             timers.emplace(std::move(t), job);
         return job;
@@ -69,14 +66,10 @@ public:
      */
     void edit(std::shared_ptr<Scheduler::Job>& job, time_point t) {
         if (not job) {
-            DHT_LOG.ERR("editing an empty job");
+            std::cerr << "editing an empty job" << std::endl;
             return;
         }
-        // std::function move doesn't garantee to leave the object empty.
-        // Force clearing old value.
-        auto task = std::move(job->do_);
-        job->do_ = {};
-        job = add(t, std::move(task));
+        job = add(t, std::move(job->do_));
     }
 
     /**
@@ -96,16 +89,17 @@ public:
             if (timer->first > now)
                 break;
 
-            auto job = std::move(timer->second);
+            const auto& job = *timer->second;
+            if (job.do_)
+                job.do_();
             timers.erase(timer);
-
-            if (job->do_)
-                job->do_();
         }
         return getNextJobTime();
     }
 
     inline time_point getNextJobTime() const {
+        //if (not timers.empty())
+        //    std::cout << "Next job in " << print_dt(timers.begin()->first - clock::now()) << std::endl;
         return timers.empty() ? time_point::max() : timers.begin()->first;
     }
 
@@ -119,7 +113,6 @@ public:
 private:
     time_point now {clock::now()};
     std::multimap<time_point, std::shared_ptr<Job>> timers {}; /* the jobs ordered by time */
-    const Logger& DHT_LOG;
 };
 
 }

include/opendht/securedht.h

@@ -70,7 +70,7 @@ public:
         return secureType(std::move(tmp_type));
     }
 
-    virtual void registerType(const ValueType& type) override {
+    virtual void registerType(const ValueType& type) {
         Dht::registerType(secureType(type));
     }
     virtual void registerType(ValueType&& type) {
@@ -85,18 +85,18 @@ public:
      * If the signature can't be checked, or if the data can't be decrypted, it is not returned.
      * Public, non-signed & non-encrypted data is retransmitted as-is.
      */
-    virtual void get(const InfoHash& id, GetCallback cb, DoneCallback donecb={}, Value::Filter&& = {}, Where&& w = {}) override;
-    virtual void get(const InfoHash& id, GetCallback cb, DoneCallbackSimple donecb={}, Value::Filter&& f = {}, Where&& w = {}) override {
-        get(id, cb, bindDoneCb(donecb), std::forward<Value::Filter>(f), std::forward<Where>(w));
+    virtual void get(const InfoHash& id, GetCallback cb, DoneCallback donecb={}, Value::Filter&& = {}) override;
+    virtual void get(const InfoHash& id, GetCallback cb, DoneCallbackSimple donecb={}, Value::Filter&& f = {}) override {
+        get(id, cb, bindDoneCb(donecb), std::forward<Value::Filter>(f));
     }
-    virtual void get(const InfoHash& key, GetCallbackSimple cb, DoneCallback donecb={}, Value::Filter&& f={}, Where&& w = {}) override {
-        get(key, bindGetCb(cb), donecb, std::forward<Value::Filter>(f), std::forward<Where>(w));
+    virtual void get(const InfoHash& key, GetCallbackSimple cb, DoneCallback donecb={}, Value::Filter&& f={}) override {
+        get(key, bindGetCb(cb), donecb, std::forward<Value::Filter>(f));
     }
-    virtual void get(const InfoHash& key, GetCallbackSimple cb, DoneCallbackSimple donecb, Value::Filter&& f={}, Where&& w = {}) override {
-        get(key, bindGetCb(cb), bindDoneCb(donecb), std::forward<Value::Filter>(f), std::forward<Where>(w));
+    virtual void get(const InfoHash& key, GetCallbackSimple cb, DoneCallbackSimple donecb, Value::Filter&& f={}) override {
+        get(key, bindGetCb(cb), bindDoneCb(donecb), std::forward<Value::Filter>(f));
     }
 
-    virtual size_t listen(const InfoHash& id, GetCallback cb, Value::Filter&& = {}, Where&& w = {}) override;
+    virtual size_t listen(const InfoHash& id, GetCallback cb, Value::Filter&& = {}) override;
 
     /**
      * Will take ownership of the value, sign it using our private key and put it in the DHT.
@@ -126,13 +126,16 @@ public:
     Value decrypt(const Value& v);
 
     void findCertificate(const InfoHash& node, std::function<void(const std::shared_ptr<crypto::Certificate>)> cb);
-    void findPublicKey(const InfoHash& node, std::function<void(const std::shared_ptr<const crypto::PublicKey>)> cb);
+    void findPublicKey(const InfoHash& node, std::function<void(const std::shared_ptr<crypto::PublicKey>)> cb);
 
     const std::shared_ptr<crypto::Certificate> registerCertificate(const InfoHash& node, const Blob& cert);
     void registerCertificate(std::shared_ptr<crypto::Certificate>& cert);
 
     const std::shared_ptr<crypto::Certificate> getCertificate(const InfoHash& node) const;
-    const std::shared_ptr<const crypto::PublicKey> getPublicKey(const InfoHash& node) const;
+    const std::shared_ptr<crypto::PublicKey> getPublicKey(const InfoHash& node) const;
+
+
+    
 
     /**
      * Allows to set a custom callback called by the library to find a locally-stored certificate.
@@ -158,7 +161,7 @@ private:
 
     // our certificate cache
     std::map<InfoHash, std::shared_ptr<crypto::Certificate>> nodesCertificates_ {};
-    std::map<InfoHash, std::shared_ptr<const crypto::PublicKey>> nodesPubKeys_ {};
+    std::map<InfoHash, std::shared_ptr<crypto::PublicKey>> nodesPubKeys_ {};
 
     std::uniform_int_distribution<Value::Id> rand_id {};
 };

include/opendht/sockaddr.h

@@ -1,75 +0,0 @@
-/*
- *  Copyright (C) 2016 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 <http://www.gnu.org/licenses/>.
- */
-
-#pragma once
-
-#ifndef _WIN32
-#include <sys/socket.h>
-#else
-#include <iso646.h>
-#include <ws2def.h>
-#include <ws2tcpip.h>
-typedef uint16_t sa_family_t;
-typedef uint16_t in_port_t;
-#endif
-
-namespace dht {
-
-std::string print_addr(const sockaddr* sa, socklen_t slen);
-std::string print_addr(const sockaddr_storage& ss, socklen_t sslen);
-
-struct SockAddr : public std::pair<sockaddr_storage, socklen_t> {
-public:
-    SockAddr() : pair<sockaddr_storage, socklen_t>::pair({},0) {}
-    SockAddr(const SockAddr& o) : pair<sockaddr_storage, socklen_t>::pair({},o.second) {
-        std::copy_n((uint8_t*)&o.first, o.second, (uint8_t*)&first);
-    }
-    SockAddr(const sockaddr* sa, socklen_t len) : pair<sockaddr_storage, socklen_t>::pair({},len) {
-        if (len > sizeof(sockaddr_storage))
-            throw std::runtime_error("Socket address length is too large");
-        std::copy_n((uint8_t*)sa, len, (uint8_t*)&first);
-    }
-    SockAddr(const sockaddr_storage& ss, socklen_t len) : SockAddr((const sockaddr*)&ss, len) {}
-
-    bool operator<(const SockAddr& o) const {
-        if (second != o.second)
-            return second < o.second;
-        return std::memcmp((uint8_t*)&first, (uint8_t*)&o.first, second) < 0;
-    }
-
-    bool equals(const SockAddr& o) const {
-        return second == o.second
-            && std::memcmp((uint8_t*)&first, (uint8_t*)&o.first, second) == 0;
-    }
-    SockAddr& operator=(const SockAddr& o) {
-        std::copy_n((const uint8_t*)&o.first, o.second, (uint8_t*)&first);
-        second = o.second;
-        return *this;
-    }
-
-    std::string toString() const {
-        return print_addr(first, second);
-    }
-    sa_family_t getFamily() const { return second > sizeof(sa_family_t) ? first.ss_family : AF_UNSPEC; }
-};
-
-bool operator==(const SockAddr& a, const SockAddr& b);
-
-std::string printAddr(const SockAddr& addr);
-
-}

include/opendht/utils.h

@@ -13,7 +13,8 @@
  *  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 <http://www.gnu.org/licenses/>.
+ *  along with this program; if not, write to the Free Software
+ *  Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301 USA.
  */
 
 #pragma once
@@ -23,6 +24,13 @@
 
 #include <msgpack.hpp>
 
+#ifndef _WIN32
+#include <sys/socket.h>
+#else
+#include <ws2def.h>
+#include <ws2tcpip.h>
+#endif
+
 #include <chrono>
 #include <random>
 #include <functional>
@@ -32,9 +40,15 @@
 
 namespace dht {
 
-using NetId = uint32_t;
+static constexpr unsigned TARGET_NODES {8};
+
+using Address = std::pair<sockaddr_storage, socklen_t>;
 using want_t = int_fast8_t;
 
+std::string print_addr(const sockaddr* sa, socklen_t slen);
+std::string print_addr(const sockaddr_storage& ss, socklen_t sslen);
+std::string printAddr(const Address& addr);
+
 template <typename Key, typename Item, typename Condition>
 void erase_if(std::map<Key, Item>& map, const Condition& condition)
 {
@@ -48,7 +62,7 @@ void erase_if(std::map<Key, Item>& map, const Condition& condition)
 class DhtException : public std::runtime_error {
     public:
         DhtException(const std::string &str = "") :
-            std::runtime_error("DhtException occurred: " + str) {}
+            std::runtime_error("DhtException occured: " + str) {}
 };
 
 
@@ -89,6 +103,45 @@ public:
     }
 };
 
+// Logging related utility functions
+
+/**
+ * Dummy function used to disable logging
+ */
+inline void NOLOG(char const*, va_list) {}
+
+/**
+ * Wrapper for logging methods
+ */
+struct LogMethod {
+    LogMethod() = default;
+
+    template<typename T>
+    LogMethod(T&& t) : func(std::forward<T>(t)) {}
+
+    void operator()(char const* format, ...) const {
+        va_list args;
+        va_start(args, format);
+        func(format, args);
+        va_end(args);
+    }
+
+    void logPrintable(const uint8_t *buf, size_t buflen) const {
+        std::string buf_clean(buflen, '\0');
+        for (size_t i=0; i<buflen; i++)
+            buf_clean[i] = isprint(buf[i]) ? buf[i] : '.';
+        (*this)("%s", buf_clean.c_str());
+    }
+private:
+    std::function<void(char const*, va_list)> func;
+};
+
+struct Logger {
+    LogMethod DEBUG = NOLOG;
+    LogMethod WARN = NOLOG;
+    LogMethod ERR = NOLOG;
+};
+
 // Serialization related definitions and utility functions
 
 using Blob = std::vector<uint8_t>;
@@ -116,6 +169,4 @@ unpackMsg(Blob b) {
 
 msgpack::unpacked unpackMsg(Blob b);
 
-msgpack::object* findMapValue(msgpack::object& map, const std::string& key);
-
 } // namespace dht

include/opendht/value.h

@@ -35,12 +35,10 @@
 #include <functional>
 #include <memory>
 #include <chrono>
-#include <set>
 
 namespace dht {
 
 struct Value;
-struct Query;
 
 /**
  * A storage policy is applied once to every incoming value storage requests.
@@ -70,7 +68,7 @@ using StorePolicy = std::function<bool(InfoHash key, std::shared_ptr<Value>& val
  */
 using EditPolicy = std::function<bool(InfoHash key, const std::shared_ptr<Value>& old_val, std::shared_ptr<Value>& new_val, InfoHash from, const sockaddr* from_addr, socklen_t from_len)>;
 
-static constexpr const size_t MAX_VALUE_SIZE {1024 * 64};
+static constexpr const size_t MAX_VALUE_SIZE {1024 * 14};
 
 struct ValueType {
     typedef uint16_t Id;
@@ -116,27 +114,12 @@ struct ValueType {
  */
 struct Value
 {
-    enum class Field : int {
-        None = 0,
-        Id,        /* Value::id */
-        ValueType, /* Value::type */
-        OwnerPk,   /* Value::owner */
-        SeqNum,    /* Value::seq */
-        UserType,  /* Value::user_type */
-
-        COUNT      /* the total number of fields */
-    };
-
     typedef uint64_t Id;
-    static const constexpr Id INVALID_ID {0};
+    static const Id INVALID_ID {0};
 
     class Filter : public std::function<bool(const Value&)> {
+        using std::function<bool(const Value&)>::function;
     public:
-        Filter() {}
-
-        template<typename Functor>
-        Filter(Functor f) : std::function<bool(const Value&)>::function(f) {}
-
         Filter chain(Filter&& f2) {
             auto f1 = *this;
             return chain(std::move(f1), std::move(f2));
@@ -152,17 +135,14 @@ struct Value
                 return f1(v) and f2(v);
             };
         }
-        static Filter chainAll(std::vector<Filter>&& set) {
-            if (set.empty()) return {};
-            return std::bind([](const Value& v, std::vector<Filter>& s) {
-                for (const auto& f : s)
+        static Filter chain(std::initializer_list<Filter> l) {
+            const std::vector<Filter> list(l.begin(), l.end());
+            return [list](const Value& v){
+                for (const auto& f : list)
                     if (f and not f(v))
                         return false;
                 return true;
-            }, std::placeholders::_1, std::move(set));
-        }
-        static Filter chain(std::initializer_list<Filter> l) {
-            return chainAll(std::vector<Filter>(l.begin(), l.end()));
+            };
         }
         static Filter chainOr(Filter&& f1, Filter&& f2) {
             if (not f1 or not f2) return AllFilter();
@@ -172,10 +152,8 @@ struct Value
         }
     };
 
-    /* Sneaky functions disguised in classes */
-
     static const Filter AllFilter() {
-        return {};
+        return [](const Value&){return true;};
     }
 
     static Filter TypeFilter(const ValueType& t) {
@@ -184,11 +162,6 @@ struct Value
             return v.type == tid;
         };
     }
-    static Filter TypeFilter(const ValueType::Id& tid) {
-        return [tid](const Value& v) {
-            return v.type == tid;
-        };
-    }
 
     static Filter IdFilter(const Id id) {
         return [id](const Value& v) {
@@ -196,34 +169,12 @@ struct Value
         };
     }
 
-    static Filter RecipientFilter(const InfoHash& r) {
+    static Filter recipientFilter(const InfoHash& r) {
         return [r](const Value& v) {
             return v.recipient == r;
         };
     }
 
-    static Filter OwnerFilter(const crypto::PublicKey& pk) {
-        return OwnerFilter(pk.getId());
-    }
-
-    static Filter OwnerFilter(const InfoHash& pkh) {
-        return [pkh](const Value& v) {
-            return v.owner and v.owner->getId() == pkh;
-        };
-    }
-
-    static Filter SeqNumFilter(uint16_t seq_no) {
-        return [seq_no](const Value& v) {
-            return v.seq == seq_no;
-        };
-    }
-
-    static Filter UserTypeFilter(const std::string& ut) {
-        return [ut](const Value& v) {
-            return v.user_type == ut;
-        };
-    }
-
     class SerializableBase
     {
     public:
@@ -297,43 +248,6 @@ struct Value
         return owner and not signature.empty();
     }
 
-    /**
-     * Sign the value using the provided private key.
-     * 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 = std::make_shared<const crypto::PublicKey>(key.getPublicKey());
-        signature = key.sign(getToSign());
-    }
-
-    /**
-     * 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 {
-        return isSigned() and owner->checkSignature(getToSign(), signature);
-    }
-
-    std::shared_ptr<const crypto::PublicKey> getOwner() const {
-        return std::static_pointer_cast<const crypto::PublicKey>(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() {}
 
     Value (Id id) : id(id) {}
@@ -464,51 +378,15 @@ struct Value
         pk.pack(std::string("dat")); msgpack_pack_to_encrypt(pk);
     }
 
-    template <typename Packer>
-    void msgpack_pack_fields(const std::set<Value::Field>& fields, Packer& pk) const
-    {
-        for (const auto& field : fields)
-            switch (field) {
-                case Value::Field::Id:
-                    pk.pack(static_cast<uint64_t>(id));
-                    break;
-                case Value::Field::ValueType:
-                    pk.pack(static_cast<uint64_t>(type));
-                    break;
-                case Value::Field::OwnerPk:
-                    if (owner)
-                        owner->msgpack_pack(pk);
-                    else
-                        InfoHash().msgpack_pack(pk);
-                    break;
-                case Value::Field::SeqNum:
-                    pk.pack(static_cast<uint64_t>(seq));
-                    break;
-                case Value::Field::UserType:
-                    pk.pack(user_type);
-                    break;
-                default:
-                    break;
-            }
-    }
-
     void msgpack_unpack(msgpack::object o);
     void msgpack_unpack_body(const msgpack::object& o);
-    Blob getPacked() const {
-        msgpack::sbuffer buffer;
-        msgpack::packer<msgpack::sbuffer> pk(&buffer);
-        pk.pack(*this);
-        return {buffer.data(), buffer.data()+buffer.size()};
-    }
-
-    void msgpack_unpack_fields(const std::set<Value::Field>& fields, const msgpack::object& o, unsigned offset);
 
     Id id {INVALID_ID};
 
     /**
      * Public key of the signer.
      */
-    std::shared_ptr<const crypto::PublicKey> owner {};
+    std::shared_ptr<crypto::PublicKey> owner {};
 
     /**
      * Hash of the recipient (optional).
@@ -546,360 +424,6 @@ struct Value
 
 using ValuesExport = std::pair<InfoHash, Blob>;
 
-/**
- * @class   FieldValue
- * @brief   Describes a value filter.
- * @details
- * This structure holds the value for a specified field. It's type can either be
- * uint64_t, InfoHash or Blob.
- */
-struct FieldValue
-{
-    FieldValue() {}
-    FieldValue(Value::Field f, uint64_t int_value) : field(f), intValue(int_value) {}
-    FieldValue(Value::Field f, InfoHash hash_value) : field(f), hashValue(hash_value) {}
-    FieldValue(Value::Field f, Blob blob_value) : field(f), blobValue(blob_value) {}
-
-    bool operator==(const FieldValue& fd) const;
-
-    // accessors
-    Value::Field getField() const { return field; }
-    uint64_t getInt() const { return intValue; }
-    InfoHash getHash() const { return hashValue; }
-    Blob getBlob() const { return blobValue; }
-
-    template <typename Packer>
-    void msgpack_pack(Packer& p) const {
-        p.pack_map(2);
-        p.pack(std::string("f")); p.pack(static_cast<uint8_t>(field));
-
-        p.pack(std::string("v"));
-        switch (field) {
-            case Value::Field::Id:
-            case Value::Field::ValueType:
-                p.pack(intValue);
-                break;
-            case Value::Field::OwnerPk:
-                p.pack(hashValue);
-                break;
-            case Value::Field::UserType:
-                p.pack_bin(blobValue.size());
-                p.pack_bin_body((const char*)blobValue.data(), blobValue.size());
-                break;
-            default:
-                throw msgpack::type_error();
-        }
-    }
-
-    void msgpack_unpack(msgpack::object msg) {
-        hashValue = {};
-        blobValue.clear();
-
-        if (auto f = findMapValue(msg, "f"))
-            field = (Value::Field)f->as<unsigned>();
-        else
-            throw msgpack::type_error();
-
-        auto v = findMapValue(msg, "v");
-        if (not v)
-            throw msgpack::type_error();
-        else
-            switch (field) {
-                case Value::Field::Id:
-                case Value::Field::ValueType:
-                    intValue = v->as<decltype(intValue)>();
-                    break;
-                case Value::Field::OwnerPk:
-                    hashValue = v->as<decltype(hashValue)>();
-                    break;
-                case Value::Field::UserType:
-                    blobValue = unpackBlob(*v);
-                    break;
-                default:
-                    throw msgpack::type_error();
-            }
-    }
-
-    Value::Filter getLocalFilter() const;
-
-private:
-    Value::Field field {Value::Field::None};
-    // three possible value types
-    uint64_t intValue {};
-    InfoHash hashValue {};
-    Blob blobValue {};
-};
-
-
-/**
- * @struct  FieldSelectorDescription
- * @brief   Describes a selection.
- * @details
- * This is meant to narrow data to a set of specified fields. This structure is
- * used to construct a Select structure.
- */
-struct FieldSelectorDescription
-{
-    FieldSelectorDescription() {}
-    FieldSelectorDescription(Value::Field f) : field(f) {}
-
-    Value::Field getField() const { return field; }
-
-    bool operator==(const FieldSelectorDescription& fd) const { return field == fd.field; }
-
-    template <typename Packer>
-    void msgpack_pack(Packer& p) const { p.pack(static_cast<uint8_t>(field)); }
-    void msgpack_unpack(msgpack::object msg) { field = static_cast<Value::Field>(msg.as<int>()); }
-private:
-    Value::Field field {Value::Field::None};
-};
-
-/**
- * @class   Select
- * @brief   Serializable Value field selection.
- * @details
- * This is a container for a list of FieldSelectorDescription instances. It
- * describes a complete SELECT query for dht::Value.
- */
-struct Select
-{
-    Select() { }
-    Select(const std::string& q_str);
-
-    bool isSatisfiedBy(const Select& os) const;
-
-    /**
-     * Selects a field of type Value::Field.
-     *
-     * @param field  the field to require.
-     *
-     * @return the resulting Select instance.
-     */
-    Select& field(Value::Field field) {
-        fieldSelection_.emplace_back(field);
-        return *this;
-    }
-
-    /**
-     * Computes the set of selected fields based on previous require* calls.
-     *
-     * @return the set of fields.
-     */
-    std::set<Value::Field> getSelection() const {
-        std::set<Value::Field> fields {};
-        for (const auto& f : fieldSelection_) {
-            fields.insert(f.getField());
-        }
-        return fields;
-    }
-
-    template <typename Packer>
-    void msgpack_pack(Packer& pk) const { pk.pack(fieldSelection_); }
-    void msgpack_unpack(const msgpack::object& o) {
-        fieldSelection_.clear();
-        fieldSelection_ = o.as<decltype(fieldSelection_)>();
-    }
-
-    friend std::ostream& operator<<(std::ostream& s, const dht::Select& q);
-private:
-    std::vector<FieldSelectorDescription> fieldSelection_ {};
-};
-
-/**
- * @class   Where
- * @brief   Serializable dht::Value filter.
- * @details
- * This is container for a list of FieldValue instances. It describes a
- * complete WHERE query for dht::Value.
- */
-struct Where
-{
-    Where() { }
-    Where(const std::string& q_str);
-
-    bool isSatisfiedBy(const Where& where) const;
-
-    /**
-     * Adds restriction on Value::Id based on the id argument.
-     *
-     * @param id  the id.
-     *
-     * @return the resulting Where instance.
-     */
-    Where& id(Value::Id id) {
-        filters_.emplace_back(Value::Field::Id, id);
-        return *this;
-    }
-
-    /**
-     * Adds restriction on Value::ValueType based on the type argument.
-     *
-     * @param type  the value type.
-     *
-     * @return the resulting Where instance.
-     */
-    Where& valueType(ValueType::Id type) {
-        filters_.emplace_back(Value::Field::ValueType, type);
-        return *this;
-    }
-
-    /**
-     * Adds restriction on Value::OwnerPk based on the owner_pk_hash argument.
-     *
-     * @param owner_pk_hash  the owner public key fingerprint.
-     *
-     * @return the resulting Where instance.
-     */
-    Where& owner(InfoHash owner_pk_hash) {
-        filters_.emplace_back(Value::Field::OwnerPk, owner_pk_hash);
-        return *this;
-    }
-
-    /**
-     * Adds restriction on Value::OwnerPk based on the owner_pk_hash argument.
-     *
-     * @param owner_pk_hash  the owner public key fingerprint.
-     *
-     * @return the resulting Where instance.
-     */
-    Where& seq(uint16_t seq_no) {
-        filters_.emplace_back(Value::Field::SeqNum, seq_no);
-        return *this;
-    }
-
-    /**
-     * Adds restriction on Value::UserType based on the user_type argument.
-     *
-     * @param user_type  the user type.
-     *
-     * @return the resulting Where instance.
-     */
-    Where& userType(std::string user_type) {
-        filters_.emplace_back(Value::Field::UserType, Blob {user_type.begin(), user_type.end()});
-        return *this;
-    }
-
-    /**
-     * Computes the Value::Filter based on the list of field value set.
-     *
-     * @return the resulting Value::Filter.
-     */
-    Value::Filter getFilter() const {
-        if (filters_.empty()) return {};
-        std::vector<Value::Filter> fset;
-        fset.reserve(filters_.size());
-        for (const auto& f : filters_) {
-            if (auto lf = f.getLocalFilter())
-                fset.emplace_back(std::move(lf));
-        }
-        return Value::Filter::chainAll(std::move(fset));
-    }
-
-    template <typename Packer>
-    void msgpack_pack(Packer& pk) const { pk.pack(filters_); }
-    void msgpack_unpack(const msgpack::object& o) {
-        filters_.clear();
-        filters_ = o.as<decltype(filters_)>();
-    }
-
-    friend std::ostream& operator<<(std::ostream& s, const dht::Where& q);
-
-private:
-    std::vector<FieldValue> filters_;
-};
-
-/**
- * @class   Query
- * @brief   Describes a query destined to another peer.
- * @details
- * This class describes the list of filters on field values and the field
- * itselves to include in the peer response to a GET operation. See
- * FieldValue.
- */
-struct Query
-{
-    static const std::string QUERY_PARSE_ERROR;
-
-    Query(Select s = {}, Where w = {}, bool none = false) : select(s), where(w), none(none) { };
-
-    /**
-     * Initializes a query based on a SQL-ish formatted string. The abstract
-     * form of such a string is the following:
-     *
-     *  [SELECT $field$ [WHERE $field$=$value$]]
-     *
-     *  where
-     *
-     *  - $field$ = *|id|value_type|owner_pk|user_type
-     *  - $value$ = $string$|$integer$
-     *  - $string$: a simple string WITHOUT SPACES.
-     *  - $integer$: a simple integer.
-     */
-    Query(std::string q_str) {
-        auto pos_W = q_str.find("WHERE");
-        auto pos_w = q_str.find("where");
-        auto pos = std::min(pos_W != std::string::npos ? pos_W : q_str.size(),
-                            pos_w != std::string::npos ? pos_w : q_str.size());
-        select = q_str.substr(0, pos);
-        where = q_str.substr(pos, q_str.size()-pos);
-    }
-
-    /**
-     * Tell if the query is satisfied by another query.
-     */
-    bool isSatisfiedBy(const Query& q) const;
-
-    template <typename Packer>
-    void msgpack_pack(Packer& pk) const {
-        pk.pack_map(2);
-        pk.pack(std::string("s")); pk.pack(select); /* packing field selectors */
-        pk.pack(std::string("w")); pk.pack(where);  /* packing filters */
-    }
-
-    void msgpack_unpack(const msgpack::object& o);
-
-    std::string toString() {
-        std::stringstream ss;
-        ss << *this;
-        return ss.str();
-    }
-
-    friend std::ostream& operator<<(std::ostream& s, const dht::Query& q) {
-        return s << "Query[" << q.select << " " << q.where << "]";
-    }
-
-    Select select {};
-    Where where {};
-    bool none {false}; /* When true, any query satisfies this. */
-};
-
-/*!
- * @class   FieldValueIndex
- * @brief   An index for field values.
- * @details
- * This structures is meant to manipulate a subset of fields normally contained
- * in Value.
- */
-struct FieldValueIndex {
-    FieldValueIndex() {}
-    FieldValueIndex(const Value& v, Select s = {});
-    /**
-     * Tells if all the fields of this are contained in the other
-     * FieldValueIndex with the same value.
-     *
-     * @param other  The other FieldValueIndex instance.
-     */
-    bool containedIn(const FieldValueIndex& other) const;
-
-    friend std::ostream& operator<<(std::ostream& os, const FieldValueIndex& fvi);
-
-    void msgpack_unpack_fields(const std::set<Value::Field>& fields,
-            const msgpack::object& o,
-            unsigned offset);
-
-    std::map<Value::Field, FieldValue> index {};
-};
-
 template <typename T,
           typename std::enable_if<std::is_base_of<Value::SerializableBase, T>::value, T>::type* = nullptr>
 Value::Filter

opendht.pc.in

@@ -6,6 +6,4 @@ Name: OpenDHT
 Description: C++ Distributed Hash Table library
 Version: @VERSION@
 Libs: -L${libdir} -lopendht
-Libs.private: -lpthread
-Requires.private: gnutls >= 3.1
 Cflags: -I${includedir}

python/Makefile.am

@@ -1,9 +1,5 @@
 if USE_CYTHON
 
-noinst_HEADERS = \
-	opendht.pyx \
-	opendht_cpp.pxd
-
 PYTHON_INSTALL_RECORD = $(builddir)/install_record.txt
 
 pybuild.stamp:

python/opendht.pyx

@@ -31,7 +31,6 @@ from libcpp.map cimport map as map
 from libcpp cimport bool
 from libcpp.utility cimport pair
 from libcpp.string cimport string
-from libcpp.memory cimport shared_ptr
 
 from cython.parallel import parallel, prange
 from cython.operator cimport dereference as deref, preincrement as inc, predecrement as dec
@@ -57,13 +56,13 @@ cdef inline void shutdown_callback(void* user_data) with gil:
         cbs['shutdown']()
     ref.Py_DECREF(cbs)
 
-cdef inline bool get_callback(shared_ptr[cpp.Value] value, void *user_data) with gil:
+cdef inline bool get_callback(cpp.shared_ptr[cpp.Value] value, void *user_data) with gil:
     cb = (<object>user_data)['get']
     pv = Value()
     pv._value = value
     return cb(pv)
 
-cdef inline void done_callback(bool done, cpp.vector[shared_ptr[cpp.Node]]* nodes, void *user_data) with gil:
+cdef inline void done_callback(bool done, cpp.vector[cpp.shared_ptr[cpp.Node]]* nodes, void *user_data) with gil:
     node_ids = []
     for n in deref(nodes):
         h = NodeEntry()
@@ -126,7 +125,7 @@ cdef class InfoHash(_WithID):
         return h
 
 cdef class Node(_WithID):
-    cdef shared_ptr[cpp.Node] _node
+    cdef cpp.shared_ptr[cpp.Node] _node
     def getId(self):
         h = InfoHash()
         h._infohash = self._node.get().getId()
@@ -137,7 +136,7 @@ cdef class Node(_WithID):
         return self._node.get().isExpired()
 
 cdef class NodeEntry(_WithID):
-    cdef cpp.pair[cpp.InfoHash, shared_ptr[cpp.Node]] _v
+    cdef cpp.pair[cpp.InfoHash, cpp.shared_ptr[cpp.Node]] _v
     def getId(self):
         h = InfoHash()
         h._infohash = self._v.first
@@ -148,7 +147,7 @@ cdef class NodeEntry(_WithID):
         return n
 
 cdef class Value(object):
-    cdef shared_ptr[cpp.Value] _value
+    cdef cpp.shared_ptr[cpp.Value] _value
     def __init__(self, bytes val=b''):
         self._value.reset(new cpp.Value(val, len(val)))
     def __str__(self):
@@ -180,8 +179,8 @@ cdef class Value(object):
             return self._value.get().size()
 
 cdef class NodeSetIter(object):
-    cdef map[cpp.InfoHash, shared_ptr[cpp.Node]]* _nodes
-    cdef map[cpp.InfoHash, shared_ptr[cpp.Node]].iterator _curIter
+    cdef map[cpp.InfoHash, cpp.shared_ptr[cpp.Node]]* _nodes
+    cdef map[cpp.InfoHash, cpp.shared_ptr[cpp.Node]].iterator _curIter
     def __init__(self, NodeSet s):
         self._nodes = &s._nodes
         self._curIter = self._nodes.begin()
@@ -194,7 +193,7 @@ cdef class NodeSetIter(object):
         return h
 
 cdef class NodeSet(object):
-    cdef map[cpp.InfoHash, shared_ptr[cpp.Node]] _nodes
+    cdef map[cpp.InfoHash, cpp.shared_ptr[cpp.Node]] _nodes
     def size(self):
         return self._nodes.size()
     def insert(self, NodeEntry l):
@@ -216,7 +215,7 @@ cdef class NodeSet(object):
         return h
     def __str__(self):
         s = ''
-        cdef map[cpp.InfoHash, shared_ptr[cpp.Node]].iterator it = self._nodes.begin()
+        cdef map[cpp.InfoHash, cpp.shared_ptr[cpp.Node]].iterator it = self._nodes.begin()
         while it != self._nodes.end():
             s += deref(it).first.toString().decode() + ' ' + deref(it).second.get().getAddrStr().decode() + '\n'
             inc(it)
@@ -232,7 +231,7 @@ cdef class PublicKey(_WithID):
         return h
 
 cdef class Certificate(_WithID):
-    cdef shared_ptr[cpp.Certificate] _cert
+    cdef cpp.shared_ptr[cpp.Certificate] _cert
     def getId(self):
         h = InfoHash()
         h._infohash = self._cert.get().getId()

python/tools/README.md

@@ -1,45 +0,0 @@
-# Benchmark
-
-The `benchmark.py` script is used for testing OpenDHT in various cases. If you
-run `benchmark.py --help`, you should find the following text:
-
-    optional arguments:
-      -h, --help            show this help message and exit
-      --performance         Launches performance benchmark test. Available args
-                            for "-t" are: gets.
-      --data-persistence    Launches data persistence benchmark test. Available
-                            args for "-t" are: delete, replace, mult_time.
-                            Available args for "-o" are : dump_str_log,
-                            keep_alive, trigger, traffic_plot, op_plot. Use "-m"
-                            to specify the number of producers on the DHT.Use "-e"
-                            to specify the number of values to put on the DHT.
-
-These options specify the feature to be tested. Each feature has its own tests.
-You specify the test by using `-t` flag (see `benchmark.py --help` for full
-help).
-
-## Python dependencies
-
-- pyroute2 >=0.3.14
-- matplotlib
-- GeoIP (used by `scanner.py` for drawing map of the world)
-- ipaddress
-- netifaces
-- networkx
-- numpy
-
-## Usage
-
-Before running the script, you have to build and install OpenDHT and its cython
-wrapper (`cython3` has to be installed) on the system so that it can be found by
-the benchmark script.
-
-    $ cd $OPENDHT_SRC_DIR
-    $ ./autogen.sh
-    $ ./configure
-    $ make && sudo make install
-
-Then, you can use the script like so:
-
-    $ cd $OPENDHT_SRC_DIR/python/tools/
-    $ python3 benchmark.py --performance -t gets -n 2048

python/tools/benchmark.py

@@ -66,11 +66,7 @@ class WorkBench():
 
     def create_virtual_net(self):
         if self.virtual_locs > 1:
-            cmd = ["python3", os.path.abspath(virtual_network_builder.__file__),
-                    "-i", self.ifname,
-                    "-n", str(self.clusters),
-                    '-l', str(self.loss),
-                    '-d', str(self.delay)]
+            cmd = ["python3", os.path.abspath(virtual_network_builder.__file__), "-i", self.ifname, "-n", str(self.clusters), '-l', str(self.loss), '-d', str(self.delay)]
             if not self.disable_ipv4:
                 cmd.append('-4')
             if not self.disable_ipv6:
@@ -94,17 +90,10 @@ class WorkBench():
                     cmd.extend(['-b', self.local_bootstrap.ip4])
                 if not self.disable_ipv6 and self.local_bootstrap.ip6:
                     cmd.extend(['-b6', self.local_bootstrap.ip6])
-            lock = threading.Condition()
-            def dcb(success):
-                nonlocal lock
-                if not success:
-                    DhtNetwork.Log.err("Failed to initialize network...")
-                with lock:
-                    lock.notify()
-            with lock:
-                self.procs[i] = DhtNetworkSubProcess('node'+str(i), cmd, stdin=subprocess.PIPE, stdout=subprocess.PIPE)
-                self.procs[i].sendPing(done_cb=dcb)
-                lock.wait()
+            self.procs[i] = DhtNetworkSubProcess('node'+str(i), cmd, stdin=subprocess.PIPE, stdout=subprocess.PIPE)
+            while DhtNetworkSubProcess.NOTIFY_TOKEN not in self.procs[i].getline():
+                # waiting for process to spawn
+                time.sleep(0.5)
         else:
             raise Exception('First create bootstrap.')
 
@@ -177,7 +166,7 @@ if __name__ == '__main__':
             help='Launches data persistence benchmark test. '\
                  'Available args for "-t" are: delete, replace, mult_time. '\
                  'Available args for "-o" are : dump_str_log, keep_alive, trigger, traffic_plot, op_plot. '\
-                 'Use "-m" to specify the number of producers on the DHT. '\
+                 'Use "-m" to specify the number of producers on the DHT.'\
                  'Use "-e" to specify the number of values to put on the DHT.')
 
     args = parser.parse_args()

python/tools/dht/network.py

@@ -25,13 +25,11 @@ import time
 import threading
 import queue
 import re
-import traceback
 
 import ipaddress
 import netifaces
 import numpy as np
 from pyroute2.netns.process.proxy import NSPopen
-import msgpack
 
 from opendht import *
 
@@ -39,6 +37,8 @@ from opendht import *
 # useful functions
 b_space_join = lambda *l: b' '.join(map(bytes, l))
 
+# TODO: find where token "notifyend" gets printed... Or switch to MSGPACK for
+# serialisation of packets between both processes.
 class DhtNetworkSubProcess(NSPopen):
     """
     Handles communication with DhtNetwork sub process.
@@ -47,21 +47,20 @@ class DhtNetworkSubProcess(NSPopen):
     process' stdout until it finds 'DhtNetworkSubProcess.NOTIFY_TOKEN' token,
     therefor, waits for the sub process to spawn.
     """
-    # Packet types
-    REQUEST = 'DhtNetworkSubProcess.request'
-    ANSWER = 'DhtNetworkSubProcess.answer'
-    OUT = 'DhtNetworkSubProcess.out'
-
     # requests
-    PING_REQ                  = "p"
-    NODE_PUT_REQ              = "np"   # "np <hash> <value>"
-    NEW_NODE_REQ              = "nn"   # "nn"
-    REMOVE_NODE_REQ           = "rn"   # "rn <id0>[ <id1>[ id2[...]]]"
-    SHUTDOWN_NODE_REQ         = "sdn"  # "sdn <id0>[ <id1>[ id2[...]]]"
-    SHUTDOWN_REPLACE_NODE_REQ = "sdrn" # "sdn <id0>[ <id1>[ id2[...]]]"
-    SHUTDOWN_CLUSTER_REQ      = "sdc"  # "sdc"
-    DUMP_STORAGE_REQ          = "strl" # "strl"
-    MESSAGE_STATS             = "gms"  # "gms"
+    NODE_PUT_REQ              = b"np"
+    NEW_NODE_REQ              = b"nn"
+    REMOVE_NODE_REQ           = b"rn"
+    SHUTDOWN_NODE_REQ         = b"sdn"
+    SHUTDOWN_REPLACE_NODE_REQ = b'sdrn'
+    SHUTDOWN_CLUSTER_REQ      = b"sdc"
+    DUMP_STORAGE_REQ          = b"strl"
+    MESSAGE_STATS             = b"gms"
+
+
+    # tokens
+    NOTIFY_TOKEN     = 'notify'
+    NOTIFY_END_TOKEN = 'notifyend'
 
     def __init__(self, ns, cmd, quit=False, **kwargs):
         super(DhtNetworkSubProcess, self).__init__(ns, cmd, **kwargs)
@@ -71,11 +70,11 @@ class DhtNetworkSubProcess(NSPopen):
         self._quit = quit
         self._lock = threading.Condition()
         self._in_queue = queue.Queue()
-        self._callbacks = {}
-        self._tid = 0
+        self._out_queue = queue.Queue()
 
         # starting thread
         self._thread = threading.Thread(target=self._communicate)
+        self._thread.daemon = True
         self._thread.start()
 
     def __repr__(self):
@@ -93,38 +92,42 @@ class DhtNetworkSubProcess(NSPopen):
         """
         Communication thread. This reads and writes to the sub process.
         """
+        ENCODING = 'utf-8'
         sleep_time = 0.1
+        stdin_line, stdout_line = '', ''
+
+        # first read of process living. Expecting NOTIFY_TOKEN
+        while DhtNetworkSubProcess.NOTIFY_TOKEN not in stdout_line:
+            stdout_line = self.stdout.readline().decode()
+            time.sleep(sleep_time)
+
+        with self._lock:
+            self._out_queue.put(stdout_line)
 
         while not self._quit:
             with self._lock:
                 try:
-                    packet = self._in_queue.get_nowait()
+                    stdin_line = self._in_queue.get_nowait()
 
                     # sending data to sub process
-                    self.stdin.write(packet)
+                    self.stdin.write(stdin_line if isinstance(stdin_line, bytes) else
+                            bytes(str(stdin_line), encoding=ENCODING))
                     self.stdin.flush()
                 except queue.Empty:
-                    pass
+                    #waiting for next stdin req to send
+                    self._lock.wait(timeout=sleep_time)
 
-                # reading from sub process
-                out_string = ''
-                for p in msgpack.Unpacker(self.stdout):
-                    if isinstance(p, dict):
-                        self._process_packet(p)
-                    else:
-                        # Some non-msgpack data could slip into the stream. We
-                        # have to treat those as characters.
-                        out_string += chr(p)
-                if out_string:
-                    print(out_string)
-
-                #waiting for next stdin req to send
-                self._lock.wait(timeout=sleep_time)
+            # reading response from sub process
+            for stdout_line in iter(self.stdout.readline, b''):
+                stdout_line = stdout_line.decode().replace('\n', '')
+                if stdout_line:
+                    with self._lock:
+                        self._out_queue.put(stdout_line)
 
         with self._lock:
             self._lock.notify()
 
-    def _stop_communicating(self):
+    def stop_communicating(self):
         """
         Stops the I/O thread from communicating with the subprocess.
         """
@@ -139,12 +142,12 @@ class DhtNetworkSubProcess(NSPopen):
         Notifies thread and sub process to terminate. This is blocking call
         until the sub process finishes.
         """
-        self._stop_communicating()
+        self.stop_communicating()
         self.send_signal(signal.SIGINT);
         self.wait()
         self.release()
 
-    def _send(self, msg):
+    def send(self, msg):
         """
         Send data to sub process.
         """
@@ -152,76 +155,83 @@ class DhtNetworkSubProcess(NSPopen):
             self._in_queue.put(msg)
             self._lock.notify()
 
-    def _process_packet(self, p):
+    def getline(self):
         """
-        Process msgpack packet received from
+        Read line from sub process.
+
+        @return:  A line on sub process' stdout.
+        @rtype :  str
         """
-        if not b'tid' in p:
-            DhtNetwork.Log.err('Bad packet...')
-        try:
-            self._callbacks[p[b'tid']](p)
-        except KeyError:
-            DhtNetwork.Log.err('Unknown tid...')
+        line = ''
+        with self._lock:
+            try:
+                line = self._out_queue.get_nowait()
+            except queue.Empty:
+                pass
+        return line
 
+    def getlinesUntilNotify(self, answer_cb=None):
+        """
+        Reads the stdout queue until a proper notification is given by the sub
+        process.
 
-    def _sendRequest(self, request, tid, done_cb):
+        @param answer_cb: Callback to call when an answer is given after notify.
+                          The function takes a list of strings as argument.
+        @type  answer_cb:  function
+        """
+        notified = False
+        answer = []
+        while True:
+            out = self.getline()
+            if out.split(' ')[0] == DhtNetworkSubProcess.NOTIFY_TOKEN:
+                notified = True
+            elif notified and out.split(' ')[0] == DhtNetworkSubProcess.NOTIFY_END_TOKEN:
+                if answer_cb:
+                    answer_cb(answer)
+                break
+            elif notified:
+                answer.append(out)
+            elif out:
+                yield out
+            else:
+                time.sleep(0.1)
+
+    def _sendRequest(self, request, answer_cb=None):
         """
         Sends a request to the sub network and wait for output.
 
         @param request: The serialized request.
-        @type  request: Msgpack object
+        @type  request: bytes
         """
-        self._callbacks[tid] = done_cb
-        self._send(request)
+        self.send(request + b'\n')
+        for line in self.getlinesUntilNotify(answer_cb=answer_cb):
+            DhtNetwork.log(line)
 
-    def sendPing(self, done_cb=None):
-        """Sends a ping request to the DhtNetworkSubProcess.
-
-        @param done_cb: The callback to be executed when we get a response. This
-                        function takes a boolean "success" as parameter.
-        @type  done_cb: Function
-        """
-        self._tid += 1
-        def dcb(packet):
-            try:
-                done_cb(packet[b'success'])
-            except KeyError:
-                done_cb(False)
-        self._sendRequest(msgpack.packb({
-            DhtNetworkSubProcess.REQUEST : True,
-            'tid' : self._tid,
-            'req' : DhtNetworkSubProcess.PING_REQ
-        }), self._tid, dcb)
-
-    def sendGetMessageStats(self, done_cb=None):
+    def sendGetMessageStats(self):
         """
         Sends DhtNetwork sub process statistics request about nodes messages
         sent.
 
-        @param done_cb: A function taking as parameter the returned list of
-                        stats.
-        @type  done_cb: function
-
         @return: A list [num_nodes, ping, find, get, put, listen].
         @rtype : list
         """
-        self._tid += 1
-        def dcb(packet):
-            nonlocal done_cb
-            if not done_cb:
-                return
-            try:
-                stats = packet[b'stats']
-                done_cb([] if not isinstance(stats, list) else done_cb(stats))
-            except KeyError:
-                done_cb([])
-        self._sendRequest(msgpack.packb({
-            DhtNetworkSubProcess.REQUEST : True,
-            'tid' : self._tid,
-            'req' : DhtNetworkSubProcess.MESSAGE_STATS
-        }), self._tid, dcb)
+        stats = []
+        def cb(answer):
+            """
+            Callback fed to getlinesUntilNotify made to recover answer from the
+            DhtNetwork sub process.
 
-    def sendClusterPutRequest(self, _hash, value, done_cb=None):
+            @param answer: the list of lines answered by the sub process.
+            @type  answer: function
+            """
+            nonlocal stats
+            if answer:
+                stats = [int(v) for v in re.findall("[0-9]+", answer.pop())]
+
+        self._sendRequest(DhtNetworkSubProcess.MESSAGE_STATS, answer_cb=cb)
+        return stats
+
+    def sendNodePutRequest(self, _hash, value):
         """
         Sends a put operation request.
 
@@ -229,89 +239,40 @@ class DhtNetworkSubProcess(NSPopen):
         @type  _hash: bytes.
         @param value: the value.
         @type  value: bytes.
-        @param done_cb: A function taking as parameter a boolean "success".
-        @type  done_cb: function
         """
-        self._tid += 1
-        def dcb(packet):
-            nonlocal done_cb
-            if not done_cb:
-                return
-            try:
-                done_cb(packet[b'success'])
-            except KeyError:
-                done_cb(False)
-        self._sendRequest(msgpack.packb({
-            DhtNetworkSubProcess.REQUEST : True,
-            'tid' : self._tid,
-            'req'   : DhtNetworkSubProcess.NODE_PUT_REQ,
-            'hash'  : _hash,
-            'value' : value
-        }), self._tid, dcb)
+        self._sendRequest(b_space_join(DhtNetworkSubProcess.NODE_PUT_REQ, _hash,
+            value))
 
-    def sendClusterRequest(self, request, ids=[], done_cb=None):
+    def sendNodesRequest(self, request, ids=b''):
         """
         Send request to a list of nodes or the whole cluster.
 
         @param request: The request. Possible values are:
+            DhtNetworkSubProcess.NODE_PUT_REQ
             DhtNetworkSubProcess.REMOVE_NODE_REQ
             DhtNetworkSubProcess.SHUTDOWN_NODE_REQ
             DhtNetworkSubProcess.SHUTDOWN_REPLACE_NODE_REQ
             DhtNetworkSubProcess.SHUTDOWN_CLUSTER_REQ
             DhtNetworkSubProcess.DUMP_STORAGE_REQ
+            DhtNetworkSubProcess.MESSAGE_STATS
         @type request: bytes
         @param ids: The list of ids concerned by the request.
         @type  ids: list
         """
-        self._tid += 1
-        def dcb(packet):
-            nonlocal done_cb
-            if not done_cb:
-                return
-            try:
-                done_cb(packet[b'success'])
-            except KeyError:
-                done_cb(False)
-        self._sendRequest(msgpack.packb({
-            DhtNetworkSubProcess.REQUEST : True,
-            'tid' : self._tid,
-            'req' : request,
-            'ids' : ids
-        }), self._tid, dcb)
-
+        self._sendRequest(b_space_join(request, b_space_join(*ids)))
 
 class DhtNetwork(object):
     nodes = []
 
-    class Log(object):
+    @staticmethod
+    def log(*to_print):
         BOLD   = "\033[1m"
         NORMAL = "\033[0m"
-        WHITE  = "\033[97m"
-        RED    = "\033[31m"
-        YELLOW = "\033[33m"
-
-        @staticmethod
-        def _log_with_color(*to_print, color=None):
-            color = color if color else DhtNetwork.Log.WHITE
-            print('%s%s[DhtNetwork-%s]%s%s' %
-                    (DhtNetwork.Log.BOLD, color, DhtNetwork.iface, DhtNetwork.Log.NORMAL, color),
-                    *to_print, DhtNetwork.Log.NORMAL, file=sys.stderr)
-
-        @staticmethod
-        def log(*to_print):
-            DhtNetwork.Log._log_with_color(*to_print, color=DhtNetwork.Log.WHITE)
-
-        @staticmethod
-        def warn(*to_print):
-            DhtNetwork.Log._log_with_color(*to_print, color=DhtNetwork.Log.YELLOW)
-
-        @staticmethod
-        def err(*to_print):
-            DhtNetwork.Log._log_with_color(*to_print, color=DhtNetwork.Log.RED)
+        print('%s[DhtNetwork-%s]%s' % (BOLD, DhtNetwork.iface, NORMAL), ':' , *to_print, file=sys.stderr)
 
     @staticmethod
     def run_node(ip4, ip6, p, bootstrap=[], is_bootstrap=False):
-        DhtNetwork.Log.log("run_node", ip4, ip6, p, bootstrap)
+        DhtNetwork.log("run_node", ip4, ip6, p, bootstrap)
         n = DhtRunner()
         n.run(ipv4=ip4 if ip4 else "", ipv6=ip6 if ip6 else "", port=p, is_bootstrap=is_bootstrap)
         for b in bootstrap:
@@ -336,7 +297,7 @@ class DhtNetwork(object):
         self.ip6 = ip6 if ip6 else ips[1]
         self.bootstrap = bootstrap
         if first_bootstrap:
-            DhtNetwork.Log.log("Starting bootstrap node")
+            DhtNetwork.log("Starting bootstrap node")
             self.nodes.append(DhtNetwork.run_node(self.ip4, self.ip6, self.port, self.bootstrap, is_bootstrap=True))
             self.bootstrap = [(self.ip4, str(self.port))]
             self.port += 1
@@ -348,9 +309,6 @@ class DhtNetwork(object):
         return self.nodes[0][1]
 
     def get(self, i=None):
-        if not self.nodes:
-            return None
-
         if i is None:
             l = list(self.nodes)
             random.shuffle(l)
@@ -363,13 +321,14 @@ class DhtNetwork(object):
             for n in self.nodes:
                 if n[1].getNodeId() == id:
                     return n
-        return None
+        else:
+            return None
 
     def launch_node(self):
         n = DhtNetwork.run_node(self.ip4, self.ip6, self.port, self.bootstrap)
         self.nodes.append(n)
         if not self.bootstrap:
-            DhtNetwork.Log.log("Using fallback bootstrap", self.ip4, self.port)
+            DhtNetwork.log("Using fallback bootstrap", self.ip4, self.port)
             self.bootstrap = [(self.ip4, str(self.port))]
         self.port += 1
         return n
@@ -387,25 +346,25 @@ class DhtNetwork(object):
         lock = threading.Condition()
         def shutdown_cb():
             nonlocal lock
-            DhtNetwork.Log.log('Done.')
+            DhtNetwork.log('Done.')
             with lock:
                 lock.notify()
 
         if not self.nodes:
             return
-        elif id:
+        elif id is not None:
             n = self.getNodeInfoById(id)
             if n:
                 if shutdown:
                     with lock:
-                        DhtNetwork.Log.log('Waiting for node to shutdown... ')
+                        DhtNetwork.log('Waiting for node to shutdown... ')
                         n[1].shutdown(shutdown_cb)
                         lock.wait()
                     if last_msg_stats:
                         last_msg_stats.append(self.getMessageStats())
                 n[1].join()
                 self.nodes.remove(n)
-                DhtNetwork.Log.log(id, 'deleted !')
+                DhtNetwork.log(id, 'deleted !')
                 return True
             else:
                 return False
@@ -426,11 +385,11 @@ class DhtNetwork(object):
         if n == l:
             return
         if n > l:
-            DhtNetwork.Log.log("Launching", n-l, "nodes", self.ip4, self.ip6)
+            DhtNetwork.log("Launching", n-l, "nodes", self.ip4, self.ip6)
             for i in range(l, n):
                 self.launch_node()
         else:
-            DhtNetwork.Log.log("Ending", l-n, "nodes", self.ip4, self.ip6)
+            DhtNetwork.log("Ending", l-n, "nodes", self.ip4, self.ip6)
             #random.shuffle(self.nodes)
             for i in range(n, l):
                 self.end_node()
@@ -450,39 +409,30 @@ if __name__ == '__main__':
     lock = threading.Condition()
     quit = False
 
-    def send_msgpack_packet(packet):
-        sys.stdout.buffer.write(packet)
-        sys.stdout.buffer.flush()
+    def notify_benchmark(answer=[]):
+        sys.stdout.write('%s\n' % DhtNetworkSubProcess.NOTIFY_TOKEN)
+        for line in answer:
+            sys.stdout.write(str(line)+'\n')
+        sys.stdout.write('%s\n' % DhtNetworkSubProcess.NOTIFY_END_TOKEN)
+        sys.stdout.flush()
 
-    def notify_benchmark(packet, success):
-        """Notifies the benchmark when an operation has been completed.
-
-        @param success: If the operation has been successful
-        @type  success: boolean
-        @param packet: The packet we are providing an answer for.
-        @type  packet: dict
-        """
-        send_msgpack_packet(msgpack.packb({
-            DhtNetworkSubProcess.ANSWER : True,
-            'tid'     : packet[b'tid'],
-            'success' : success
-        }))
-
-    def send_stats(packet, stats):
-        send_msgpack_packet(msgpack.packb({
-            DhtNetworkSubProcess.ANSWER : True,
-            'tid'   : packet[b'tid'],
-            'stats' : stats
-        }))
-
-    def listen_to_mother_nature(q):
+    def listen_to_mother_nature(stdin, q):
         global quit
+
+        def parse_req(req):
+            split_req = req.split(' ')
+
+            op = split_req[0]
+            hashes = [this_hash.replace('\n', '').encode() for this_hash in split_req[1:]]
+
+            return (op, hashes)
+
         while not quit:
-            for p in msgpack.Unpacker(sys.stdin.buffer.raw):
-                if isinstance(p, dict) and DhtNetworkSubProcess.REQUEST.encode() in p:
-                    with lock:
-                        q.put(p)
-                        lock.notify()
+            req = stdin.readline()
+            parsed_req = parse_req(req)
+            q.put(parsed_req)
+            with lock:
+                lock.notify()
 
     def handler(signum, frame):
         global quit
@@ -516,52 +466,48 @@ if __name__ == '__main__':
         net.resize(args.node_num)
 
         q = queue.Queue()
-        t = threading.Thread(target=listen_to_mother_nature, args=tuple([q]))
+        t = threading.Thread(target=listen_to_mother_nature, args=(sys.stdin, q))
         t.daemon = True
         t.start()
 
+        notify_benchmark()
+
         msg_stats = []
 
         with lock:
             while not quit:
+                lock.wait()
                 try:
-                    packet = q.get_nowait()
+                    req,req_args = q.get_nowait()
                 except queue.Empty:
-                    lock.wait()
+                    pass
                 else:
-                    NODE_PUT_REQ              = DhtNetworkSubProcess.NODE_PUT_REQ
-                    NEW_NODE_REQ              = DhtNetworkSubProcess.NEW_NODE_REQ
-                    REMOVE_NODE_REQ           = DhtNetworkSubProcess.REMOVE_NODE_REQ
-                    SHUTDOWN_NODE_REQ         = DhtNetworkSubProcess.SHUTDOWN_NODE_REQ
-                    SHUTDOWN_REPLACE_NODE_REQ = DhtNetworkSubProcess.SHUTDOWN_REPLACE_NODE_REQ
-                    SHUTDOWN_CLUSTER_REQ      = DhtNetworkSubProcess.SHUTDOWN_CLUSTER_REQ
-                    DUMP_STORAGE_REQ          = DhtNetworkSubProcess.DUMP_STORAGE_REQ
-                    MESSAGE_STATS             = DhtNetworkSubProcess.MESSAGE_STATS
+                    NODE_PUT_REQ              = DhtNetworkSubProcess.NODE_PUT_REQ.decode()
+                    NEW_NODE_REQ              = DhtNetworkSubProcess.NEW_NODE_REQ.decode()
+                    REMOVE_NODE_REQ           = DhtNetworkSubProcess.REMOVE_NODE_REQ.decode()
+                    SHUTDOWN_NODE_REQ         = DhtNetworkSubProcess.SHUTDOWN_NODE_REQ.decode()
+                    SHUTDOWN_REPLACE_NODE_REQ = DhtNetworkSubProcess.SHUTDOWN_REPLACE_NODE_REQ.decode()
+                    SHUTDOWN_CLUSTER_REQ      = DhtNetworkSubProcess.SHUTDOWN_CLUSTER_REQ.decode()
+                    DUMP_STORAGE_REQ          = DhtNetworkSubProcess.DUMP_STORAGE_REQ.decode()
+                    MESSAGE_STATS             = DhtNetworkSubProcess.MESSAGE_STATS.decode()
 
-                    req = packet[b'req'].decode()
-                    success = True
-                    if req in [SHUTDOWN_NODE_REQ, SHUTDOWN_REPLACE_NODE_REQ, REMOVE_NODE_REQ]:
-                        def delete_request(req, nid):
+                    if req in [SHUTDOWN_NODE_REQ,
+                               SHUTDOWN_REPLACE_NODE_REQ,
+                               REMOVE_NODE_REQ]:
+                        def delete_request(req, n):
                             global msg_stats
-                            if not nid:
-                                return
                             if req == SHUTDOWN_NODE_REQ:
-                                net.end_node(id=nid, shutdown=True, last_msg_stats=msg_stats)
+                                net.end_node(id=n, shutdown=True, last_msg_stats=msg_stats)
                             elif req == SHUTDOWN_REPLACE_NODE_REQ:
-                                net.replace_node(id=nid, shutdown=True, last_msg_stats=msg_stats)
+                                net.replace_node(id=n, shutdown=True, last_msg_stats=msg_stats)
                             elif req == REMOVE_NODE_REQ:
-                                net.end_node(id=nid, last_msg_stats=msg_stats)
+                                net.end_node(id=n, last_msg_stats=msg_stats)
 
-                        nodes = packet[b'ids']
-                        if nodes:
-                            for nid in nodes:
-                                delete_request(req, nid)
+                        if len(req) > 0:
+                            for n in req_args:
+                                delete_request(req, n)
                         else:
-                            n = net.get()
-                            if n:
-                                delete_request(req, n.getNodeId())
-                            else:
-                                success = False
+                            delete_request(req, net.get().getNodeId())
                     elif req == SHUTDOWN_CLUSTER_REQ:
                         for n in net.nodes:
                             net.end_node(id=n[2], shutdown=True, last_msg_stats=msg_stats)
@@ -569,26 +515,20 @@ if __name__ == '__main__':
                     elif req == NEW_NODE_REQ:
                         net.launch_node()
                     elif req == NODE_PUT_REQ:
-                        _hash = packet[b'hash']
-                        v = packet[b'value']
-                        n = net.get()
-                        if n:
-                            n.put(InfoHash(_hash), Value(v))
-                        else:
-                            success = False
+                        _hash, v = req_args[:2]
+                        net.get().put(InfoHash(_hash), Value(v))
+
                     elif req == DUMP_STORAGE_REQ:
-                        hashes = packet[b'ids']
-                        for n in [m[1] for m in net.nodes if m[1].getNodeId() in hashes]:
+                        for n in [m[1] for m in net.nodes if m[1].getNodeId() in req_args]:
                             net.log(n.getStorageLog())
-                    elif req == MESSAGE_STATS:
+                    elif MESSAGE_STATS in req:
                         stats = sum([np.array(x) for x in [net.getMessageStats()]+msg_stats])
-                        send_stats(packet, [int(_) for _ in stats])
+                        notify_benchmark(answer=[stats])
                         msg_stats.clear()
                         continue
-                    notify_benchmark(packet, success)
+                    notify_benchmark()
     except Exception as e:
-        traceback.print_tb(e.__traceback__)
-        print(type(e).__name__+':', e, file=sys.stderr)
+        DhtNetwork.log(e)
     finally:
         if net:
             net.resize(0)

python/tools/dht/tests.py

@@ -3,7 +3,6 @@
 # Author(s): Adrien Béraud <adrien.beraud@savoirfairelinux.com>
 #            Simon Désaulniers <sim.desaulniers@gmail.com>
 
-import sys
 import os
 import threading
 import random
@@ -34,22 +33,8 @@ bit_format = None
 Kbit_format = FuncFormatter(lambda x, pos: '%1.1f' % (x*1024**-1) + 'Kb')
 Mbit_format = FuncFormatter(lambda x, pos: '%1.1f' % (x*1024**-2) + 'Mb')
 
-def random_str_val(size=1024):
-    """Creates a random string value of specified size.
-
-    @param size:  Size, in bytes, of the value.
-    @type  size:  int
-
-    @return:  Random string value
-    @rtype :  str
-    """
-    return ''.join(random.choice(string.hexdigits) for _ in range(size))
-
-
 def random_hash():
-    """Creates random InfoHash.
-    """
-    return InfoHash(random_str_val(size=40).encode())
+    return InfoHash(''.join(random.SystemRandom().choice(string.hexdigits) for _ in range(40)).encode())
 
 def timer(f, *args):
     """
@@ -109,21 +94,6 @@ def display_plot(yvals, xvals=None, yformatter=None, display_time=3, **kwargs):
         plt.plot(yvals, **kwargs)
     plt.pause(display_time)
 
-def display_traffic_plot(ifname):
-    """Displays the traffic plot for a given interface name.
-
-    @param ifname:  Interface name.
-    @type  ifname:  string
-    """
-    ydata = []
-    xdata = []
-    # warning: infinite loop
-    interval = 2
-    for rate in iftop_traffic_data(ifname, interval=interval):
-        ydata.append(rate)
-        xdata.append((xdata[-1] if len(xdata) > 0 else 0) + interval)
-        display_plot(ydata, xvals=xdata, yformatter=Kbit_format, color='blue')
-
 def iftop_traffic_data(ifname, interval=2, rate_type='send_receive'):
     """
     Generator (yields data) function collecting traffic data from iftop
@@ -385,15 +355,15 @@ class DhtFeatureTest(FeatureTest):
     @staticmethod
     def getcb(value):
         vstr = value.__str__()[:100]
-        DhtNetwork.Log.log('[GET]: %s' % vstr + ("..." if len(vstr) > 100 else ""))
+        DhtNetwork.log('[GET]: %s' % vstr + ("..." if len(vstr) > 100 else ""))
         DhtFeatureTest.foreignValues.append(value)
         return True
 
     @staticmethod
     def putDoneCb(ok, nodes):
+        if not ok:
+            DhtNetwork.log("[PUT]: failed!")
         with FeatureTest.lock:
-            if not ok:
-                DhtNetwork.Log.log("[PUT]: failed!")
             FeatureTest.done -= 1
             FeatureTest.lock.notify()
 
@@ -401,7 +371,7 @@ class DhtFeatureTest(FeatureTest):
     def getDoneCb(ok, nodes):
         with FeatureTest.lock:
             if not ok:
-                DhtNetwork.Log.log("[GET]: failed!")
+                DhtNetwork.log("[GET]: failed!")
             else:
                 for node in nodes:
                     if not node.getNode().isExpired():
@@ -412,10 +382,10 @@ class DhtFeatureTest(FeatureTest):
     def _dhtPut(self, producer, _hash, *values):
         with FeatureTest.lock:
             for val in values:
-                vstr = val.__str__()[:100]
-                DhtNetwork.Log.log('[PUT]:', _hash.toString(), '->', vstr + ("..." if len(vstr) > 100 else ""))
-                FeatureTest.done += 1
-                producer.put(_hash, val, DhtFeatureTest.putDoneCb)
+                    vstr = val.__str__()[:100]
+                    DhtNetwork.log('[PUT]:', _hash.toString(), '->', vstr + ("..." if len(vstr) > 100 else ""))
+                    FeatureTest.done += 1
+                    producer.put(_hash, val, DhtFeatureTest.putDoneCb)
             while FeatureTest.done > 0:
                 FeatureTest.lock.wait()
 
@@ -424,7 +394,7 @@ class DhtFeatureTest(FeatureTest):
         DhtFeatureTest.foreignNodes = []
         with FeatureTest.lock:
             FeatureTest.done += 1
-            DhtNetwork.Log.log('[GET]:', _hash.toString())
+            DhtNetwork.log('[GET]:', _hash.toString())
             consumer.get(_hash, DhtFeatureTest.getcb, DhtFeatureTest.getDoneCb)
             while FeatureTest.done > 0:
                 FeatureTest.lock.wait()
@@ -505,22 +475,22 @@ class PersistenceTest(DhtFeatureTest):
     def _result(self, local_values, new_nodes):
         bootstrap = self._bootstrap
         if not DhtFeatureTest.successfullTransfer(local_values, DhtFeatureTest.foreignValues):
-            DhtNetwork.Log.log('[GET]: Only %s on %s values persisted.' %
+            DhtNetwork.log('[GET]: Only %s on %s values persisted.' %
                     (len(DhtFeatureTest.foreignValues), len(local_values)))
         else:
-            DhtNetwork.Log.log('[GET]: All values successfully persisted.')
+            DhtNetwork.log('[GET]: All values successfully persisted.')
         if DhtFeatureTest.foreignValues:
             if new_nodes:
-                DhtNetwork.Log.log('Values are newly found on:')
+                DhtNetwork.log('Values are newly found on:')
                 for node in new_nodes:
-                    DhtNetwork.Log.log(node)
+                    DhtNetwork.log(node)
                 if self._dump_storage:
-                    DhtNetwork.Log.log('Dumping all storage log from '\
+                    DhtNetwork.log('Dumping all storage log from '\
                                   'hosting nodes.')
                     for proc in self._workbench.procs:
-                        proc.sendClusterRequest(DhtNetworkSubProcess.DUMP_STORAGE_REQ, DhtFeatureTest.foreignNodes)
+                        proc.sendNodesRequest(DhtNetworkSubProcess.DUMP_STORAGE_REQ, DhtFeatureTest.foreignNodes)
             else:
-                DhtNetwork.Log.log("Values didn't reach new hosting nodes after shutdown.")
+                DhtNetwork.log("Values didn't reach new hosting nodes after shutdown.")
 
     def run(self):
         try:
@@ -535,8 +505,7 @@ class PersistenceTest(DhtFeatureTest):
             else:
                 raise NameError("This test is not defined '" + self._test + "'")
         except Exception as e:
-            traceback.print_tb(e.__traceback__)
-            print(type(e).__name__+':', e, file=sys.stderr)
+            print(e)
         finally:
             if self._traffic_plot or self._op_plot:
                 plot_fname = "traffic-plot"
@@ -558,86 +527,68 @@ class PersistenceTest(DhtFeatureTest):
         bootstrap = self._bootstrap
         # Value representing an ICE packet. Each ICE packet is around 1KB.
         VALUE_SIZE = 1024
-        num_values_per_hash = self._num_values/wb.node_num if self._num_values else 5
-
-        # nodes and values counters
-        total_nr_values = 0
+        NUM_VALUES = self._num_values/wb.node_num if self._num_values else 5
+        nr_values = NUM_VALUES * wb.node_num
         nr_nodes = wb.node_num
-        op_cv = threading.Condition()
+        nr_nodes_cv = threading.Condition()
 
-        # values string in string format. Used for sending cluster request.
+        values = [b''.join(random.choice(string.hexdigits).encode() for _ in range(VALUE_SIZE)) for __ in range(NUM_VALUES)]
         hashes = [random_hash() for _ in range(wb.node_num)]
 
-        def normalBehavior(do, t):
-            nonlocal total_nr_values, op_cv
+        # initial set of values
+        i = 0
+        for h in hashes:
+           self._dhtPut(bootstrap.front(), h, *[Value(v) for v in values])
+           print("at: ", i)
+           i += 1
+
+        def normalBehavior(do, t, log=None):
+            nonlocal nr_values
             while True:
-                with op_cv:
-                    do()
-                time.sleep(random.uniform(0.0, float(t)))
+                do()
+                time.sleep(random.choice(range(t)))
 
         def putRequest():
-            nonlocal hashes, VALUE_SIZE, total_nr_values
-            lock = threading.Condition()
-            def dcb(success):
-                nonlocal total_nr_values, lock
-                if success:
-                    total_nr_values += 1
-                    DhtNetwork.Log.log("INFO: "+ str(total_nr_values)+" values put on the dht since begining")
-                with lock:
-                    lock.notify()
-            with lock:
-                DhtNetwork.Log.warn("Random value put on the DHT...")
-                random.choice(wb.procs).sendClusterPutRequest(random.choice(hashes).toString(),
-                                                              random_str_val(size=VALUE_SIZE).encode(),
-                                                              done_cb=dcb)
-                lock.wait()
-
-        puts = threading.Thread(target=normalBehavior, args=(putRequest, 30.0/wb.node_num))
+            nonlocal hashes, values, nr_values
+            nr_values += 1
+            DhtNetwork.log("Random value put on the DHT.", "(now "+ str(nr_values)+" values on the dht)")
+            random.choice(wb.procs).sendNodePutRequest(random.choice(hashes).toString(), random.choice(values))
+        puts = threading.Thread(target=normalBehavior, args=(putRequest, 30))
         puts.daemon = True
         puts.start()
-
         def newNodeRequest():
             nonlocal nr_nodes
-            lock = threading.Condition()
-            def dcb(success):
-                nonlocal nr_nodes, lock
+            with nr_nodes_cv:
                 nr_nodes += 1
-                DhtNetwork.Log.log("INFO: now "+str(nr_nodes)+" nodes on the dht")
-                with lock:
-                    lock.notify()
-            with lock:
-                DhtNetwork.Log.warn("Node joining...")
-                random.choice(wb.procs).sendClusterRequest(DhtNetworkSubProcess.NEW_NODE_REQ, done_cb=dcb)
-                lock.wait()
-
-        connections = threading.Thread(target=normalBehavior, args=(newNodeRequest, 1*50.0/wb.node_num))
+                DhtNetwork.log("Node joining the DHT.", "(now "+str(nr_nodes)+" nodes on the dht)")
+                nr_nodes_cv.notify()
+            random.choice(wb.procs).sendNodesRequest(DhtNetworkSubProcess.NEW_NODE_REQ)
+        connections = threading.Thread(target=normalBehavior, args=(newNodeRequest, 1*60))
         connections.daemon = True
         connections.start()
-
         def shutdownNodeRequest():
             nonlocal nr_nodes
-            lock = threading.Condition()
-            def dcb(success):
-                nonlocal nr_nodes, lock
-                if success:
-                    nr_nodes -= 1
-                    DhtNetwork.Log.log("INFO: now "+str(nr_nodes)+" nodes on the dht")
-                else:
-                    DhtNetwork.Log.err("Oops.. No node to shutodwn.")
-
-                with lock:
-                    lock.notify()
-            with lock:
-                DhtNetwork.Log.warn("Node shutting down...")
-                random.choice(wb.procs).sendClusterRequest(DhtNetworkSubProcess.SHUTDOWN_NODE_REQ, done_cb=dcb)
-                lock.wait()
-
-        shutdowns = threading.Thread(target=normalBehavior, args=(shutdownNodeRequest, 1*60.0/wb.node_num))
+            with nr_nodes_cv:
+                nr_nodes -= 1
+                DhtNetwork.log("Node quitting the DHT.", "(now "+str(nr_nodes)+" nodes on the dht)")
+                nr_nodes_cv.notify()
+            random.choice(wb.procs).sendNodesRequest(DhtNetworkSubProcess.SHUTDOWN_NODE_REQ)
+        shutdowns = threading.Thread(target=normalBehavior, args=(shutdownNodeRequest, 1*60))
         shutdowns.daemon = True
         shutdowns.start()
 
+        for h in hashes:
+           self._trigger_dp(trigger_nodes, h)
+
         if self._traffic_plot:
-            display_traffic_plot('br'+wb.ifname)
+            ydata = []
+            xdata = []
+            # warning: infinite loop
+            interval = 2
+            for rate in iftop_traffic_data("br"+wb.ifname, interval=interval):
+                ydata.append(rate)
+                xdata.append((xdata[-1] if len(xdata) > 0 else 0) + interval)
+                display_plot(ydata, xvals=xdata, yformatter=Kbit_format, color='blue')
         else:
             # blocks in matplotlib thread
             while True:
@@ -667,54 +618,36 @@ class PersistenceTest(DhtFeatureTest):
         self._dhtGet(consumer, myhash)
         if not DhtFeatureTest.successfullTransfer(local_values, DhtFeatureTest.foreignValues):
             if DhtFeatureTest.foreignValues:
-                DhtNetwork.Log.log('[GET]: Only ', len(DhtFeatureTest.foreignValues) ,' on ',
+                DhtNetwork.log('[GET]: Only ', len(DhtFeatureTest.foreignValues) ,' on ',
                         len(local_values), ' values successfully put.')
             else:
-                DhtNetwork.Log.log('[GET]: 0 values successfully put')
+                DhtNetwork.log('[GET]: 0 values successfully put')
 
 
         if DhtFeatureTest.foreignValues and DhtFeatureTest.foreignNodes:
-            DhtNetwork.Log.log('Values are found on :')
+            DhtNetwork.log('Values are found on :')
             for node in DhtFeatureTest.foreignNodes:
-                DhtNetwork.Log.log(node)
+                DhtNetwork.log(node)
 
             for _ in range(max(1, int(self._workbench.node_num/32))):
-                DhtNetwork.Log.log('Removing all nodes hosting target values...')
+                DhtNetwork.log('Removing all nodes hosting target values...')
                 cluster_ops_count = 0
                 for proc in self._workbench.procs:
-                    DhtNetwork.Log.log('[REMOVE]: sending shutdown request to', proc)
-                    lock = threading.Condition()
-                    def dcb(success):
-                        nonlocal lock
-                        if not success:
-                            DhtNetwork.Log.err("Failed to shutdown.")
-                        with lock:
-                            lock.notify()
-
-                    with lock:
-                        proc.sendClusterRequest(
+                    DhtNetwork.log('[REMOVE]: sending shutdown request to', proc)
+                    proc.sendNodesRequest(
                             DhtNetworkSubProcess.SHUTDOWN_NODE_REQ,
-                            DhtFeatureTest.foreignNodes,
-                            done_cb=dcb
-                        )
-                        lock.wait()
-                    DhtNetwork.Log.log('sending message stats request')
-                    def msg_dcb(stats):
-                        nonlocal cluster_ops_count, lock
-                        if stats:
-                            cluster_ops_count += sum(stats[1:])
-                        with lock:
-                            lock.notify()
-                    with lock:
-                        proc.sendGetMessageStats(done_cb=msg_dcb)
-                        lock.wait()
-                    DhtNetwork.Log.log("5 seconds wait...")
+                            DhtFeatureTest.foreignNodes
+                    )
+                    DhtNetwork.log('sending message stats request')
+                    stats = proc.sendGetMessageStats()
+                    cluster_ops_count += sum(stats[1:])
+                    DhtNetwork.log("5 seconds wait...")
                     time.sleep(5)
                 ops_count.append(cluster_ops_count/self._workbench.node_num)
 
                 # checking if values were transfered to new nodes
                 foreignNodes_before_delete = DhtFeatureTest.foreignNodes
-                DhtNetwork.Log.log('[GET]: trying to fetch persistent values')
+                DhtNetwork.log('[GET]: trying to fetch persistent values')
                 self._dhtGet(consumer, myhash)
                 new_nodes = set(DhtFeatureTest.foreignNodes) - set(foreignNodes_before_delete)
 
@@ -723,11 +656,7 @@ class PersistenceTest(DhtFeatureTest):
             if self._op_plot:
                 display_plot(ops_count, color='blue')
         else:
-            DhtNetwork.Log.log("[GET]: either couldn't fetch values or nodes hosting values...")
-
-        if traffic_plot_thread:
-            print("Traffic plot running for ever. Ctrl-c for stopping it.")
-            traffic_plot_thread.join()
+            DhtNetwork.log("[GET]: either couldn't fetch values or nodes hosting values...")
 
     @reset_before_test
     def _replaceClusterTest(self):
@@ -749,16 +678,16 @@ class PersistenceTest(DhtFeatureTest):
         self._dhtGet(consumer, myhash)
         initial_nodes = DhtFeatureTest.foreignNodes
 
-        DhtNetwork.Log.log('Replacing', clusters, 'random clusters successively...')
+        DhtNetwork.log('Replacing', clusters, 'random clusters successively...')
         for n in range(clusters):
             i = random.randint(0, len(self._workbench.procs)-1)
             proc = self._workbench.procs[i]
-            DhtNetwork.Log.log('Replacing', proc)
-            proc.sendClusterRequest(DhtNetworkSubProcess.SHUTDOWN_CLUSTER_REQ)
+            DhtNetwork.log('Replacing', proc)
+            proc.sendNodesRequest(DhtNetworkSubProcess.SHUTDOWN_CLUSTER_REQ)
             self._workbench.stop_cluster(i)
             self._workbench.start_cluster(i)
 
-        DhtNetwork.Log.log('[GET]: trying to fetch persistent values')
+        DhtNetwork.log('[GET]: trying to fetch persistent values')
         self._dhtGet(consumer, myhash)
         new_nodes = set(DhtFeatureTest.foreignNodes) - set(initial_nodes)
 
@@ -783,7 +712,7 @@ class PersistenceTest(DhtFeatureTest):
     # Generating considerable amount of values of size 1KB.
         VALUE_SIZE = 1024
         NUM_VALUES = self._num_values if self._num_values else 50
-        values = [Value(random_str_val(size=VALUE_SIZE).encode()) for _ in range(NUM_VALUES)]
+        values = [Value(''.join(random.choice(string.hexdigits) for _ in range(VALUE_SIZE)).encode()) for _ in range(NUM_VALUES)]
 
         bootstrap.resize(N_PRODUCERS+2)
         consumer = bootstrap.get(N_PRODUCERS+1)
@@ -797,21 +726,21 @@ class PersistenceTest(DhtFeatureTest):
             nodes = set([])
             self._gottaGetThemAllPokeNodes(consumer, hashes, nodes=nodes)
 
-            DhtNetwork.Log.log("Values are found on:")
+            DhtNetwork.log("Values are found on:")
             for n in nodes:
-                DhtNetwork.Log.log(n)
+                DhtNetwork.log(n)
 
-            DhtNetwork.Log.log("Creating 8 nodes around all of these hashes...")
+            DhtNetwork.log("Creating 8 nodes around all of these hashes...")
             for _hash in hashes:
                 self._trigger_dp(trigger_nodes, _hash, count=8)
 
-            DhtNetwork.Log.log('Waiting', DP_TIMEOUT+1, 'minutes for normal storage maintenance.')
+            DhtNetwork.log('Waiting', DP_TIMEOUT+1, 'minutes for normal storage maintenance.')
             time.sleep((DP_TIMEOUT+1)*60)
 
-            DhtNetwork.Log.log('Deleting old nodes from previous search.')
+            DhtNetwork.log('Deleting old nodes from previous search.')
             for proc in self._workbench.procs:
-                DhtNetwork.Log.log('[REMOVE]: sending delete request to', proc)
-                proc.sendClusterRequest(
+                DhtNetwork.log('[REMOVE]: sending delete request to', proc)
+                proc.sendNodesRequest(
                     DhtNetworkSubProcess.REMOVE_NODE_REQ,
                     nodes)
 
@@ -852,8 +781,7 @@ class PerformanceTest(DhtFeatureTest):
             else:
                 raise NameError("This test is not defined '" + self._test + "'")
         except Exception as e:
-            traceback.print_tb(e.__traceback__)
-            print(type(e).__name__+':', e, file=sys.stderr)
+            print(e)
         finally:
             self._bootstrap.resize(1)
 
@@ -892,7 +820,7 @@ class PerformanceTest(DhtFeatureTest):
 
         def getcb(v):
             nonlocal bootstrap
-            DhtNetwork.Log.log("found", v)
+            DhtNetwork.log("found", v)
             return True
 
         def donecb(ok, nodes, start):
@@ -900,7 +828,7 @@ class PerformanceTest(DhtFeatureTest):
             t = time.time()-start
             with lock:
                 if not ok:
-                    DhtNetwork.Log.log("failed !")
+                    DhtNetwork.log("failed !")
                 times.append(t)
                 done -= 1
                 lock.notify()
@@ -932,7 +860,7 @@ class PerformanceTest(DhtFeatureTest):
         for n in range(10):
             self._workbench.replace_cluster()
             plt.pause(2)
-            DhtNetwork.Log.log("Getting 50 random hashes succesively.")
+            DhtNetwork.log("Getting 50 random hashes succesively.")
             for i in range(50):
                 with lock:
                     for _ in range(1):
@@ -967,28 +895,28 @@ class PerformanceTest(DhtFeatureTest):
 
         for _ in range(max(1, int(self._workbench.node_num/32))):
             self._dhtGet(consumer, myhash)
-            DhtNetwork.Log.log("Waiting 15 seconds...")
+            DhtNetwork.log("Waiting 15 seconds...")
             time.sleep(15)
 
             self._dhtPut(producer, myhash, *local_values)
 
             #checking if values were transfered
             self._dhtGet(consumer, myhash)
-            DhtNetwork.Log.log('Values are found on :')
+            DhtNetwork.log('Values are found on :')
             for node in DhtFeatureTest.foreignNodes:
-                DhtNetwork.Log.log(node)
+                DhtNetwork.log(node)
 
             if not DhtFeatureTest.successfullTransfer(local_values, DhtFeatureTest.foreignValues):
                 if DhtFeatureTest.foreignValues:
-                    DhtNetwork.Log.log('[GET]: Only ', len(DhtFeatureTest.foreignValues) ,' on ',
+                    DhtNetwork.log('[GET]: Only ', len(DhtFeatureTest.foreignValues) ,' on ',
                             len(local_values), ' values successfully put.')
                 else:
-                    DhtNetwork.Log.log('[GET]: 0 values successfully put')
+                    DhtNetwork.log('[GET]: 0 values successfully put')
 
-            DhtNetwork.Log.log('Removing all nodes hosting target values...')
+            DhtNetwork.log('Removing all nodes hosting target values...')
             for proc in self._workbench.procs:
-                DhtNetwork.Log.log('[REMOVE]: sending shutdown request to', proc)
-                proc.sendClusterRequest(
+                DhtNetwork.log('[REMOVE]: sending shutdown request to', proc)
+                proc.sendNodesRequest(
                         DhtNetworkSubProcess.SHUTDOWN_NODE_REQ,
                         DhtFeatureTest.foreignNodes
                 )

resources/opendht_logo.svg

@@ -1,105 +0,0 @@
-<?xml version="1.0" encoding="UTF-8" standalone="no"?>
-<!-- Created with Inkscape (http://www.inkscape.org/) -->
-
-<svg
-   xmlns:dc="http://purl.org/dc/elements/1.1/"
-   xmlns:cc="http://creativecommons.org/ns#"
-   xmlns:rdf="http://www.w3.org/1999/02/22-rdf-syntax-ns#"
-   xmlns:svg="http://www.w3.org/2000/svg"
-   xmlns="http://www.w3.org/2000/svg"
-   xmlns:sodipodi="http://sodipodi.sourceforge.net/DTD/sodipodi-0.dtd"
-   xmlns:inkscape="http://www.inkscape.org/namespaces/inkscape"
-   id="svg3492"
-   version="1.1"
-   inkscape:version="0.91 r13725"
-   xml:space="preserve"
-   width="256"
-   height="256"
-   viewBox="0 0 256 255.99999"
-   inkscape:export-xdpi="90"
-   inkscape:export-ydpi="90"><title
-     id="title3357">OpenDHT</title><metadata
-     id="metadata3498"><rdf:RDF><cc:Work
-         rdf:about=""><dc:format>image/svg+xml</dc:format><dc:type
-           rdf:resource="http://purl.org/dc/dcmitype/StillImage" /><dc:title>OpenDHT</dc:title><dc:rights><cc:Agent><dc:title></dc:title></cc:Agent></dc:rights><dc:creator><cc:Agent><dc:title>Savoir-faire Linux Inc.</dc:title></cc:Agent></dc:creator><dc:relation>https://opendht.net</dc:relation></cc:Work></rdf:RDF></metadata><defs
-     id="defs3496"><clipPath
-       clipPathUnits="userSpaceOnUse"
-       id="clipPath3534"><path
-         d="m 0,0 1152,0 0,648 L 0,648 0,0 Z"
-         id="path3536"
-         inkscape:connector-curvature="0" /></clipPath></defs><sodipodi:namedview
-     pagecolor="#ffffff"
-     bordercolor="#666666"
-     borderopacity="1"
-     objecttolerance="10"
-     gridtolerance="10"
-     guidetolerance="10"
-     inkscape:pageopacity="0"
-     inkscape:pageshadow="2"
-     inkscape:window-width="1855"
-     inkscape:window-height="1056"
-     id="namedview3494"
-     showgrid="false"
-     units="px"
-     fit-margin-top="5"
-     fit-margin-right="5"
-     fit-margin-bottom="5"
-     fit-margin-left="5"
-     inkscape:zoom="1.9555556"
-     inkscape:cx="63.161953"
-     inkscape:cy="141.16919"
-     inkscape:window-x="1985"
-     inkscape:window-y="24"
-     inkscape:window-maximized="1"
-     inkscape:current-layer="g3532" /><g
-     id="g3500"
-     inkscape:groupmode="layer"
-     inkscape:label="SFL-logo-DHT"
-     transform="matrix(1.25,0,0,-1.25,-589.00821,604.10631)"><g
-       id="g3530"><g
-         id="g3532"
-         clip-path="url(#clipPath3534)"><g
-           id="g3359"
-           transform="matrix(0.9931555,0,0,0.99058777,1.5489503,0.9813209)"><g
-             transform="translate(485.3798,383.7826)"
-             id="g3538"><path
-               inkscape:connector-curvature="0"
-               id="path3540"
-               style="fill:#47b3d1;fill-opacity:1;fill-rule:nonzero;stroke:none"
-               d="m 0,0 c 0,50.03 40.554,90.586 90.586,90.586 50.028,0 90.587,-40.556 90.587,-90.586 0,-50.03 -40.559,-90.588 -90.587,-90.588 C 40.554,-90.588 0,-50.03 0,0" /></g><g
-             transform="translate(576.0338,313.2273)"
-             id="g3542"><path
-               inkscape:connector-curvature="0"
-               id="path3544"
-               style="fill:#0091ba;fill-opacity:1;fill-rule:nonzero;stroke:none"
-               d="m 0,0 c 0,54.307 -24.009,102.977 -61.954,136.075 -17.613,-16.526 -28.632,-39.996 -28.632,-66.056 0,-49.629 39.916,-89.912 89.393,-90.555 0.396,-0.009 0.792,-0.033 1.193,-0.033 L 0,-0.082 0,0 Z" /></g><g
-             transform="translate(575.9661,292.6626)"
-             id="g3546"><path
-               inkscape:connector-curvature="0"
-               id="path3548"
-               style="fill:#007aa3;fill-opacity:1;fill-rule:nonzero;stroke:none"
-               d="m 0,0 c -1.439,49.482 -41.105,89.355 -90.504,91.12 0,-0.178 -0.015,-0.355 -0.015,-0.536 C -90.519,40.578 -49.998,0.039 0,0" /></g><g
-             transform="translate(532.4411,337.497)"
-             id="g3550"><path
-               inkscape:connector-curvature="0"
-               id="path3552"
-               style="fill:#006382;fill-opacity:1;fill-rule:nonzero;stroke:none"
-               d="m 0,0 c -10.555,0 -20.236,-3.754 -27.777,-10 16.565,-21.182 42.472,-34.815 71.442,-34.834 0.01,0.415 -0.072,0.821 -0.072,1.241 C 43.593,-19.516 24.077,0 0,0" /></g><g
-             transform="translate(576.0338,313.2273)"
-             id="g3554"><path
-               inkscape:connector-curvature="0"
-               id="path3556"
-               style="fill:#0091ba;fill-opacity:1;fill-rule:nonzero;stroke:none"
-               d="M 0,0 C 0,54.307 24.014,102.977 61.954,136.075 79.567,119.549 90.586,96.079 90.586,70.019 90.586,20.39 50.675,-19.893 1.198,-20.536 0.797,-20.545 0.406,-20.569 0.005,-20.569 l 0,20.487 C 0.005,-0.053 0,-0.029 0,0" /></g><g
-             transform="translate(576.1063,292.6626)"
-             id="g3558"><path
-               inkscape:connector-curvature="0"
-               id="path3560"
-               style="fill:#007aa3;fill-opacity:1;fill-rule:nonzero;stroke:none"
-               d="m 0,0 c 1.44,49.482 41.105,89.355 90.504,91.12 0,-0.178 0.01,-0.355 0.01,-0.536 C 90.514,40.578 49.998,0.039 0,0" /></g><g
-             transform="translate(619.6266,337.497)"
-             id="g3562"><path
-               inkscape:connector-curvature="0"
-               id="path3564"
-               style="fill:#006382;fill-opacity:1;fill-rule:nonzero;stroke:none"
-               d="m 0,0 c 10.56,0 20.241,-3.754 27.782,-10 -16.57,-21.182 -42.405,-34.815 -71.375,-34.834 -0.009,0.415 0,0.821 0,1.241 C -43.593,-19.516 -24.072,0 0,0" /></g></g></g></g></g></svg>

src/Makefile.am

@@ -4,7 +4,7 @@ lib_LTLIBRARIES = libopendht.la
 
 AM_CPPFLAGS = -I../include/opendht
 libopendht_la_CXXFLAGS = @CXXFLAGS@
-libopendht_la_LDFLAGS = @LDFLAGS@ @GnuTLS_LIBS@ @Nettle_LIBS@
+libopendht_la_LDFLAGS = @LDFLAGS@ @GNUTLS_LIBS@ @nettle_LIBS@
 libopendht_la_LIBADD = ./argon2/libargon2.la
 
 libopendht_la_SOURCES = \
@@ -35,9 +35,7 @@ nobase_include_HEADERS = \
         ../include/opendht/routing_table.h \
         ../include/opendht/network_engine.h \
         ../include/opendht/scheduler.h \
-        ../include/opendht/rate_limiter.h \
         ../include/opendht/utils.h \
-        ../include/opendht/sockaddr.h \
         ../include/opendht/infohash.h \
         ../include/opendht/node.h \
         ../include/opendht/value.h \
@@ -46,9 +44,5 @@ nobase_include_HEADERS = \
         ../include/opendht/dhtrunner.h \
         ../include/opendht/default_types.h \
         ../include/opendht/log.h \
-        ../include/opendht/log_enable.h \
         ../include/opendht/rng.h \
         ../include/opendht/indexation/pht.h
-
-nobase_noinst_HEADERS = \
-        ../include/opendht/request.h

src/argon2/Makefile.am

@@ -10,14 +10,4 @@ libargon2_la_SOURCES = \
         encoding.c \
         ref.c
 
-noinst_HEADERS = \
-        argon2.h \
-        core.h \
-        blake2/blake2.h \
-        blake2/blake2-impl.h \
-        blake2/blamka-round-ref.h \
-        blake2/blamka-round-opt.h \
-        thread.h \
-        encoding.h \
-        opt.h \
-        ref.h
+EXTRA_libargon2_la_SOURCES = opt.c

src/argon2/argon2.c

@@ -88,12 +88,18 @@ int argon2_hash(const uint32_t t_cost, const uint32_t m_cost,
     int result;
     uint8_t *out;
 
+    /* Detect and reject overflowing sizes */
+    /* TODO: This should probably be fixed in the function signature */
+    if (pwdlen > ARGON2_MAX_PWD_LENGTH) {
+        return ARGON2_PWD_TOO_LONG;
+    }
+
     if (hashlen > ARGON2_MAX_OUTLEN) {
         return ARGON2_OUTPUT_TOO_LONG;
     }
 
-    if (hashlen < ARGON2_MIN_OUTLEN) {
-        return ARGON2_OUTPUT_TOO_SHORT;
+    if (saltlen > ARGON2_MAX_SALT_LENGTH) {
+        return ARGON2_SALT_TOO_LONG;
     }
 
     out = malloc(hashlen);
@@ -206,19 +212,13 @@ int argon2_verify(const char *encoded, const void *pwd, const size_t pwdlen,
     int ret;
     int decode_result;
     uint32_t encoded_len;
-    size_t encoded_len_tmp;
 
     if(encoded == NULL) {
         return ARGON2_DECODING_FAIL;
     }
 
-    encoded_len_tmp = strlen(encoded);
     /* max values, to be updated in decode_string */
-    if (UINT32_MAX < encoded_len_tmp) {
-        return ARGON2_DECODING_FAIL;
-    }
-
-    encoded_len = (uint32_t)encoded_len_tmp;
+    encoded_len = strlen(encoded);
     ctx.adlen = encoded_len;
     ctx.saltlen = encoded_len;
     ctx.outlen = encoded_len;
@@ -226,8 +226,6 @@ int argon2_verify(const char *encoded, const void *pwd, const size_t pwdlen,
     ctx.free_cbk = NULL;
     ctx.secret = NULL;
     ctx.secretlen = 0;
-    ctx.pwdlen = 0;
-    ctx.pwd = NULL;
     ctx.ad = malloc(ctx.adlen);
     ctx.salt = malloc(ctx.saltlen);
     ctx.out = malloc(ctx.outlen);
@@ -394,5 +392,5 @@ size_t argon2_encodedlen(uint32_t t_cost, uint32_t m_cost, uint32_t parallelism,
                          uint32_t saltlen, uint32_t hashlen) {
     return strlen("$argon2x$v=$m=,t=,p=$$") + numlen(t_cost) + numlen(m_cost)
         + numlen(parallelism) + b64len(saltlen) + b64len(hashlen)
-        + numlen(ARGON2_VERSION_NUMBER) + 1;
+        + numlen(ARGON2_VERSION_NUMBER);
 }

src/argon2/encoding.c

@@ -412,18 +412,7 @@ int encode_string(char *dst, size_t dst_len, argon2_context *ctx,
 }
 
 size_t b64len(uint32_t len) {
-    size_t olen = ((size_t)len / 3) << 2;
-
-    switch (len % 3) {
-    case 2:
-        olen++;
-    /* fall through */
-    case 1:
-        olen += 2;
-        break;
-    }
-
-    return olen;
+    return (((size_t)len + 2) / 3) * 4;
 }
 
 size_t numlen(uint32_t num) {

src/crypto.cpp

@@ -83,18 +83,6 @@ namespace crypto {
 static constexpr std::array<size_t, 3> AES_LENGTHS {{128/8, 192/8, 256/8}};
 static constexpr size_t PASSWORD_SALT_LENGTH {16};
 
-constexpr gnutls_digest_algorithm_t gnutlsHashAlgo(size_t min_res) {
-    return (min_res > 256/8) ? GNUTLS_DIG_SHA512 : (
-           (min_res > 128/8) ? GNUTLS_DIG_SHA256 : (
-                               GNUTLS_DIG_SHA1));
-}
-
-constexpr size_t gnutlsHashSize(int algo) {
-    return (algo == GNUTLS_DIG_SHA512) ? 512/8 : (
-           (algo == GNUTLS_DIG_SHA256) ? 256/8 : (
-           (algo == GNUTLS_DIG_SHA1)   ? 160/8 : 0 ));
-}
-
 size_t aesKeySize(size_t max)
 {
     size_t aes_key_len = 0;
@@ -140,7 +128,8 @@ Blob aesEncrypt(const Blob& data, const Blob& key)
 Blob aesEncrypt(const Blob& data, const std::string& password)
 {
     Blob salt;
-    Blob key = stretchKey(password, salt, 256 / 8);
+    Blob key = stretchKey(password, salt);
+    key.resize(256 / 8);
     Blob encrypted = aesEncrypt(data, key);
     encrypted.insert(encrypted.begin(), salt.begin(), salt.end());
     return encrypted;
@@ -186,12 +175,13 @@ Blob aesDecrypt(const Blob& data, const std::string& password)
     if (data.size() <= PASSWORD_SALT_LENGTH)
         throw DecryptError("Wrong data size");    
     Blob salt {data.begin(), data.begin()+PASSWORD_SALT_LENGTH};
-    Blob key = stretchKey(password, salt, 256/8);
+    Blob key = stretchKey(password, salt);
+    key.resize(256 / 8);
     Blob encrypted {data.begin()+PASSWORD_SALT_LENGTH, data.end()};
     return aesDecrypt(encrypted, key);
 }
 
-Blob stretchKey(const std::string& password, Blob& salt, size_t key_length)
+Blob stretchKey(const std::string& password, Blob& salt)
 {
     if (salt.empty()) {
         salt.resize(PASSWORD_SALT_LENGTH);
@@ -203,27 +193,35 @@ Blob stretchKey(const std::string& password, Blob& salt, size_t key_length)
     auto ret = argon2i_hash_raw(16, 64*1024, 1, password.data(), password.size(), salt.data(), salt.size(), res.data(), res.size());
     if (ret != ARGON2_OK)
         throw CryptoException("Can't compute argon2i !");
-    return hash(res, key_length);
+    return hash(res);
 }
 
-Blob hash(const Blob& data, size_t hash_len)
+Blob hash(const Blob& data)
 {
-    auto algo = gnutlsHashAlgo(hash_len);    
-    size_t res_size = gnutlsHashSize(algo);
     Blob res;
+    size_t res_size = 64;
     res.resize(res_size);
     const gnutls_datum_t gdat {(uint8_t*)data.data(), (unsigned)data.size()};
-    if (gnutls_fingerprint(algo, &gdat, res.data(), &res_size))
+    if (gnutls_fingerprint(GNUTLS_DIG_SHA512, &gdat, res.data(), &res_size))
         throw CryptoException("Can't compute hash !");
-    res.resize(std::min(hash_len, res_size));
+    res.resize(res_size);
     return res;
 }
 
 PrivateKey::PrivateKey()
-{}
+{
+#if GNUTLS_VERSION_NUMBER < 0x030300
+    if (gnutls_global_init() != GNUTLS_E_SUCCESS)
+        throw CryptoException("Can't initialize GnuTLS.");
+#endif
+}
 
 PrivateKey::PrivateKey(gnutls_x509_privkey_t k) : x509_key(k)
 {
+#if GNUTLS_VERSION_NUMBER < 0x030300
+    if (gnutls_global_init() != GNUTLS_E_SUCCESS)
+        throw CryptoException("Can't initialize GnuTLS.");
+#endif
     gnutls_privkey_init(&key);
     if (gnutls_privkey_import_x509(key, k, GNUTLS_PRIVKEY_IMPORT_COPY) != GNUTLS_E_SUCCESS) {
         key = nullptr;
@@ -233,6 +231,10 @@ PrivateKey::PrivateKey(gnutls_x509_privkey_t k) : x509_key(k)
 
 PrivateKey::PrivateKey(const Blob& import, const std::string& password)
 {
+#if GNUTLS_VERSION_NUMBER < 0x030300
+    if (gnutls_global_init() != GNUTLS_E_SUCCESS)
+        throw CryptoException("Can't initialize GnuTLS.");
+#endif
     int err = gnutls_x509_privkey_init(&x509_key);
     if (err != GNUTLS_E_SUCCESS)
         throw CryptoException("Can't initialize private key !");
@@ -264,9 +266,14 @@ PrivateKey::PrivateKey(const Blob& import, const std::string& password)
 
 PrivateKey::PrivateKey(PrivateKey&& o) noexcept : key(o.key), x509_key(o.x509_key)
 {
+#if GNUTLS_VERSION_NUMBER < 0x030300
+    // gnutls_global_init already succeeded at least once here so no real need to check.
+    int ret = gnutls_global_init();
+    assert(ret == GNUTLS_E_SUCCESS);
+#endif
     o.key = nullptr;
     o.x509_key = nullptr;
-}
+};
 
 PrivateKey::~PrivateKey()
 {
@@ -278,6 +285,9 @@ PrivateKey::~PrivateKey()
         gnutls_x509_privkey_deinit(x509_key);
         x509_key = nullptr;
     }
+#if GNUTLS_VERSION_NUMBER < 0x030300
+    gnutls_global_deinit();
+#endif
 }
 
 PrivateKey&
@@ -301,8 +311,6 @@ PrivateKey::sign(const Blob& data) const
 {
     if (!key)
         throw CryptoException("Can't sign data: no private key set !");
-    if (std::numeric_limits<unsigned>::max() < data.size())
-        throw CryptoException("Can't sign data: too large !");
     gnutls_datum_t sig;
     const gnutls_datum_t dat {(unsigned char*)data.data(), (unsigned)data.size()};
     if (gnutls_privkey_sign_data(key, GNUTLS_DIG_SHA512, 0, &dat, &sig) != GNUTLS_E_SUCCESS)
@@ -728,6 +736,10 @@ Certificate::toString(bool chain) const
 PrivateKey
 PrivateKey::generate(unsigned key_length)
 {
+#if GNUTLS_VERSION_NUMBER < 0x030300
+    if (gnutls_global_init() != GNUTLS_E_SUCCESS)
+        throw CryptoException("Can't initialize GnuTLS.");
+#endif
     gnutls_x509_privkey_t key;
     if (gnutls_x509_privkey_init(&key) != GNUTLS_E_SUCCESS)
         throw CryptoException("Can't initialize private key.");
@@ -742,8 +754,17 @@ PrivateKey::generate(unsigned key_length)
 Identity
 generateIdentity(const std::string& name, crypto::Identity ca, unsigned key_length, bool is_ca)
 {
+#if GNUTLS_VERSION_NUMBER < 0x030300
+    if (gnutls_global_init() != GNUTLS_E_SUCCESS)
+        return {};
+#endif
     auto key = std::make_shared<PrivateKey>(PrivateKey::generate(key_length));
+
     auto cert = std::make_shared<Certificate>(Certificate::generate(*key, name, ca, is_ca));
+
+#if GNUTLS_VERSION_NUMBER < 0x030300
+    gnutls_global_deinit();
+#endif
     return {std::move(key), std::move(cert)};
 }
 

src/dhtrunner.cpp

@@ -37,8 +37,6 @@
 
 namespace dht {
 
-constexpr std::chrono::seconds DhtRunner::BOOTSTRAP_PERIOD;
-
 DhtRunner::DhtRunner() : dht_()
 {
 #ifdef _WIN32
@@ -106,7 +104,7 @@ DhtRunner::run(const sockaddr_in* local4, const sockaddr_in6* local6, DhtRunner:
                     std::lock_guard<std::mutex> lck(storage_mtx);
                     if (not pending_ops_prio.empty())
                         return true;
-                    if (not pending_ops.empty() and getStatus() != NodeStatus::Connecting)
+                    if (not pending_ops.empty() and getStatus() >= NodeStatus::Connecting)
                         return true;
                 }
                 return false;
@@ -129,14 +127,10 @@ DhtRunner::join()
 {
     running = false;
     cv.notify_all();
-    bootstrap_cv.notify_all();
     if (dht_thread.joinable())
         dht_thread.join();
     if (rcv_thread.joinable())
         rcv_thread.join();
-    if (bootstrap_thread.joinable())
-        bootstrap_thread.join();
-
     {
         std::lock_guard<std::mutex> lck(storage_mtx);
         pending_ops = decltype(pending_ops)();
@@ -214,12 +208,6 @@ DhtRunner::setLoggers(LogMethod error, LogMethod warn, LogMethod debug) {
     dht_->setLoggers(std::forward<LogMethod>(error), std::forward<LogMethod>(warn), std::forward<LogMethod>(debug));
 }
 
-void
-DhtRunner::setLogFilter(const InfoHash& f) {
-    std::lock_guard<std::mutex> lck(dht_mtx);
-    dht_->setLogFilter(f);
-}
-
 void
 DhtRunner::registerType(const ValueType& type) {
     std::lock_guard<std::mutex> lck(dht_mtx);
@@ -253,12 +241,6 @@ DhtRunner::getStorageLog() const
     return dht_->getStorageLog();
 }
 std::string
-DhtRunner::getStorageLog(const InfoHash& f) const
-{
-    std::lock_guard<std::mutex> lck(dht_mtx);
-    return dht_->getStorageLog(f);
-}
-std::string
 DhtRunner::getRoutingTablesLog(sa_family_t af) const
 {
     std::lock_guard<std::mutex> lck(dht_mtx);
@@ -270,7 +252,7 @@ DhtRunner::getSearchesLog(sa_family_t af) const
     std::lock_guard<std::mutex> lck(dht_mtx);
     return dht_->getSearchesLog(af);
 }
-std::vector<SockAddr>
+std::vector<Address>
 DhtRunner::getPublicAddress(sa_family_t af)
 {
     std::lock_guard<std::mutex> lck(dht_mtx);
@@ -311,7 +293,7 @@ DhtRunner::loop_()
         ops.front()(*dht_);
         ops.pop();
     }
-    if (getStatus() != NodeStatus::Connecting) {
+    if (getStatus() >= NodeStatus::Connecting) {
         {
             std::lock_guard<std::mutex> lck(storage_mtx);
             ops = std::move(pending_ops);
@@ -333,7 +315,7 @@ DhtRunner::loop_()
         for (const auto& pck : received) {
             auto& buf = pck.first;
             auto& from = pck.second;
-            wakeup = dht_->periodic(buf.data(), buf.size()-1, from);
+            wakeup = dht_->periodic(buf.data(), buf.size()-1, (sockaddr*)&from.first, from.second);
         }
         received.clear();
     } else {
@@ -345,15 +327,6 @@ DhtRunner::loop_()
     if (nstatus4 != status4 || nstatus6 != status6) {
         status4 = nstatus4;
         status6 = nstatus6;
-        if (status4 == NodeStatus::Disconnected and status6 == NodeStatus::Disconnected) {
-            // We have lost connection with the DHT.  Try to recover using bootstrap nodes.
-            std::unique_lock<std::mutex> lck(bootstrap_mtx);
-            bootstrap_nodes = bootstrap_nodes_all;
-            tryBootstrapCoutinuously();
-        } else {
-            std::unique_lock<std::mutex> lck(bootstrap_mtx);
-            bootstrap_nodes.clear();
-        }
         if (statusCb)
             statusCb(status4, status6);
     }
@@ -427,19 +400,19 @@ DhtRunner::doRun(const sockaddr_in* sin4, const sockaddr_in6* sin6, SecureDht::C
 
                 if(rc > 0) {
                     std::array<uint8_t, 1024 * 64> buf;
-                    SockAddr from;
-                    from.second = sizeof(from.first);
+                    sockaddr_storage from;
+                    socklen_t fromlen {sizeof(from)};
 
                     if(s4 >= 0 && FD_ISSET(s4, &readfds))
-                        rc = recvfrom(s4, (char*)buf.data(), buf.size(), 0, (struct sockaddr*)&from.first, &from.second);
+                        rc = recvfrom(s4, (char*)buf.data(), buf.size(), 0, (struct sockaddr*)&from, &fromlen);
                     else if(s6 >= 0 && FD_ISSET(s6, &readfds))
-                        rc = recvfrom(s6, (char*)buf.data(), buf.size(), 0, (struct sockaddr*)&from.first, &from.second);
+                        rc = recvfrom(s6, (char*)buf.data(), buf.size(), 0, (struct sockaddr*)&from, &fromlen);
                     else
                         break;
                     if (rc > 0) {
                         {
                             std::lock_guard<std::mutex> lck(sock_mtx);
-                            rcv.emplace_back(Blob {buf.begin(), buf.begin()+rc+1}, from);
+                            rcv.emplace_back(Blob {buf.begin(), buf.begin()+rc+1}, std::make_pair(from, fromlen));
                         }
                         cv.notify_all();
                     }
@@ -456,99 +429,78 @@ DhtRunner::doRun(const sockaddr_in* sin4, const sockaddr_in6* sin6, SecureDht::C
 }
 
 void
-DhtRunner::get(InfoHash hash, GetCallback vcb, DoneCallback dcb, Value::Filter f, Where w)
+DhtRunner::get(InfoHash hash, GetCallback vcb, DoneCallback dcb, Value::Filter f)
 {
-    {
-        std::lock_guard<std::mutex> lck(storage_mtx);
-        pending_ops.emplace([=](SecureDht& dht) mutable {
-            dht.get(hash, vcb, dcb, std::move(f), std::move(w));
-        });
-    }
+    std::lock_guard<std::mutex> lck(storage_mtx);
+    pending_ops.emplace([=](SecureDht& dht) mutable {
+        dht.get(hash, vcb, dcb, std::move(f));
+    });
     cv.notify_all();
 }
 
 void
-DhtRunner::get(const std::string& key, GetCallback vcb, DoneCallbackSimple dcb, Value::Filter f, Where w)
+DhtRunner::get(const std::string& key, GetCallback vcb, DoneCallbackSimple dcb, Value::Filter f)
 {
-    get(InfoHash::get(key), vcb, dcb, f, w);
-}
-void DhtRunner::query(const InfoHash& hash, QueryCallback cb, DoneCallback done_cb, Query q) {
-    {
-        std::lock_guard<std::mutex> lck(storage_mtx);
-        pending_ops.emplace([=](SecureDht& dht) mutable {
-            dht.query(hash, cb, done_cb, std::move(q));
-        });
-    }
-    cv.notify_all();
+    get(InfoHash::get(key), vcb, dcb, f);
 }
 
 std::future<size_t>
-DhtRunner::listen(InfoHash hash, GetCallback vcb, Value::Filter f, Where w)
+DhtRunner::listen(InfoHash hash, GetCallback vcb, Value::Filter f)
 {
+    std::lock_guard<std::mutex> lck(storage_mtx);
     auto ret_token = std::make_shared<std::promise<size_t>>();
-    {
-        std::lock_guard<std::mutex> lck(storage_mtx);
-        pending_ops.emplace([=](SecureDht& dht) mutable {
-            ret_token->set_value(dht.listen(hash, vcb, std::move(f), std::move(w)));
-        });
-    }
+    pending_ops.emplace([=](SecureDht& dht) mutable {
+        ret_token->set_value(dht.listen(hash, vcb, std::move(f)));
+    });
     cv.notify_all();
     return ret_token->get_future();
 }
 
 std::future<size_t>
-DhtRunner::listen(const std::string& key, GetCallback vcb, Value::Filter f, Where w)
+DhtRunner::listen(const std::string& key, GetCallback vcb, Value::Filter f)
 {
-    return listen(InfoHash::get(key), vcb, f, w);
+    return listen(InfoHash::get(key), vcb, f);
 }
 
 void
 DhtRunner::cancelListen(InfoHash h, size_t token)
 {
-    {
-        std::lock_guard<std::mutex> lck(storage_mtx);
-        pending_ops.emplace([=](SecureDht& dht) {
-            dht.cancelListen(h, token);
-        });
-    }
+    std::lock_guard<std::mutex> lck(storage_mtx);
+    pending_ops.emplace([=](SecureDht& dht) {
+        dht.cancelListen(h, token);
+    });
     cv.notify_all();
 }
 
 void
 DhtRunner::cancelListen(InfoHash h, std::shared_future<size_t> token)
 {
-    {
-        std::lock_guard<std::mutex> lck(storage_mtx);
-        pending_ops.emplace([=](SecureDht& dht) {
-            auto tk = token.get();
-            dht.cancelListen(h, tk);
-        });
-    }
+    std::lock_guard<std::mutex> lck(storage_mtx);
+    pending_ops.emplace([=](SecureDht& dht) {
+        auto tk = token.get();
+        dht.cancelListen(h, tk);
+    });
     cv.notify_all();
 }
 
 void
 DhtRunner::put(InfoHash hash, Value&& value, DoneCallback cb, time_point created, bool permanent)
 {
-    {
-        std::lock_guard<std::mutex> lck(storage_mtx);
-        auto sv = std::make_shared<Value>(std::move(value));
-        pending_ops.emplace([=](SecureDht& dht) {
-            dht.put(hash, sv, cb, created, permanent);
-        });
-    }
+    std::lock_guard<std::mutex> lck(storage_mtx);
+    auto sv = std::make_shared<Value>(std::move(value));
+    pending_ops.emplace([=](SecureDht& dht) {
+        dht.put(hash, sv, cb, created, permanent);
+    });
     cv.notify_all();
 }
 
 void
 DhtRunner::put(InfoHash hash, std::shared_ptr<Value> value, DoneCallback cb, time_point created, bool permanent)
 {
-    {
-        std::lock_guard<std::mutex> lck(storage_mtx);
-        pending_ops.emplace([=](SecureDht& dht) {
-            dht.put(hash, value, cb, created, permanent);
-        });
-    }
+    std::lock_guard<std::mutex> lck(storage_mtx);
+    pending_ops.emplace([=](SecureDht& dht) {
+        dht.put(hash, value, cb, created, permanent);
+    });
     cv.notify_all();
 }
 
@@ -561,24 +513,20 @@ DhtRunner::put(const std::string& key, Value&& value, DoneCallbackSimple cb, tim
 void
 DhtRunner::cancelPut(const InfoHash& h , const Value::Id& id)
 {
-    {
-        std::lock_guard<std::mutex> lck(storage_mtx);
-        pending_ops.emplace([=](SecureDht& dht) {
-            dht.cancelPut(h, id);
-        });
-    }
+    std::lock_guard<std::mutex> lck(storage_mtx);
+    pending_ops.emplace([=](SecureDht& dht) {
+        dht.cancelPut(h, id);
+    });
     cv.notify_all();
 }
 
 void
 DhtRunner::putSigned(InfoHash hash, std::shared_ptr<Value> value, DoneCallback cb)
 {
-    {
-        std::lock_guard<std::mutex> lck(storage_mtx);
-        pending_ops.emplace([=](SecureDht& dht) {
-            dht.putSigned(hash, value, cb);
-        });
-    }
+    std::lock_guard<std::mutex> lck(storage_mtx);
+    pending_ops.emplace([=](SecureDht& dht) {
+        dht.putSigned(hash, value, cb);
+    });
     cv.notify_all();
 }
 
@@ -597,12 +545,10 @@ DhtRunner::putSigned(const std::string& key, Value&& value, DoneCallbackSimple c
 void
 DhtRunner::putEncrypted(InfoHash hash, InfoHash to, std::shared_ptr<Value> value, DoneCallback cb)
 {
-    {
-        std::lock_guard<std::mutex> lck(storage_mtx);
-        pending_ops.emplace([=](SecureDht& dht) {
-            dht.putEncrypted(hash, to, value, cb);
-        });
-    }
+    std::lock_guard<std::mutex> lck(storage_mtx);
+    pending_ops.emplace([=](SecureDht& dht) {
+        dht.putEncrypted(hash, to, value, cb);
+    });
     cv.notify_all();
 }
 
@@ -618,77 +564,18 @@ DhtRunner::putEncrypted(const std::string& key, InfoHash to, Value&& value, Done
     putEncrypted(InfoHash::get(key), to, std::forward<Value>(value), cb);
 }
 
-void
-DhtRunner::tryBootstrapCoutinuously()
-{
-    if (bootstrap_thread.joinable()) {
-        if (bootstraping)
-            return; // already running
-        else
-            bootstrap_thread.join();
-    }
-    bootstraping = true;
-    bootstrap_thread = std::thread([this]() {
-        auto next = clock::now();
-        do {
-            decltype(bootstrap_nodes) nodes;
-            {
-                std::lock_guard<std::mutex> lck(bootstrap_mtx);
-                nodes = bootstrap_nodes;
-            }
-
-            next += BOOTSTRAP_PERIOD;
-            {
-                std::mutex mtx;
-                std::unique_lock<std::mutex> blck(mtx);
-                unsigned ping_count(0);
-                // Reverse: try last inserted bootstrap nodes first
-                for (auto it = nodes.rbegin(); it != nodes.rend(); it++) {
-                    ++ping_count;
-                    try {
-                        bootstrap(getAddrInfo(it->first, it->second), [&](bool) {
-                            if (not running)
-                                return;
-                            {
-                                std::unique_lock<std::mutex> blck(mtx);
-                                --ping_count;
-                            }
-                            bootstrap_cv.notify_all();
-                        });
-                    } catch (std::invalid_argument& e) {
-                        --ping_count;
-                        std::cerr << e.what() << std::endl;
-                    }
-                }
-                // wait at least until the next BOOTSTRAP_PERIOD
-                bootstrap_cv.wait_until(blck, next, [&]() { return not running; });
-                // wait for bootstrap requests to end.
-                if (running)
-                   bootstrap_cv.wait(blck, [&]() { return not running or ping_count == 0; });
-            }
-            // update state
-            {
-                std::lock_guard<std::mutex> lck(dht_mtx);
-                bootstraping = running and
-                               status4 == NodeStatus::Disconnected and
-                               status6 == NodeStatus::Disconnected;
-            }
-        } while (bootstraping);
-    });
-}
-
 std::vector<std::pair<sockaddr_storage, socklen_t>>
-DhtRunner::getAddrInfo(const std::string& host, const std::string& service)
+DhtRunner::getAddrInfo(const char* host, const char* service)
 {
     std::vector<std::pair<sockaddr_storage, socklen_t>> ips {};
-    if (host.empty())
+    if (not host or not service or strlen(host) == 0)
         return ips;
 
     addrinfo hints;
     memset(&hints, 0, sizeof(hints));
     hints.ai_socktype = SOCK_DGRAM;
     addrinfo* info = nullptr;
-    int rc = getaddrinfo(host.c_str(), service.c_str(), &hints, &info);
+    int rc = getaddrinfo(host, service, &hints, &info);
     if(rc != 0)
         throw std::invalid_argument(std::string("Error: `") + host + ":" + service + "`: " + gai_strerror(rc));
 
@@ -703,35 +590,18 @@ DhtRunner::getAddrInfo(const std::string& host, const std::string& service)
 }
 
 void
-DhtRunner::bootstrap(const std::string& host, const std::string& service)
+DhtRunner::bootstrap(const char* host, const char* service)
 {
-    std::lock_guard<std::mutex> lck(bootstrap_mtx);
-    bootstrap_nodes_all.emplace_back(host, service);
-    bootstrap_nodes.emplace_back(host, service);
-    tryBootstrapCoutinuously();
+    bootstrap(getAddrInfo(host, service));
 }
 
 void
-DhtRunner::clearBootstrap()
-{
-    std::lock_guard<std::mutex> lck(bootstrap_mtx);
-    bootstrap_nodes_all.clear();
-}
-
-void
-DhtRunner::bootstrap(const std::vector<std::pair<sockaddr_storage, socklen_t>>& nodes, DoneCallbackSimple&& cb)
+DhtRunner::bootstrap(const std::vector<std::pair<sockaddr_storage, socklen_t>>& nodes)
 {
     std::lock_guard<std::mutex> lck(storage_mtx);
-    pending_ops_prio.emplace([=](SecureDht& dht) mutable {
-        auto rem = cb ? std::make_shared<std::pair<size_t, bool>>(nodes.size(), false) : nullptr;
+    pending_ops_prio.emplace([=](SecureDht& dht) {
         for (auto& node : nodes)
-            dht.pingNode((sockaddr*)&node.first, node.second, cb ? [rem,cb](bool ok) {
-                auto& r = *rem;
-                r.first--;
-                r.second |= ok;
-                if (not r.first)
-                    cb(r.second);
-            } : DoneCallbackSimple{});
+            dht.pingNode((sockaddr*)&node.first, node.second);
     });
     cv.notify_all();
 }
@@ -739,36 +609,30 @@ DhtRunner::bootstrap(const std::vector<std::pair<sockaddr_storage, socklen_t>>&
 void
 DhtRunner::bootstrap(const std::vector<NodeExport>& nodes)
 {
-    {
-        std::lock_guard<std::mutex> lck(storage_mtx);
-        pending_ops_prio.emplace([=](SecureDht& dht) {
-            for (auto& node : nodes)
-                dht.insertNode(node);
-        });
-    }
+    std::lock_guard<std::mutex> lck(storage_mtx);
+    pending_ops_prio.emplace([=](SecureDht& dht) {
+        for (auto& node : nodes)
+            dht.insertNode(node);
+    });
     cv.notify_all();
 }
 
 void
 DhtRunner::connectivityChanged()
 {
-    {
-        std::lock_guard<std::mutex> lck(storage_mtx);
-        pending_ops.emplace([=](SecureDht& dht) {
-            dht.connectivityChanged();
-        });
-    }
+    std::lock_guard<std::mutex> lck(storage_mtx);
+    pending_ops.emplace([=](SecureDht& dht) {
+        dht.connectivityChanged();
+    });
     cv.notify_all();
 }
 
 void
 DhtRunner::findCertificate(InfoHash hash, std::function<void(const std::shared_ptr<crypto::Certificate>)> cb) {
-    {
-        std::lock_guard<std::mutex> lck(storage_mtx);
-        pending_ops.emplace([=](SecureDht& dht) {
-            dht.findCertificate(hash, cb);
-        });
-    }
+    std::lock_guard<std::mutex> lck(storage_mtx);
+    pending_ops.emplace([=](SecureDht& dht) {
+        dht.findCertificate(hash, cb);
+    });
     cv.notify_all();
 }
 

src/infohash.cpp

@@ -91,17 +91,4 @@ std::ostream& operator<< (std::ostream& s, const InfoHash& h)
     return s;
 }
 
-std::istream& operator>> (std::istream& s, InfoHash& h)
-{
-    std::array<char, HASH_LEN*2> dat;
-    s.exceptions(std::istream::eofbit | std::istream::failbit);
-    s.read(&(*dat.begin()), dat.size());
-    for (size_t i = 0; i < HASH_LEN; i++) {
-        unsigned res = 0;
-        sscanf(dat.data() + 2*i, "%02x", &res);
-        h[i] = res;
-    }
-    return s;
-}
-
 }

src/node.cpp

@@ -65,9 +65,10 @@ Node::getPendingMessageCount() const
 }
 
 void
-Node::update(const SockAddr& new_addr)
+Node::update(const sockaddr* sa, socklen_t salen)
 {
-    addr = new_addr;
+    std::copy_n((const uint8_t*)sa, salen, (uint8_t*)&ss);
+    sslen = salen;
 }
 
 /** To be called when a message was sent to the node */
@@ -118,7 +119,7 @@ Node::toString() const
 
 std::ostream& operator<< (std::ostream& s, const Node& h)
 {
-    s << h.id << " " << h.addr.toString();
+    s << h.id << " " << print_addr(h.ss, h.sslen);
     return s;
 }
 

src/node_cache.cpp

@@ -27,42 +27,10 @@ NodeCache::getNode(const InfoHash& id, sa_family_t family) {
 }
 
 std::shared_ptr<Node>
-NodeCache::getNode(const InfoHash& id, const SockAddr& addr, time_point now, bool confirm) {
+NodeCache::getNode(const InfoHash& id, const sockaddr* sa, socklen_t sa_len, time_point now, int confirm) {
     if (id == zeroes)
-        return std::make_shared<Node>(id, addr);
-    return (addr.getFamily() == AF_INET ? cache_4 : cache_6).getNode(id, addr, now, confirm);
-}
-
-std::vector<std::shared_ptr<Node>>
-NodeCache::getCachedNodes(const InfoHash& id, sa_family_t sa_f, size_t count) {
-    const auto& c = (sa_f == AF_INET ? cache_4 : cache_6);
-    auto it_p = c.lower_bound(id),
-         it_n = it_p;
-
-    std::vector<std::shared_ptr<Node>> nodes;
-    nodes.reserve(std::min(c.size(), count));
-    NodeMap::const_iterator it;
-
-    if (it_p != c.begin()) /* Create 2 separate iterator if we could */
-        --it_p;
-
-    while (nodes.size() < count and (it_n != c.end() or it_p != c.end())) {
-        /* If one of the iterator is at the end, then take the other one
-           If they are both in middle of somewhere comapre both and take
-           the closest to the id. */
-        if (it_p == c.end())       it = it_n++;
-        else if (it_n == c.end())  it = it_p--;
-        else                       it = id.xorCmp(it_p->first, it_n->first) < 0 ? it_p-- : it_n++;
-
-        if (it == c.begin())
-            it_p = c.end();
-
-        if (auto n = it->second.lock())
-            if ( not n->isExpired() )
-                nodes.emplace_back(std::move(n));
-    }
-
-    return nodes;
+        return std::make_shared<Node>(id, sa, sa_len);
+    return (sa->sa_family == AF_INET ? cache_4 : cache_6).getNode(id, sa, sa_len, now, confirm);
 }
 
 void
@@ -89,15 +57,15 @@ NodeCache::NodeMap::getNode(const InfoHash& id)
 }
 
 std::shared_ptr<Node>
-NodeCache::NodeMap::getNode(const InfoHash& id, const SockAddr& addr, time_point now, bool confirm)
+NodeCache::NodeMap::getNode(const InfoHash& id, const sockaddr* sa, socklen_t sa_len, time_point now, int confirm)
 {
     auto it = emplace(id, std::weak_ptr<Node>{});
     auto node = it.first->second.lock();
     if (not node) {
-        node = std::make_shared<Node>(id, addr);
+        node = std::make_shared<Node>(id, sa, sa_len);
         it.first->second = node;
     } else if (confirm || node->time < now - Node::NODE_EXPIRE_TIME) {
-        node->update(addr);
+        node->update(sa, sa_len);
     }
     return node;
 }

src/securedht.cpp

@@ -50,6 +50,12 @@ SecureDht::SecureDht(int s, int s6, SecureDht::Config conf)
     if (s < 0 && s6 < 0)
         return;
 
+#if GNUTLS_VERSION_NUMBER < 0x030300
+    int rc = gnutls_global_init();
+    if (rc != GNUTLS_E_SUCCESS)
+        throw DhtException(std::string("Error initializing GnuTLS: ")+gnutls_strerror(rc));
+#endif
+
     for (const auto& type : DEFAULT_TYPES)
         registerType(type);
 
@@ -70,12 +76,18 @@ SecureDht::SecureDht(int s, int s6, SecureDht::Config conf)
         }, [this](bool ok) {
             if (ok)
                 DHT_LOG.DEBUG("SecureDht: public key announced successfully");
+            else
+                DHT_LOG.ERR("SecureDht: error while announcing public key!");
         }, {}, true);
     }
 }
 
 SecureDht::~SecureDht()
-{}
+{
+#if GNUTLS_VERSION_NUMBER < 0x030300
+    gnutls_global_deinit();
+#endif
+}
 
 ValueType
 SecureDht::secureType(ValueType&& type)
@@ -87,7 +99,7 @@ SecureDht::secureType(ValueType&& type)
                 return false;
             }
             else
-                DHT_LOG.WARN("Signature verification succeeded");
+                DHT_LOG.WARN("Signature verification succeded");
         }
         return type.storePolicy(id, v, nid, a, al);
     };
@@ -129,7 +141,7 @@ SecureDht::getCertificate(const InfoHash& node) const
         return it->second;
 }
 
-const std::shared_ptr<const crypto::PublicKey>
+const std::shared_ptr<crypto::PublicKey>
 SecureDht::getPublicKey(const InfoHash& node) const
 {
     if (node == getId())
@@ -214,7 +226,7 @@ SecureDht::findCertificate(const InfoHash& node, std::function<void(const std::s
 }
 
 void
-SecureDht::findPublicKey(const InfoHash& node, std::function<void(const std::shared_ptr<const crypto::PublicKey>)> cb)
+SecureDht::findPublicKey(const InfoHash& node, std::function<void(const std::shared_ptr<crypto::PublicKey>)> cb)
 {
     auto pk = getPublicKey(node);
     if (pk && *pk) {
@@ -279,15 +291,15 @@ SecureDht::getCallbackFilter(GetCallback cb, Value::Filter&& filter)
 }
 
 void
-SecureDht::get(const InfoHash& id, GetCallback cb, DoneCallback donecb, Value::Filter&& f, Where&& w)
+SecureDht::get(const InfoHash& id, GetCallback cb, DoneCallback donecb, Value::Filter&& f)
 {
-    Dht::get(id, getCallbackFilter(cb, std::forward<Value::Filter>(f)), donecb, {}, std::forward<Where>(w));
+    Dht::get(id, getCallbackFilter(cb, std::forward<Value::Filter>(f)), donecb);
 }
 
 size_t
-SecureDht::listen(const InfoHash& id, GetCallback cb, Value::Filter&& f, Where&& w)
+SecureDht::listen(const InfoHash& id, GetCallback cb, Value::Filter&& f)
 {
-    return Dht::listen(id, getCallbackFilter(cb, std::forward<Value::Filter>(f)), {}, std::forward<Where>(w));
+    return Dht::listen(id, getCallbackFilter(cb, std::forward<Value::Filter>(f)));
 }
 
 void
@@ -321,7 +333,7 @@ SecureDht::putSigned(const InfoHash& hash, std::shared_ptr<Value> val, DoneCallb
         },
         [hash,val,this,callback,permanent] (bool /* ok */) {
             sign(*val);
-            put(hash, val, callback, time_point::max(), permanent);
+            put(hash, val, callback, {}, permanent);
         },
         Value::IdFilter(val->id)
     );
@@ -330,7 +342,7 @@ SecureDht::putSigned(const InfoHash& hash, std::shared_ptr<Value> val, DoneCallb
 void
 SecureDht::putEncrypted(const InfoHash& hash, const InfoHash& to, std::shared_ptr<Value> val, DoneCallback callback, bool permanent)
 {
-    findPublicKey(to, [=](const std::shared_ptr<const crypto::PublicKey> pk) {
+    findPublicKey(to, [=](const std::shared_ptr<crypto::PublicKey> pk) {
         if(!pk || !*pk) {
             if (callback)
                 callback(false, {});
@@ -338,7 +350,7 @@ SecureDht::putEncrypted(const InfoHash& hash, const InfoHash& to, std::shared_pt
         }
         DHT_LOG.WARN("Encrypting data for PK: %s", pk->getId().toString().c_str());
         try {
-            put(hash, encrypt(*val, *pk), callback, time_point::max(), permanent);
+            put(hash, encrypt(*val, *pk), callback, {}, permanent);
         } catch (const std::exception& e) {
             DHT_LOG.ERR("Error putting encrypted data: %s", e.what());
             if (callback)
@@ -350,13 +362,22 @@ SecureDht::putEncrypted(const InfoHash& hash, const InfoHash& to, std::shared_pt
 void
 SecureDht::sign(Value& v) const
 {
-    v.sign(*key_);
+    if (v.isEncrypted())
+        throw DhtException("Can't sign encrypted data.");
+    v.owner = std::make_shared<crypto::PublicKey>(key_->getPublicKey());
+    v.signature = key_->sign(v.getToSign());
 }
 
 Value
 SecureDht::encrypt(Value& v, const crypto::PublicKey& to) const
 {
-    return v.encrypt(*key_, to);
+    if (v.isEncrypted())
+        throw DhtException("Data is already encrypted.");
+    v.setRecipient(to.getId());
+    sign(v);
+    Value nv {v.id};
+    nv.setCypher(to.encrypt(v.getToEncrypt()));
+    return nv;
 }
 
 Value

src/utils.cpp

@@ -18,7 +18,6 @@
  */
 
 #include "utils.h"
-#include "sockaddr.h"
 #include "default_types.h"
 
 namespace dht {
@@ -48,14 +47,10 @@ print_addr(const sockaddr_storage& ss, socklen_t sslen)
 }
 
 std::string
-printAddr(const SockAddr& addr) {
+printAddr(const Address& addr) {
     return print_addr((const sockaddr*)&addr.first, addr.second);
 }
 
-bool operator==(const SockAddr& a, const SockAddr& b) {
-    return a.equals(b);
-}
-
 time_point from_time_t(std::time_t t) {
     return clock::now() + (std::chrono::system_clock::from_time_t(t) - std::chrono::system_clock::now());
 }
@@ -90,15 +85,4 @@ unpackMsg(Blob b) {
     return msgpack::unpack((const char*)b.data(), b.size());
 }
 
-msgpack::object*
-findMapValue(msgpack::object& map, const std::string& key) {
-    if (map.type != msgpack::type::MAP) throw msgpack::type_error();
-    for (unsigned i = 0; i < map.via.map.size; i++) {
-        auto& o = map.via.map.ptr[i];
-        if (o.key.type == msgpack::type::STR && o.key.as<std::string>() == key)
-            return &o.val;
-    }
-    return nullptr;
-}
-
 }

src/value.cpp

@@ -25,8 +25,6 @@
 
 namespace dht {
 
-const std::string Query::QUERY_PARSE_ERROR {"Error parsing query."};
-
 std::ostream& operator<< (std::ostream& s, const Value& v)
 {
     s << "Value[id:" << std::hex << v.id << std::dec << " ";
@@ -65,7 +63,7 @@ const ValueType ValueType::USER_DATA = {0, "User Data"};
 bool
 ValueType::DEFAULT_STORE_POLICY(InfoHash, std::shared_ptr<Value>& v, InfoHash, const sockaddr*, socklen_t)
 {
-    return v->size() <= MAX_VALUE_SIZE;
+    return v->data.size() <= MAX_VALUE_SIZE;
 }
 
 msgpack::object*
@@ -144,9 +142,8 @@ Value::msgpack_unpack_body(const msgpack::object& o)
                 seq = rseq->as<decltype(seq)>();
             else
                 throw msgpack::type_error();
-            crypto::PublicKey new_owner;
-            new_owner.msgpack_unpack(*rowner);
-            owner = std::make_shared<const crypto::PublicKey>(std::move(new_owner));
+            owner = std::make_shared<crypto::PublicKey>();
+            owner->msgpack_unpack(*rowner);
             if (auto rrecipient = findMapValue(*rbody, "to")) {
                 recipient = rrecipient->as<InfoHash>();
             }
@@ -159,329 +156,4 @@ Value::msgpack_unpack_body(const msgpack::object& o)
     }
 }
 
-bool
-FieldValue::operator==(const FieldValue& vfd) const
-{
-    if (field != vfd.field)
-        return false;
-    switch (field) {
-        case Value::Field::Id:
-        case Value::Field::ValueType:
-        case Value::Field::SeqNum:
-            return intValue == vfd.intValue;
-        case Value::Field::OwnerPk:
-            return hashValue == vfd.hashValue;
-        case Value::Field::UserType:
-            return blobValue == vfd.blobValue;
-        case Value::Field::None:
-            return true;
-        default:
-            return false;
-    }
 }
-
-Value::Filter
-FieldValue::getLocalFilter() const
-{
-    switch (field) {
-        case Value::Field::Id:
-            return Value::IdFilter(intValue);
-        case Value::Field::ValueType:
-            return Value::TypeFilter(intValue);
-        case Value::Field::OwnerPk:
-            return Value::OwnerFilter(hashValue);
-        case Value::Field::SeqNum:
-            return Value::SeqNumFilter(intValue);
-        case Value::Field::UserType:
-            return Value::UserTypeFilter(std::string {blobValue.begin(), blobValue.end()});
-        default:
-            return Value::AllFilter();
-    }
-}
-
-FieldValueIndex::FieldValueIndex(const Value& v, Select s)
-{
-    auto selection = s.getSelection();
-    if (not selection.empty()) {
-        std::transform(selection.begin(), selection.end(), std::inserter(index, index.end()),
-            [](const std::set<Value::Field>::value_type& f) {
-                return std::make_pair(f, FieldValue {});
-        });
-    } else {
-        index.clear();
-        for (size_t f = 1 ; f < static_cast<int>(Value::Field::COUNT) ; ++f)
-            index[static_cast<Value::Field>(f)] = {};
-    }
-    for (const auto& fvp : index) {
-        const auto& f = fvp.first;
-        switch (f) {
-            case Value::Field::Id:
-                index[f] = {f, v.id};
-                break;
-            case Value::Field::ValueType:
-                index[f] = {f, v.type};
-                break;
-            case Value::Field::OwnerPk:
-                index[f] = {f, v.owner ? v.owner->getId() : InfoHash() };
-                break;
-            case Value::Field::SeqNum:
-                index[f] = {f, v.seq};
-                break;
-            case Value::Field::UserType:
-                index[f] = {f, Blob {v.user_type.begin(), v.user_type.end()}};
-                break;
-            default:
-                break;
-        }
-    }
-}
-
-bool FieldValueIndex::containedIn(const FieldValueIndex& other) const {
-    if (index.size() > other.index.size())
-        return false;
-    for (const auto& field : index) {
-        auto other_field = other.index.find(field.first);
-        if (other_field == other.index.end())
-            return false;
-    }
-    return true;
-}
-
-std::ostream& operator<<(std::ostream& os, const FieldValueIndex& fvi) {
-    os << "Index[";
-    for (auto v = fvi.index.begin(); v != fvi.index.end(); ++v) {
-        switch (v->first) {
-            case Value::Field::Id:
-                os << "Id:" << std::hex << v->second.getInt();
-                break;
-            case Value::Field::ValueType:
-                os << "ValueType:" << v->second.getInt();
-                break;
-            case Value::Field::OwnerPk:
-                os << "Owner:" << v->second.getHash().toString();
-                break;
-            case Value::Field::SeqNum:
-                os << "Seq:" << v->second.getInt();
-                break;
-            case Value::Field::UserType: {
-                auto ut = v->second.getBlob();
-                os << "UserType:" << std::string(ut.begin(), ut.end());
-                break;
-            }
-            default:
-                break;
-        }
-        os << (std::next(v) != fvi.index.end() ? "," : "");
-    }
-    return os << "]";
-}
-
-void
-FieldValueIndex::msgpack_unpack_fields(const std::set<Value::Field>& fields, const msgpack::object& o, unsigned offset)
-{
-    index.clear();
-
-    unsigned j = 0;
-    for (const auto& field : fields) {
-        auto& field_value = o.via.array.ptr[offset+(j++)];
-        switch (field) {
-            case Value::Field::Id:
-            case Value::Field::ValueType:
-            case Value::Field::SeqNum:
-                index[field] = FieldValue(field, field_value.as<uint64_t>());
-                break;
-            case Value::Field::OwnerPk:
-                index[field] = FieldValue(field, field_value.as<InfoHash>());
-                break;
-            case Value::Field::UserType:
-                index[field] = FieldValue(field, field_value.as<Blob>());
-                break;
-            default:
-                throw msgpack::type_error();
-        }
-    }
-}
-
-void trim_str(std::string& str) {
-    auto first = std::min(str.size(), str.find_first_not_of(" "));
-    auto last = std::min(str.size(), str.find_last_not_of(" "));
-    str = str.substr(first, last - first + 1);
-}
-
-Select::Select(const std::string& q_str) {
-    std::istringstream q_iss {q_str};
-    std::string token {};
-    q_iss >> token;
-
-    if (token == "SELECT" or token == "select") {
-        q_iss >> token;
-        std::istringstream fields {token};
-
-        while (std::getline(fields, token, ',')) {
-            trim_str(token);
-            if (token == "id")
-                field(Value::Field::Id);
-            else if (token == "value_type")
-                field(Value::Field::ValueType);
-            else if (token == "owner_pk")
-                field(Value::Field::OwnerPk);
-            if (token == "seq")
-                field(Value::Field::SeqNum);
-            else if (token == "user_type")
-                field(Value::Field::UserType);
-        }
-    }
-}
-
-Where::Where(const std::string& q_str) {
-    std::istringstream q_iss {q_str};
-    std::string token {};
-    q_iss >> token;
-    if (token == "WHERE" or token == "where") {
-        std::getline(q_iss, token);
-        std::istringstream restrictions {token};
-        while (std::getline(restrictions, token, ',')) {
-            trim_str(token);
-            std::istringstream eq_ss {token};
-            std::string field_str, value_str;
-            std::getline(eq_ss, field_str, '=');
-            trim_str(field_str);
-            std::getline(eq_ss, value_str, '=');
-            trim_str(value_str);
-
-            if (not value_str.empty()) {
-                uint64_t v = 0;
-                std::string s {};
-                std::istringstream convert {value_str};
-                convert >> v;
-                if (convert.failbit
-                        and value_str.size() > 1
-                        and value_str[0] == '\"'
-                        and value_str[value_str.size()-1] == '\"')
-                    s = value_str.substr(1, value_str.size()-2);
-                else
-                    s = value_str;
-                if (field_str == "id")
-                    id(v);
-                else if (field_str == "value_type")
-                    valueType(v);
-                else if (field_str == "owner_pk")
-                    owner(InfoHash(s));
-                else if (field_str == "seq")
-                    seq(v);
-                else if (field_str == "user_type")
-                    userType(s);
-                else
-                    throw std::invalid_argument(Query::QUERY_PARSE_ERROR + " (WHERE) wrong token near: " + field_str);
-            }
-        }
-    }
-}
-
-void
-Query::msgpack_unpack(const msgpack::object& o)
-{
-	if (o.type != msgpack::type::MAP)
-		throw msgpack::type_error();
-
-	auto rfilters = findMapValue(o, "w"); /* unpacking filters */
-	if (rfilters)
-        where.msgpack_unpack(*rfilters);
-	else
-		throw msgpack::type_error();
-
-	auto rfield_selector = findMapValue(o, "s"); /* unpacking field selectors */
-	if (rfield_selector)
-        select.msgpack_unpack(*rfield_selector);
-	else
-		throw msgpack::type_error();
-}
-
-template <typename T>
-bool subset(std::vector<T> fds, std::vector<T> qfds)
-{
-    for (auto& fd : fds) {
-        auto correspondance = std::find_if(qfds.begin(), qfds.end(), [&fd](T& _vfd) { return fd == _vfd; });
-        if (correspondance == qfds.end())
-            return false;
-    }
-    return true;
-}
-
-bool Select::isSatisfiedBy(const Select& os) const {
-    /* empty, means all values are selected. */
-    if (fieldSelection_.empty() and not os.fieldSelection_.empty())
-        return false;
-    else
-        return subset(fieldSelection_, os.fieldSelection_);
-}
-
-bool Where::isSatisfiedBy(const Where& ow) const {
-    return subset(ow.filters_, filters_);
-}
-
-bool Query::isSatisfiedBy(const Query& q) const {
-    return none or (where.isSatisfiedBy(q.where) and select.isSatisfiedBy(q.select));
-}
-
-std::ostream& operator<<(std::ostream& s, const dht::Select& select) {
-    s << "SELECT " << (select.fieldSelection_.empty() ? "*" : "");
-    for (auto fs = select.fieldSelection_.begin() ; fs != select.fieldSelection_.end() ; ++fs) {
-        switch (fs->getField()) {
-            case Value::Field::Id:
-                s << "id";
-                break;
-            case Value::Field::ValueType:
-                s << "value_type";
-                break;
-            case Value::Field::UserType:
-                s << "user_type";
-                break;
-            case Value::Field::OwnerPk:
-                s << "owner_public_key";
-                break;
-            case Value::Field::SeqNum:
-                s << "seq";
-                break;
-            default:
-                break;
-        }
-        s << (std::next(fs) != select.fieldSelection_.end() ? "," : "");
-    }
-    return s;
-}
-
-std::ostream& operator<<(std::ostream& s, const dht::Where& where) {
-    if (not where.filters_.empty()) {
-        s << "WHERE ";
-        for (auto f = where.filters_.begin() ; f != where.filters_.end() ; ++f) {
-            switch (f->getField()) {
-                case Value::Field::Id:
-                    s << "id=" << f->getInt();
-                    break;
-                case Value::Field::ValueType:
-                    s << "value_type=" << f->getInt();
-                    break;
-                case Value::Field::OwnerPk:
-                    s << "owner_pk_hash=" << f->getHash().toString();
-                    break;
-                case Value::Field::SeqNum:
-                    s << "seq=" << f->getInt();
-                    break;
-                case Value::Field::UserType: {
-                    auto b = f->getBlob();
-                    s << "user_type=" << std::string {b.begin(), b.end()};
-                    break;
-                }
-                default:
-                    break;
-            }
-            s << (std::next(f) != where.filters_.end() ? "," : "");
-        }
-    }
-    return s;
-}
-
-
-}
-

tools/Makefile.am

@@ -1,13 +1,12 @@
 bin_PROGRAMS = dhtnode dhtchat dhtscanner
-noinst_HEADERS = tools_common.h
 
 AM_CPPFLAGS = -I../include
 
 dhtnode_SOURCES = dhtnode.cpp
-dhtnode_LDFLAGS = -lopendht -lreadline -L../src/.libs @GnuTLS_LIBS@
+dhtnode_LDFLAGS = -lopendht -lreadline -L../src/.libs @GNUTLS_LIBS@
 
 dhtchat_SOURCES = dhtchat.cpp
-dhtchat_LDFLAGS = -lopendht -lreadline -L../src/.libs @GnuTLS_LIBS@
+dhtchat_LDFLAGS = -lopendht -lreadline -L../src/.libs  @GNUTLS_LIBS@
 
 dhtscanner_SOURCES = dhtscanner.cpp
-dhtscanner_LDFLAGS = -lopendht -lreadline -L../src/.libs @GnuTLS_LIBS@
+dhtscanner_LDFLAGS = -lopendht -lreadline -L../src/.libs  @GNUTLS_LIBS@

tools/dhtchat.cpp

@@ -48,8 +48,12 @@ main(int argc, char **argv)
 {
     auto params = parseArgs(argc, argv);
 
+    // TODO: remove with GnuTLS >= 3.3
+    if (int rc = gnutls_global_init())
+        throw std::runtime_error(std::string("Error initializing GnuTLS: ")+gnutls_strerror(rc));
+
     DhtRunner dht;
-    dht.run(params.port, dht::crypto::generateIdentity("DHT Chat Node"), true, params.network);
+    dht.run(params.port, dht::crypto::generateIdentity("DHT Chat Node"), true);
 
     if (not params.bootstrap.first.empty())
         dht.bootstrap(params.bootstrap.first.c_str(), params.bootstrap.second.c_str());
@@ -125,5 +129,6 @@ main(int argc, char **argv)
 
     std::cout << std::endl <<  "Stopping node..." << std::endl;
     dht.join();
+    gnutls_global_deinit();
     return 0;
 }

tools/dhtnode.cpp

@@ -29,7 +29,7 @@ extern "C" {
 using namespace dht;
 
 void print_usage() {
-    std::cout << "Usage: dhtnode [-v [-l logfile]] [-i] [-d] [-n network_id] [-p local_port] [-b bootstrap_host[:port]]" << std::endl << std::endl;
+    std::cout << "Usage: dhtnode [-p local_port] [-b bootstrap_host:port]" << std::endl << std::endl;
     std::cout << "dhtnode, a simple OpenDHT command line node runner." << std::endl;
     std::cout << "Report bugs to: http://opendht.net" << std::endl;
 }
@@ -40,6 +40,8 @@ void print_id_req() {
 
 void print_node_info(const std::shared_ptr<DhtRunner>& dht, const dht_params& params) {
     std::cout << "OpenDht node " << dht->getNodeId() << " running on port " <<  dht->getBoundPort() << std::endl;
+    if (params.is_bootstrap_node)
+        std::cout << "Running in bootstrap mode (discouraged)." << std::endl;
     if (params.generate_identity)
         std::cout << "Public key ID " << dht->getId() << std::endl;
 }
@@ -48,7 +50,7 @@ void print_help() {
     std::cout << "OpenDht command line interface (CLI)" << std::endl;
     std::cout << "Possible commands:" << std::endl
               << "  h, help    Print this help message." << std::endl
-              << "  x, quit    Quit the program." << std::endl
+              << "  q, quit    Quit the program." << std::endl
               << "  log        Start/stop printing DHT logs." << std::endl;
 
     std::cout << std::endl << "Node information:" << std::endl
@@ -71,7 +73,7 @@ void print_help() {
               << std::endl;
 }
 
-void cmd_loop(std::shared_ptr<DhtRunner>& dht, dht_params& params)
+void cmd_loop(std::shared_ptr<DhtRunner>& dht, std::map<std::string, indexation::Pht> indexes, dht_params& params)
 {
     print_node_info(dht, params);
     std::cout << " (type 'h' or 'help' for a list of possible commands)" << std::endl << std::endl;
@@ -79,8 +81,6 @@ void cmd_loop(std::shared_ptr<DhtRunner>& dht, dht_params& params)
     // using the GNU History API
     using_history();
 
-    std::map<std::string, indexation::Pht> indexes;
-
     while (true)
     {
         // using the GNU Readline API
@@ -92,7 +92,7 @@ void cmd_loop(std::shared_ptr<DhtRunner>& dht, dht_params& params)
         std::string op, idstr, value, index, keystr;
         iss >> op;
 
-        if (op == "x" || op == "exit" || op == "quit") {
+        if (op == "x" || op == "q" || op == "exit" || op == "quit") {
             break;
         } else if (op == "h" || op == "help") {
             print_help();
@@ -113,12 +113,7 @@ void cmd_loop(std::shared_ptr<DhtRunner>& dht, dht_params& params)
             std::cout << dht->getRoutingTablesLog(AF_INET6) << std::endl;
             continue;
         } else if (op == "ld") {
-            iss >> idstr;
-            InfoHash filter(idstr);
-            if (filter == InfoHash{})
-                std::cout << dht->getStorageLog() << std::endl;
-            else
-                std::cout << dht->getStorageLog(filter) << std::endl;
+            std::cout << dht->getStorageLog() << std::endl;
             continue;
         } else if (op == "ls") {
             std::cout << "Searches:" << std::endl;
@@ -145,14 +140,11 @@ void cmd_loop(std::shared_ptr<DhtRunner>& dht, dht_params& params)
             }
             continue;
         } else if (op == "log") {
-            iss >> idstr;
-            InfoHash filter(idstr);
-            params.log = filter == InfoHash{} ? !params.log : true;
+            params.log = !params.log;
             if (params.log)
                 log::enableLogging(*dht);
             else
                 log::disableLogging(*dht);
-            dht->setLogFilter(filter);
             continue;
         }
 
@@ -200,8 +192,6 @@ void cmd_loop(std::shared_ptr<DhtRunner>& dht, dht_params& params)
         // Dht
         auto start = std::chrono::high_resolution_clock::now();
         if (op == "g") {
-            std::string rem;
-            std::getline(iss, rem);
             dht->get(id, [start](std::shared_ptr<Value> value) {
                 auto now = std::chrono::high_resolution_clock::now();
                 std::cout << "Get: found value (after " << print_dt(now-start) << "s)" << std::endl;
@@ -210,31 +200,14 @@ void cmd_loop(std::shared_ptr<DhtRunner>& dht, dht_params& params)
             }, [start](bool ok) {
                 auto end = std::chrono::high_resolution_clock::now();
                 std::cout << "Get: " << (ok ? "completed" : "failure") << " (took " << print_dt(end-start) << "s)" << std::endl;
-            }, {}, dht::Where {std::move(rem)});
-        }
-        else if (op == "q") {
-            std::string rem;
-            std::getline(iss, rem);
-            dht->query(id, [start](const std::vector<std::shared_ptr<FieldValueIndex>>& field_value_indexes) {
-                auto now = std::chrono::high_resolution_clock::now();
-                for (auto& index : field_value_indexes) {
-                    std::cout << "Query: found field value index (after " << print_dt(now-start) << "s)" << std::endl;
-                    std::cout << "\t" << *index << std::endl;
-                }
-                return true;
-            }, [start](bool ok) {
-                auto end = std::chrono::high_resolution_clock::now();
-                std::cout << "Query: " << (ok ? "completed" : "failure") << " (took " << print_dt(end-start) << "s)" << std::endl;
-            }, dht::Query {std::move(rem)});
+            });
         }
         else if (op == "l") {
-            std::string rem;
-            std::getline(iss, rem);
             dht->listen(id, [](std::shared_ptr<Value> value) {
                 std::cout << "Listen: found value:" << std::endl;
                 std::cout << "\t" << *value << std::endl;
                 return true;
-            }, {}, dht::Where {std::move(rem)});
+            });
         }
         else if (op == "p") {
             std::string v;
@@ -339,7 +312,11 @@ void cmd_loop(std::shared_ptr<DhtRunner>& dht, dht_params& params)
 int
 main(int argc, char **argv)
 {
+    if (int rc = gnutls_global_init())  // TODO: remove with GnuTLS >= 3.3
+        throw std::runtime_error(std::string("Error initializing GnuTLS: ")+gnutls_strerror(rc));
+
     auto dht = std::make_shared<DhtRunner>();
+    std::map<std::string, indexation::Pht> indexes;
 
     try {
         auto params = parseArgs(argc, argv);
@@ -357,7 +334,7 @@ main(int argc, char **argv)
             crt = dht::crypto::generateIdentity("DHT Node", ca_tmp);
         }
 
-        dht->run(params.port, crt, true, params.network);
+        dht->run(params.port, crt, true, params.is_bootstrap_node);
 
         if (params.log) {
             if (not params.logfile.empty())
@@ -375,7 +352,7 @@ main(int argc, char **argv)
             while (true)
                 std::this_thread::sleep_for(std::chrono::seconds(30));
         } else {
-            cmd_loop(dht, params);
+            cmd_loop(dht, indexes, params);
         }
 
     } catch(const std::exception&e) {
@@ -398,5 +375,7 @@ main(int argc, char **argv)
     cv.wait(lk, [&](){ return done.load(); });
 
     dht->join();
+    gnutls_global_deinit();
+
     return 0;
 }

tools/dhtscanner.cpp

@@ -70,11 +70,17 @@ int
 main(int argc, char **argv)
 {
     auto params = parseArgs(argc, argv);
+
+    // TODO: remove with GnuTLS >= 3.3
+    int rc = gnutls_global_init();
+    if (rc != GNUTLS_E_SUCCESS)
+        throw std::runtime_error(std::string("Error initializing GnuTLS: ")+gnutls_strerror(rc));
+
     auto ca_tmp = dht::crypto::generateIdentity("DHT Node CA");
     auto crt_tmp = dht::crypto::generateIdentity("Scanner node", ca_tmp);
 
     DhtRunner dht;
-    dht.run(params.port, crt_tmp, true, params.network);
+    dht.run(params.port, crt_tmp, true, [](dht::NodeStatus /* ipv4 */, dht::NodeStatus /* ipv6 */) {});
 
     if (not params.bootstrap.first.empty())
         dht.bootstrap(params.bootstrap.first.c_str(), params.bootstrap.second.c_str());
@@ -104,5 +110,6 @@ main(int argc, char **argv)
         std::cout << "Node " << *n << std::endl;
 
     dht.join();
+    gnutls_global_deinit();
     return 0;
 }

tools/tools_common.h

@@ -14,7 +14,8 @@
  *  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 <http://www.gnu.org/licenses/>.
+ *  along with this program; if not, write to the Free Software
+ *  Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301 USA.
  */
 
 // Common utility methods used by C++ OpenDHT tools.
@@ -36,6 +37,81 @@
 #include <sstream>
 #include <fstream>
 
+/**
+ * Terminal colors for logging
+ */
+namespace Color {
+    enum Code {
+        FG_RED      = 31,
+        FG_GREEN    = 32,
+        FG_YELLOW   = 33,
+        FG_BLUE     = 34,
+        FG_DEFAULT  = 39,
+        BG_RED      = 41,
+        BG_GREEN    = 42,
+        BG_BLUE     = 44,
+        BG_DEFAULT  = 49
+    };
+    class Modifier {
+        const Code code;
+    public:
+        constexpr Modifier(Code pCode) : code(pCode) {}
+        friend std::ostream&
+        operator<<(std::ostream& os, const Modifier& mod) {
+            return os << "\033[" << mod.code << 'm';
+        }
+    };
+}
+
+constexpr const Color::Modifier def(Color::FG_DEFAULT);
+constexpr const Color::Modifier red(Color::FG_RED);
+constexpr const Color::Modifier yellow(Color::FG_YELLOW);
+
+/**
+ * Print va_list to std::ostream (used for logging).
+ */
+void
+printLog(std::ostream& s, char const* m, va_list args) {
+    static constexpr int BUF_SZ = 8192;
+    char buffer[BUF_SZ];
+    int ret = vsnprintf(buffer, sizeof(buffer), m, args);
+    if (ret < 0)
+        return;
+    s.write(buffer, std::min(ret, BUF_SZ));
+    if (ret >= BUF_SZ)
+        s << "[[TRUNCATED]]";
+    s.put('\n');
+}
+
+void
+enableLogging(dht::DhtRunner& dht)
+{
+    dht.setLoggers(
+        [](char const* m, va_list args){ std::cerr << red; printLog(std::cerr, m, args); std::cerr << def; },
+        [](char const* m, va_list args){ std::cout << yellow; printLog(std::cout, m, args); std::cout << def; },
+        [](char const* m, va_list args){ printLog(std::cout, m, args); }
+    );
+}
+
+void
+enableFileLogging(dht::DhtRunner& dht, const std::string& path)
+{
+    auto logfile = std::make_shared<std::fstream>();
+    logfile->open(path, std::ios::out);
+
+    dht.setLoggers(
+        [=](char const* m, va_list args){ printLog(*logfile, m, args); },
+        [=](char const* m, va_list args){ printLog(*logfile, m, args); },
+        [=](char const* m, va_list args){ printLog(*logfile, m, args); }
+    );
+}
+
+void
+disableLogging(dht::DhtRunner& dht)
+{
+    dht.setLoggers(dht::NOLOG, dht::NOLOG, dht::NOLOG);
+}
+
 /**
  * Split "[host]:port" or "host:port" to pair<"host", "port">.
  */
@@ -107,29 +183,27 @@ struct dht_params {
     bool log {false};
     std::string logfile {};
     in_port_t port {0};
-    dht::NetId network {0};
+    bool is_bootstrap_node {false};
     bool generate_identity {false};
     bool daemonize {false};
     std::pair<std::string, std::string> bootstrap {};
 };
 
 static const constexpr struct option long_options[] = {
-   {"help",       no_argument      , nullptr, 'h'},
+   {"help",       no_argument,       nullptr, 'h'},
    {"port",       required_argument, nullptr, 'p'},
-   {"net",        required_argument, nullptr, 'n'},
-   {"bootstrap",  required_argument, nullptr, 'b'},
+   {"bootstrap",  optional_argument, nullptr, 'b'},
    {"identity",   no_argument      , nullptr, 'i'},
    {"verbose",    no_argument      , nullptr, 'v'},
    {"daemonize",  no_argument      , nullptr, 'd'},
-   {"logfile",    required_argument, nullptr, 'l'},
-   {nullptr,      0                , nullptr,  0}
+   {nullptr,      0,                 nullptr,  0}
 };
 
 dht_params
 parseArgs(int argc, char **argv) {
     dht_params params;
     int opt;
-    while ((opt = getopt_long(argc, argv, "hidvp:n:b:l:", long_options, nullptr)) != -1) {
+    while ((opt = getopt_long(argc, argv, ":hidv:p:b:", long_options, nullptr)) != -1) {
         switch (opt) {
         case 'p': {
                 int port_arg = atoi(optarg);
@@ -139,24 +213,24 @@ parseArgs(int argc, char **argv) {
                     std::cout << "Invalid port: " << port_arg << std::endl;
             }
             break;
-        case 'n':
-            params.network = strtoul(optarg, nullptr, 0);
-            break;
         case 'b':
-            params.bootstrap = splitPort((optarg[0] == '=') ? optarg+1 : optarg);
-            if (not params.bootstrap.first.empty() and params.bootstrap.second.empty()) {
-                std::stringstream ss;
-                ss << DHT_DEFAULT_PORT;
-                params.bootstrap.second = ss.str();
+            if (optarg) {
+                params.bootstrap = splitPort((optarg[0] == '=') ? optarg+1 : optarg);
+                if (not params.bootstrap.first.empty() and params.bootstrap.second.empty()) {
+                    std::stringstream ss;
+                    ss << DHT_DEFAULT_PORT;
+                    params.bootstrap.second = ss.str();
+                }
             }
+            else
+                params.is_bootstrap_node = true;
             break;
         case 'h':
             params.help = true;
             break;
-        case 'l':
-            params.logfile = optarg;
-            break;
         case 'v':
+            if (optarg)
+                params.logfile = optarg;
             params.log = true;
             break;
         case 'i':
@@ -165,6 +239,19 @@ parseArgs(int argc, char **argv) {
         case 'd':
             params.daemonize = true;
             break;
+        case ':':
+            switch (optopt) {
+            case 'b':
+                params.is_bootstrap_node = true;
+                break;
+            case 'v':
+                params.log = true;
+                break;
+            default:
+                std::cout << "option requires an argument -- '" << optopt << '\'' << std::endl;
+                break;
+            }
+            break;
         default:
             break;
         }