mirror of
https://github.com/savoirfairelinux/opendht.git
synced 2026-11-09 12:57:03 +00:00
Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
ebf2362b00 | ||
|
|
619a2d89e2 |
No files matched your search
@@ -76,5 +76,4 @@ doc/Doxyfile
|
||||
|
||||
# build dir
|
||||
build
|
||||
build_dev
|
||||
.DS_Store
|
||||
Vendored
+8
-8
@@ -5,9 +5,9 @@
|
||||
"name": "dhtnode",
|
||||
"type": "cppdbg",
|
||||
"request": "launch",
|
||||
"program": "${workspaceFolder}/build_dev/tools/dhtnode",
|
||||
"program": "${workspaceFolder}/build/tools/dhtnode",
|
||||
"args": ["-v", "-p", "4222"],
|
||||
"cwd": "${workspaceFolder}/build_dev",
|
||||
"cwd": "${workspaceFolder}/build",
|
||||
"environment": [],
|
||||
"setupCommands": [
|
||||
{
|
||||
@@ -30,8 +30,8 @@
|
||||
"name": "tests",
|
||||
"type": "cppdbg",
|
||||
"request": "launch",
|
||||
"program": "${workspaceFolder}/build_dev/opendht_unit_tests",
|
||||
"cwd": "${workspaceFolder}/build_dev",
|
||||
"program": "${workspaceFolder}/build/opendht_unit_tests",
|
||||
"cwd": "${workspaceFolder}/build",
|
||||
"environment": [],
|
||||
"setupCommands": [
|
||||
{
|
||||
@@ -41,14 +41,14 @@
|
||||
}
|
||||
],
|
||||
"preLaunchTask": "build",
|
||||
"linux": {
|
||||
"linux": {
|
||||
"MIMode": "gdb",
|
||||
"externalConsole": false,
|
||||
},
|
||||
"osx": {
|
||||
},
|
||||
"osx": {
|
||||
"MIMode": "lldb",
|
||||
"externalConsole": true,
|
||||
},
|
||||
},
|
||||
}
|
||||
]
|
||||
}
|
||||
Vendored
+3
-3
@@ -5,7 +5,7 @@
|
||||
"label": "build",
|
||||
"type": "shell",
|
||||
"options": {
|
||||
"cwd": "${workspaceRoot}/build_dev"
|
||||
"cwd": "${workspaceRoot}/build"
|
||||
},
|
||||
"command": "make",
|
||||
"args": ["-j4"],
|
||||
@@ -15,7 +15,7 @@
|
||||
"label": "cmake",
|
||||
"type": "shell",
|
||||
"options": {
|
||||
"cwd": "${workspaceRoot}/build_dev"
|
||||
"cwd": "${workspaceRoot}/build"
|
||||
},
|
||||
"command": "cmake",
|
||||
"args": [
|
||||
@@ -37,7 +37,7 @@
|
||||
"command": "mkdir",
|
||||
"args": [
|
||||
"-p",
|
||||
"${workspaceRoot}/build_dev"
|
||||
"${workspaceRoot}/build"
|
||||
]
|
||||
},
|
||||
]
|
||||
|
||||
+11
-18
@@ -19,7 +19,7 @@ include(FindPkgConfig)
|
||||
include(cmake/CheckAtomic.cmake)
|
||||
|
||||
set (opendht_VERSION_MAJOR 2)
|
||||
set (opendht_VERSION_MINOR 4.1)
|
||||
set (opendht_VERSION_MINOR 4.0)
|
||||
set (opendht_VERSION ${opendht_VERSION_MAJOR}.${opendht_VERSION_MINOR})
|
||||
set (PACKAGE_VERSION ${opendht_VERSION})
|
||||
set (VERSION "${opendht_VERSION}")
|
||||
@@ -92,7 +92,7 @@ if (NOT MSVC)
|
||||
pkg_search_module(OPENSSL REQUIRED openssl)
|
||||
if (OPENSSL_FOUND)
|
||||
message(STATUS "Found OpenSSL ${OPENSSL_VERSION} ${OPENSSL_INCLUDE_DIRS}")
|
||||
include_direcopendht/ries(SYSTEM ${OPENSSL_INCLUDE_DIRS})
|
||||
include_directories(SYSTEM ${OPENSSL_INCLUDE_DIRS})
|
||||
link_directories (${OPENSSL_LIBRARY_DIRS})
|
||||
else ()
|
||||
message(SEND_ERROR "OpenSSL is required for DHT proxy as specified")
|
||||
@@ -403,23 +403,15 @@ if (OPENDHT_SHARED)
|
||||
endif ()
|
||||
|
||||
if (OPENDHT_C)
|
||||
add_library (opendht-c SHARED
|
||||
c/opendht.cpp
|
||||
c/opendht_c.h
|
||||
)
|
||||
target_compile_definitions(opendht-c PRIVATE OPENDHT_C_BUILD)
|
||||
if (OPENDHT_SHARED)
|
||||
add_library (opendht-c SHARED
|
||||
c/opendht.cpp
|
||||
c/opendht_c.h
|
||||
)
|
||||
target_compile_definitions(opendht-c PRIVATE OPENDHT_C_BUILD)
|
||||
target_link_libraries(opendht-c PRIVATE opendht)
|
||||
install (TARGETS opendht-c DESTINATION ${CMAKE_INSTALL_LIBDIR} EXPORT opendht-c)
|
||||
endif ()
|
||||
|
||||
if (OPENDHT_STATIC)
|
||||
add_library (opendht-c-static STATIC
|
||||
c/opendht.cpp
|
||||
c/opendht_c.h
|
||||
)
|
||||
target_link_libraries(opendht-c-static PRIVATE opendht-static)
|
||||
install (TARGETS opendht-c-static DESTINATION ${CMAKE_INSTALL_LIBDIR} EXPORT opendht-c-static)
|
||||
target_link_libraries(opendht-c opendht)
|
||||
else ()
|
||||
target_link_libraries(opendht-c opendht-static)
|
||||
endif ()
|
||||
|
||||
# PkgConfig module
|
||||
@@ -428,6 +420,7 @@ if (OPENDHT_C)
|
||||
opendht-c.pc
|
||||
@ONLY
|
||||
)
|
||||
install (TARGETS opendht-c DESTINATION ${CMAKE_INSTALL_LIBDIR} EXPORT opendht-c)
|
||||
install (FILES ${CMAKE_CURRENT_BINARY_DIR}/opendht-c.pc DESTINATION ${CMAKE_INSTALL_LIBDIR}/pkgconfig)
|
||||
install (FILES c/opendht_c.h DESTINATION ${CMAKE_INSTALL_PREFIX}/include/opendht)
|
||||
endif ()
|
||||
|
||||
@@ -4,10 +4,6 @@ SUBDIRS =
|
||||
|
||||
SUBDIRS += src
|
||||
|
||||
if ENABLE_C
|
||||
SUBDIRS += c
|
||||
endif
|
||||
|
||||
if ENABLE_TOOLS
|
||||
SUBDIRS += tools
|
||||
endif
|
||||
|
||||
@@ -1,12 +0,0 @@
|
||||
lib_LTLIBRARIES = libopendht-c.la
|
||||
noinst_HEADERS = opendht_c.h
|
||||
|
||||
AM_CPPFLAGS = -isystem @top_srcdir@/include @JsonCpp_CFLAGS@ @MsgPack_CFLAGS@
|
||||
|
||||
libopendht_c_la_LIBADD = $(LIBOBJS) ../src/.libs/libopendht.la
|
||||
|
||||
libopendht_c_la_SOURCES = \
|
||||
opendht.cpp
|
||||
|
||||
nobase_include_HEADERS = \
|
||||
opendht_c.h
|
||||
+2
-9
@@ -17,9 +17,7 @@
|
||||
*/
|
||||
|
||||
#include "opendht_c.h"
|
||||
|
||||
#include <opendht.h>
|
||||
#include <opendht/log.h>
|
||||
#include "opendht.h"
|
||||
|
||||
using ValueSp = std::shared_ptr<dht::Value>;
|
||||
using PrivkeySp = std::shared_ptr<dht::crypto::PrivateKey>;
|
||||
@@ -335,12 +333,7 @@ int dht_runner_run_config(dht_runner* r, in_port_t port, const dht_runner_config
|
||||
config.push_token = conf->push_token ? std::string(conf->push_token) : std::string{};
|
||||
config.peer_discovery = conf->peer_discovery;
|
||||
config.peer_publish = conf->peer_publish;
|
||||
|
||||
dht::DhtRunner::Context context;
|
||||
if (conf->log) {
|
||||
context.logger = dht::log::getStdLogger();
|
||||
}
|
||||
runner->run(port, config, std::move(context));
|
||||
runner->run(port, config);
|
||||
} catch(...) {
|
||||
return ENOTCONN;
|
||||
}
|
||||
|
||||
+1
-2
@@ -22,7 +22,7 @@
|
||||
extern "C" {
|
||||
#endif
|
||||
|
||||
#include <opendht/def.h>
|
||||
#include "def.h"
|
||||
#include <sys/socket.h>
|
||||
#include <netinet/in.h>
|
||||
#include <stdbool.h>
|
||||
@@ -147,7 +147,6 @@ struct OPENDHT_PUBLIC dht_runner_config {
|
||||
bool peer_publish;
|
||||
dht_certificate* server_ca;
|
||||
dht_identity client_identity;
|
||||
bool log;
|
||||
};
|
||||
typedef struct dht_runner_config dht_runner_config;
|
||||
OPENDHT_C_PUBLIC void dht_runner_config_default(dht_runner_config* config);
|
||||
|
||||
+1
-9
@@ -1,7 +1,7 @@
|
||||
dnl define macros
|
||||
m4_define([opendht_major_version], 2)
|
||||
m4_define([opendht_minor_version], 4)
|
||||
m4_define([opendht_patch_version], 1)
|
||||
m4_define([opendht_patch_version], 0)
|
||||
m4_define([opendht_version],
|
||||
[opendht_major_version.opendht_minor_version.opendht_patch_version])
|
||||
|
||||
@@ -32,13 +32,6 @@ AS_IF([test "x$enable_logs" != "xno"], [
|
||||
AC_DEFINE([OPENDHT_LOG], [false], [Define if DHT logs are enabled])
|
||||
])
|
||||
|
||||
dnl Check for C binding
|
||||
AC_ARG_ENABLE([c], [AS_HELP_STRING([--disable-c], [Disable DHT C binding])])
|
||||
AM_CONDITIONAL(ENABLE_C, test x$enable_c != "xno")
|
||||
AM_COND_IF(ENABLE_C, [
|
||||
AC_DEFINE([OPENDHT_C], [], [Define if DHT C biding is enabled])
|
||||
])
|
||||
|
||||
dnl Check for indexation
|
||||
AC_ARG_ENABLE([indexation], [AS_HELP_STRING([--disable-indexation], [Disable DHT indexation])])
|
||||
AM_CONDITIONAL(ENABLE_INDEXATION, test x$enable_indexation != "xno")
|
||||
@@ -205,7 +198,6 @@ AC_SUBST(PROJECT_BINARY_DIR, "../src/.libs")
|
||||
|
||||
AC_CONFIG_FILES([Makefile
|
||||
src/Makefile
|
||||
c/Makefile
|
||||
tools/Makefile
|
||||
python/Makefile
|
||||
python/setup.py
|
||||
|
||||
@@ -9,7 +9,7 @@ RUN apt-get update && apt-get install -y \
|
||||
|
||||
RUN apt-get update && apt-get install -y \
|
||||
build-essential pkg-config cmake git wget \
|
||||
libtool autotools-dev autoconf \
|
||||
autotools-dev autoconf \
|
||||
cython3 python3-dev python3-setuptools \
|
||||
libncurses5-dev libreadline-dev nettle-dev libcppunit-dev \
|
||||
libgnutls28-dev libuv1-dev libjsoncpp-dev libargon2-dev \
|
||||
|
||||
@@ -8,8 +8,7 @@ RUN apt-get update && apt-get install -y \
|
||||
&& echo 'debconf debconf/frontend select Noninteractive' | debconf-set-selections
|
||||
|
||||
RUN apt-get update \
|
||||
&& apt-get install -y llvm llvm-dev lldb clang gdb make cmake pkg-config \
|
||||
libtool git wget libncurses5-dev libreadline-dev \
|
||||
&& apt-get install -y llvm llvm-dev lldb clang gdb make cmake pkg-config git wget libncurses5-dev libreadline-dev \
|
||||
nettle-dev libgnutls28-dev libuv1-dev libmsgpack-dev libjsoncpp-dev cython3 python3-dev \
|
||||
python3-setuptools libcppunit-dev python3-pip \
|
||||
autotools-dev autoconf libssl-dev libargon2-dev \
|
||||
|
||||
@@ -447,13 +447,13 @@ private:
|
||||
// Storage
|
||||
void storageAddListener(const InfoHash& id, const Sp<Node>& node, size_t tid, Query&& = {}, int version = 0);
|
||||
bool storageStore(const InfoHash& id, const Sp<Value>& value, time_point created, const SockAddr& sa = {}, bool permanent = false);
|
||||
bool storageErase(const InfoHash& id, Value::Id vid);
|
||||
bool storageRefresh(const InfoHash& id, Value::Id vid);
|
||||
void expireStore();
|
||||
void expireStorage(InfoHash h);
|
||||
void expireStore(decltype(store)::iterator);
|
||||
|
||||
void storageRemoved(const InfoHash& id, Storage& st, const std::vector<Sp<Value>>& values, size_t totalSize);
|
||||
void storageChanged(const InfoHash& id, Storage& st, const Sp<Value>&, bool newValue);
|
||||
void storageChanged(const InfoHash& id, Storage& st, ValueStorage&, bool newValue);
|
||||
std::string printStorageLog(const decltype(store)::value_type&) const;
|
||||
|
||||
/**
|
||||
|
||||
@@ -273,13 +273,13 @@ public:
|
||||
* Usefull to restart a node and get things running fast without putting load on the network.
|
||||
*/
|
||||
void bootstrap(std::vector<SockAddr> nodes, DoneCallbackSimple cb={});
|
||||
void bootstrap(SockAddr addr, DoneCallbackSimple cb={});
|
||||
void bootstrap(const SockAddr& addr, DoneCallbackSimple cb={});
|
||||
|
||||
/**
|
||||
* Insert known nodes to the routing table, without necessarly ping them.
|
||||
* Usefull to restart a node and get things running fast without putting load on the network.
|
||||
*/
|
||||
void bootstrap(std::vector<NodeExport> nodes);
|
||||
void bootstrap(const std::vector<NodeExport>& nodes);
|
||||
|
||||
/**
|
||||
* Add host:service to bootstrap nodes, and ping this node.
|
||||
|
||||
@@ -252,12 +252,12 @@ public:
|
||||
* @param values The values to send.
|
||||
* @param version If version = 1, a request will be used to answer to the listener
|
||||
*/
|
||||
void tellListener(const Sp<Node>& n, Tid socket_id, const InfoHash& hash, want_t want, const Blob& ntoken,
|
||||
void tellListener(Sp<Node> n, Tid socket_id, const InfoHash& hash, want_t want, const Blob& ntoken,
|
||||
std::vector<Sp<Node>>&& nodes, std::vector<Sp<Node>>&& nodes6,
|
||||
std::vector<Sp<Value>>&& values, const Query& q, int version);
|
||||
|
||||
void tellListenerRefreshed(const Sp<Node>& n, Tid socket_id, const InfoHash& hash, const Blob& ntoken, const std::vector<Value::Id>& values, int version);
|
||||
void tellListenerExpired(const Sp<Node>& n, Tid socket_id, const InfoHash& hash, const Blob& ntoken, const std::vector<Value::Id>& values, int version);
|
||||
void tellListenerRefreshed(Sp<Node> n, Tid socket_id, const InfoHash& hash, const Blob& ntoken, const std::vector<Value::Id>& values, int version);
|
||||
void tellListenerExpired(Sp<Node> n, Tid socket_id, const InfoHash& hash, const Blob& ntoken, const std::vector<Value::Id>& values, int version);
|
||||
|
||||
bool isRunning(sa_family_t af) const;
|
||||
inline want_t want () const { return dht_socket->hasIPv4() and dht_socket->hasIPv6() ? (WANT4 | WANT6) : -1; }
|
||||
@@ -278,7 +278,7 @@ public:
|
||||
* @return the request with information concerning its success.
|
||||
*/
|
||||
Sp<Request>
|
||||
sendPing(const Sp<Node>& n, RequestCb&& on_done, RequestExpiredCb&& on_expired);
|
||||
sendPing(Sp<Node> n, RequestCb&& on_done, RequestExpiredCb&& on_expired);
|
||||
|
||||
/**
|
||||
* Send a "ping" request to a given node.
|
||||
@@ -309,7 +309,7 @@ public:
|
||||
*
|
||||
* @return the request with information concerning its success.
|
||||
*/
|
||||
Sp<Request> sendFindNode(const Sp<Node>& n,
|
||||
Sp<Request> sendFindNode(Sp<Node> n,
|
||||
const InfoHash& hash,
|
||||
want_t want = -1,
|
||||
RequestCb&& on_done = {},
|
||||
@@ -328,7 +328,7 @@ public:
|
||||
*
|
||||
* @return the request with information concerning its success.
|
||||
*/
|
||||
Sp<Request> sendGetValues(const Sp<Node>& n,
|
||||
Sp<Request> sendGetValues(Sp<Node> n,
|
||||
const InfoHash& hash,
|
||||
const Query& query,
|
||||
want_t want,
|
||||
@@ -357,7 +357,7 @@ public:
|
||||
*
|
||||
* @return the request with information concerning its success.
|
||||
*/
|
||||
Sp<Request> sendListen(const Sp<Node>& n,
|
||||
Sp<Request> sendListen(Sp<Node> n,
|
||||
const InfoHash& hash,
|
||||
const Query& query,
|
||||
const Blob& token,
|
||||
@@ -377,7 +377,7 @@ public:
|
||||
*
|
||||
* @return the request with information concerning its success.
|
||||
*/
|
||||
Sp<Request> sendAnnounceValue(const Sp<Node>& n,
|
||||
Sp<Request> sendAnnounceValue(Sp<Node> n,
|
||||
const InfoHash& hash,
|
||||
const Sp<Value>& v,
|
||||
time_point created,
|
||||
@@ -397,7 +397,7 @@ public:
|
||||
*
|
||||
* @return the request with information concerning its success.
|
||||
*/
|
||||
Sp<Request> sendRefreshValue(const Sp<Node>& n,
|
||||
Sp<Request> sendRefreshValue(Sp<Node> n,
|
||||
const InfoHash& hash,
|
||||
const Value::Id& vid,
|
||||
const Blob& token,
|
||||
@@ -416,7 +416,7 @@ public:
|
||||
*
|
||||
* @return the request with information concerning its success.
|
||||
*/
|
||||
Sp<Request> sendUpdateValues(const Sp<Node>& n,
|
||||
Sp<Request> sendUpdateValues(Sp<Node> n,
|
||||
const InfoHash& infohash,
|
||||
const std::vector<Sp<Value>>& values,
|
||||
time_point created,
|
||||
|
||||
@@ -42,7 +42,7 @@ namespace dht {
|
||||
namespace net {
|
||||
|
||||
static const constexpr in_port_t DHT_DEFAULT_PORT = 4222;
|
||||
static const constexpr size_t RX_QUEUE_MAX_SIZE = 1024 * 64;
|
||||
static const constexpr size_t RX_QUEUE_MAX_SIZE = 1024 * 16;
|
||||
static const constexpr std::chrono::milliseconds RX_QUEUE_MAX_DELAY(650);
|
||||
|
||||
int bindSocket(const SockAddr& addr, SockAddr& bound);
|
||||
|
||||
+65
-69
@@ -1194,20 +1194,15 @@ Dht::cancelPut(const InfoHash& id, const Value::Id& vid)
|
||||
};
|
||||
canceled |= sr_cancel_put(dht4.searches);
|
||||
canceled |= sr_cancel_put(dht6.searches);
|
||||
if (canceled) {
|
||||
auto st = store.find(id);
|
||||
if (st != store.end()) {
|
||||
if (auto value = st->second.remove(id, vid))
|
||||
storageRemoved(id, st->second, {value}, value->size());
|
||||
}
|
||||
}
|
||||
if (canceled)
|
||||
storageErase(id, vid);
|
||||
return canceled;
|
||||
}
|
||||
|
||||
// Storage
|
||||
|
||||
void
|
||||
Dht::storageChanged(const InfoHash& id, Storage& st, const Sp<Value>& v, bool newValue)
|
||||
Dht::storageChanged(const InfoHash& id, Storage& st, ValueStorage& v, bool newValue)
|
||||
{
|
||||
if (newValue) {
|
||||
if (not st.local_listeners.empty()) {
|
||||
@@ -1217,8 +1212,8 @@ Dht::storageChanged(const InfoHash& id, Storage& st, const Sp<Value>& v, bool ne
|
||||
cbs.reserve(st.local_listeners.size());
|
||||
for (const auto& l : st.local_listeners) {
|
||||
std::vector<Sp<Value>> vals;
|
||||
if (not l.second.filter or l.second.filter(*v))
|
||||
vals.push_back(v);
|
||||
if (not l.second.filter or l.second.filter(*v.data))
|
||||
vals.push_back(v.data);
|
||||
if (not vals.empty()) {
|
||||
if (logger_)
|
||||
logger_->d(id, "[store %s] sending update local listener with token %lu",
|
||||
@@ -1239,14 +1234,14 @@ Dht::storageChanged(const InfoHash& id, Storage& st, const Sp<Value>& v, bool ne
|
||||
for (const auto& node_listeners : st.listeners) {
|
||||
for (const auto& l : node_listeners.second) {
|
||||
auto f = l.second.query.where.getFilter();
|
||||
if (f and not f(*v))
|
||||
if (f and not f(*v.data))
|
||||
continue;
|
||||
if (logger_)
|
||||
logger_->w(id, node_listeners.first->id, "[store %s] [node %s] sending update",
|
||||
id.toString().c_str(),
|
||||
node_listeners.first->toString().c_str());
|
||||
std::vector<Sp<Value>> vals {};
|
||||
vals.push_back(v);
|
||||
vals.push_back(v.data);
|
||||
Blob ntoken = makeToken(node_listeners.first->getAddr(), false);
|
||||
network_engine.tellListener(node_listeners.first, l.first, id, 0, ntoken, {}, {},
|
||||
std::move(vals), l.second.query, l.second.version);
|
||||
@@ -1287,18 +1282,26 @@ Dht::storageStore(const InfoHash& id, const Sp<Value>& value, time_point created
|
||||
vs->expiration_job = scheduler.add(expiration, std::bind(&Dht::expireStorage, this, id));
|
||||
}
|
||||
if (total_store_size > max_store_size) {
|
||||
auto value = vs->data;
|
||||
auto value_diff = store.second.values_diff;
|
||||
expireStore();
|
||||
storageChanged(id, st->second, value, value_diff > 0);
|
||||
} else {
|
||||
storageChanged(id, st->second, vs->data, store.second.values_diff > 0);
|
||||
}
|
||||
storageChanged(id, st->second, *vs, store.second.values_diff > 0);
|
||||
}
|
||||
|
||||
return std::get<0>(store);
|
||||
}
|
||||
|
||||
bool
|
||||
Dht::storageErase(const InfoHash& id, Value::Id vid)
|
||||
{
|
||||
auto st = store.find(id);
|
||||
if (st == store.end())
|
||||
return false;
|
||||
auto ret = st->second.remove(id, vid);
|
||||
total_store_size += ret.size_diff;
|
||||
total_values += ret.values_diff;
|
||||
return ret.values_diff;
|
||||
}
|
||||
|
||||
void
|
||||
Dht::storageAddListener(const InfoHash& id, const Sp<Node>& node, size_t socket_id, Query&& query, int version)
|
||||
{
|
||||
@@ -1330,8 +1333,36 @@ Dht::expireStore(decltype(store)::iterator i)
|
||||
const auto& id = i->first;
|
||||
auto& st = i->second;
|
||||
auto stats = st.expire(id, scheduler.time());
|
||||
total_store_size += stats.first;
|
||||
total_values -= stats.second.size();
|
||||
if (not stats.second.empty()) {
|
||||
storageRemoved(id, st, stats.second, -stats.first);
|
||||
if (logger_)
|
||||
logger_->d(id, "[store %s] discarded %ld expired values (%ld bytes)",
|
||||
id.toString().c_str(), stats.second.size(), -stats.first);
|
||||
|
||||
if (not st.listeners.empty()) {
|
||||
if (logger_)
|
||||
logger_->d(id, "[store %s] %lu remote listeners", id.toString().c_str(), st.listeners.size());
|
||||
|
||||
std::vector<Value::Id> ids;
|
||||
ids.reserve(stats.second.size());
|
||||
for (const auto& v : stats.second)
|
||||
ids.emplace_back(v->id);
|
||||
|
||||
for (const auto& node_listeners : st.listeners) {
|
||||
for (const auto& l : node_listeners.second) {
|
||||
if (logger_)
|
||||
logger_->w(id, node_listeners.first->id, "[store %s] [node %s] sending expired",
|
||||
id.toString().c_str(),
|
||||
node_listeners.first->toString().c_str());
|
||||
Blob ntoken = makeToken(node_listeners.first->getAddr(), false);
|
||||
network_engine.tellListenerExpired(node_listeners.first, l.first, id, ntoken, ids, l.second.version);
|
||||
}
|
||||
}
|
||||
}
|
||||
for (const auto& local_listeners : st.local_listeners) {
|
||||
local_listeners.second.get_cb(stats.second, true);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1343,41 +1374,6 @@ Dht::expireStorage(InfoHash h)
|
||||
expireStore(i);
|
||||
}
|
||||
|
||||
void
|
||||
Dht::storageRemoved(const InfoHash& id, Storage& st, const std::vector<Sp<Value>>& values, size_t totalSize)
|
||||
{
|
||||
if (logger_)
|
||||
logger_->d(id, "[store %s] discarded %ld values (%ld bytes)",
|
||||
id.toString().c_str(), values.size(), totalSize);
|
||||
|
||||
total_store_size -= totalSize;
|
||||
total_values -= values.size();
|
||||
|
||||
if (not st.listeners.empty()) {
|
||||
if (logger_)
|
||||
logger_->d(id, "[store %s] %lu remote listeners", id.toString().c_str(), st.listeners.size());
|
||||
|
||||
std::vector<Value::Id> ids;
|
||||
ids.reserve(values.size());
|
||||
for (const auto& v : values)
|
||||
ids.emplace_back(v->id);
|
||||
|
||||
for (const auto& node_listeners : st.listeners) {
|
||||
for (const auto& l : node_listeners.second) {
|
||||
if (logger_)
|
||||
logger_->w(id, node_listeners.first->id, "[store %s] [node %s] sending expired",
|
||||
id.toString().c_str(),
|
||||
node_listeners.first->toString().c_str());
|
||||
Blob ntoken = makeToken(node_listeners.first->getAddr(), false);
|
||||
network_engine.tellListenerExpired(node_listeners.first, l.first, id, ntoken, ids, l.second.version);
|
||||
}
|
||||
}
|
||||
}
|
||||
for (const auto& local_listeners : st.local_listeners) {
|
||||
local_listeners.second.get_cb(values, true);
|
||||
}
|
||||
}
|
||||
|
||||
void
|
||||
Dht::expireStore()
|
||||
{
|
||||
@@ -1403,23 +1399,23 @@ Dht::expireStore()
|
||||
break;
|
||||
}
|
||||
auto largest = store_quota.begin();
|
||||
for (auto it = std::next(largest); it != store_quota.end(); ++it) {
|
||||
for (auto it = ++largest; it != store_quota.end(); ++it) {
|
||||
if (it->second.size() > largest->second.size())
|
||||
largest = it;
|
||||
}
|
||||
if (largest != store_quota.end()) {
|
||||
while (true) {
|
||||
auto exp_value = largest->second.getOldest();
|
||||
auto storage = store.find(exp_value.first);
|
||||
if (storage != store.end()) {
|
||||
if (logger_)
|
||||
logger_->w("Storage quota full: discarding value from %s at %s %016" PRIx64, largest->first.toString().c_str(), exp_value.first.to_c_str(), exp_value.second);
|
||||
|
||||
if (auto value = storage->second.remove(exp_value.first, exp_value.second)) {
|
||||
storageRemoved(storage->first, storage->second, {value}, value->size());
|
||||
break;
|
||||
}
|
||||
}
|
||||
if (logger_)
|
||||
logger_->w("No space left: discarding value of largest consumer %s", largest->first.toString().c_str());
|
||||
while (true) {
|
||||
auto exp_value = largest->second.getOldest();
|
||||
auto storage = store.find(exp_value.first);
|
||||
if (storage != store.end()) {
|
||||
auto ret = storage->second.remove(exp_value.first, exp_value.second);
|
||||
total_store_size += ret.size_diff;
|
||||
total_values += ret.values_diff;
|
||||
if (logger_)
|
||||
logger_->w("Discarded %ld bytes, still %ld used", largest->first.toString().c_str(), total_store_size);
|
||||
if (ret.values_diff)
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1668,7 +1664,7 @@ Dht::dumpTables() const
|
||||
std::string
|
||||
Dht::getStorageLog() const
|
||||
{
|
||||
std::ostringstream out;
|
||||
std::stringstream out;
|
||||
for (const auto& s : store)
|
||||
out << printStorageLog(s);
|
||||
out << std::endl << std::endl;
|
||||
@@ -2562,7 +2558,7 @@ Dht::storageRefresh(const InfoHash& id, Value::Id vid)
|
||||
}
|
||||
}
|
||||
|
||||
auto expiration = s->second.refresh(id, now, vid, types);
|
||||
auto expiration = s->second.refresh(now, vid, types);
|
||||
if (expiration.first) {
|
||||
scheduler.cancel(expiration.first->expiration_job);
|
||||
if (expiration.second != time_point::max()) {
|
||||
|
||||
+26
-54
@@ -163,15 +163,14 @@ DhtRunner::run(const Config& config, Context&& context)
|
||||
net::PacketList ret;
|
||||
{
|
||||
std::lock_guard<std::mutex> lck(sock_mtx);
|
||||
rcv.splice(rcv.end(), std::move(pkts));
|
||||
size_t dropped = 0;
|
||||
while (rcv.size() > net::RX_QUEUE_MAX_SIZE) {
|
||||
auto maxSize = net::RX_QUEUE_MAX_SIZE - pkts.size();
|
||||
while (rcv.size() > maxSize) {
|
||||
if (logger_)
|
||||
logger_->e("Dropping packet: queue is full!");
|
||||
rcv.pop_front();
|
||||
dropped++;
|
||||
}
|
||||
if (dropped and logger_) {
|
||||
logger_->w("[runner %p] dropped %zu packets: queue is full!", this, dropped);
|
||||
}
|
||||
|
||||
rcv.splice(rcv.end(), std::move(pkts));
|
||||
ret = std::move(rcv_free);
|
||||
}
|
||||
cv.notify_all();
|
||||
@@ -298,24 +297,16 @@ DhtRunner::shutdown(ShutdownCallback cb, bool stop) {
|
||||
}
|
||||
if (logger_)
|
||||
logger_->d("[runner %p] state changed to Stopping, %zu ongoing ops", this, ongoing_ops.load());
|
||||
#ifdef OPENDHT_PROXY_CLIENT
|
||||
ongoing_ops += 2;
|
||||
#else
|
||||
ongoing_ops++;
|
||||
#endif
|
||||
shutdownCallbacks_.emplace_back(std::move(cb));
|
||||
pending_ops.emplace([=](SecureDht&) mutable {
|
||||
auto onShutdown = [this]{ opEnded(); };
|
||||
#ifdef OPENDHT_PROXY_CLIENT
|
||||
if (dht_via_proxy_)
|
||||
dht_via_proxy_->shutdown(onShutdown, stop);
|
||||
else
|
||||
opEnded();
|
||||
#endif
|
||||
if (dht_)
|
||||
dht_->shutdown(onShutdown, stop);
|
||||
else
|
||||
opEnded();
|
||||
});
|
||||
cv.notify_all();
|
||||
}
|
||||
@@ -826,28 +817,19 @@ void
|
||||
DhtRunner::cancelListen(InfoHash h, size_t token)
|
||||
{
|
||||
std::lock_guard<std::mutex> lck(storage_mtx);
|
||||
if (running != State::Running)
|
||||
return;
|
||||
ongoing_ops++;
|
||||
#ifdef OPENDHT_PROXY_CLIENT
|
||||
pending_ops.emplace([=](SecureDht&) {
|
||||
auto it = listeners_.find(token);
|
||||
if (it != listeners_.end()) {
|
||||
if (it->second.tokenClassicDht)
|
||||
dht_->cancelListen(h, it->second.tokenClassicDht);
|
||||
if (it->second.tokenProxyDht and dht_via_proxy_)
|
||||
dht_via_proxy_->cancelListen(h, it->second.tokenProxyDht);
|
||||
listeners_.erase(it);
|
||||
} else {
|
||||
if (logger_)
|
||||
logger_->w("[runner %p] cancelListen: unknown token %zu.", this, token);
|
||||
}
|
||||
opEnded();
|
||||
if (it == listeners_.end()) return;
|
||||
if (it->second.tokenClassicDht)
|
||||
dht_->cancelListen(h, it->second.tokenClassicDht);
|
||||
if (it->second.tokenProxyDht and dht_via_proxy_)
|
||||
dht_via_proxy_->cancelListen(h, it->second.tokenProxyDht);
|
||||
listeners_.erase(it);
|
||||
});
|
||||
#else
|
||||
pending_ops.emplace([=](SecureDht& dht) {
|
||||
dht.cancelListen(h, token);
|
||||
opEnded();
|
||||
});
|
||||
#endif // OPENDHT_PROXY_CLIENT
|
||||
cv.notify_all();
|
||||
@@ -857,29 +839,19 @@ void
|
||||
DhtRunner::cancelListen(InfoHash h, std::shared_future<size_t> ftoken)
|
||||
{
|
||||
std::lock_guard<std::mutex> lck(storage_mtx);
|
||||
if (running != State::Running)
|
||||
return;
|
||||
ongoing_ops++;
|
||||
#ifdef OPENDHT_PROXY_CLIENT
|
||||
pending_ops.emplace([this, h, ftoken = std::move(ftoken)](SecureDht&) {
|
||||
auto token = ftoken.get();
|
||||
auto it = listeners_.find(token);
|
||||
if (it != listeners_.end()) {
|
||||
if (it->second.tokenClassicDht)
|
||||
dht_->cancelListen(h, it->second.tokenClassicDht);
|
||||
if (it->second.tokenProxyDht and dht_via_proxy_)
|
||||
dht_via_proxy_->cancelListen(h, it->second.tokenProxyDht);
|
||||
listeners_.erase(it);
|
||||
} else {
|
||||
if (logger_)
|
||||
logger_->w("[runner %p] cancelListen: unknown token %zu.", this, token);
|
||||
}
|
||||
opEnded();
|
||||
pending_ops.emplace([=](SecureDht&) {
|
||||
auto it = listeners_.find(ftoken.get());
|
||||
if (it == listeners_.end()) return;
|
||||
if (it->second.tokenClassicDht)
|
||||
dht_->cancelListen(h, it->second.tokenClassicDht);
|
||||
if (it->second.tokenProxyDht and dht_via_proxy_)
|
||||
dht_via_proxy_->cancelListen(h, it->second.tokenProxyDht);
|
||||
listeners_.erase(it);
|
||||
});
|
||||
#else
|
||||
pending_ops.emplace([this, h, ftoken = std::move(ftoken)](SecureDht& dht) {
|
||||
pending_ops.emplace([=](SecureDht& dht) {
|
||||
dht.cancelListen(h, ftoken.get());
|
||||
opEnded();
|
||||
});
|
||||
#endif // OPENDHT_PROXY_CLIENT
|
||||
cv.notify_all();
|
||||
@@ -914,7 +886,7 @@ DhtRunner::put(InfoHash hash, std::shared_ptr<Value> value, DoneCallback cb, tim
|
||||
return;
|
||||
}
|
||||
ongoing_ops++;
|
||||
pending_ops.emplace([=, value = std::move(value), cb = std::move(cb)](SecureDht& dht) mutable {
|
||||
pending_ops.emplace([=, cb = std::move(cb)](SecureDht& dht) mutable {
|
||||
dht.put(hash, value, bindOpDoneCallback(std::move(cb)), created, permanent);
|
||||
});
|
||||
cv.notify_all();
|
||||
@@ -1094,7 +1066,7 @@ DhtRunner::bootstrap(std::vector<SockAddr> nodes, DoneCallbackSimple cb)
|
||||
}
|
||||
|
||||
void
|
||||
DhtRunner::bootstrap(SockAddr addr, DoneCallbackSimple cb)
|
||||
DhtRunner::bootstrap(const SockAddr& addr, DoneCallbackSimple cb)
|
||||
{
|
||||
std::unique_lock<std::mutex> lck(storage_mtx);
|
||||
if (running != State::Running) {
|
||||
@@ -1103,7 +1075,7 @@ DhtRunner::bootstrap(SockAddr addr, DoneCallbackSimple cb)
|
||||
return;
|
||||
}
|
||||
ongoing_ops++;
|
||||
pending_ops_prio.emplace([addr = std::move(addr), cb = bindOpDoneCallback(std::move(cb))](SecureDht& dht) mutable {
|
||||
pending_ops_prio.emplace([addr, cb = bindOpDoneCallback(std::move(cb))](SecureDht& dht) mutable {
|
||||
dht.pingNode(std::move(addr), std::move(cb));
|
||||
});
|
||||
cv.notify_all();
|
||||
@@ -1122,12 +1094,12 @@ DhtRunner::bootstrap(const InfoHash& id, const SockAddr& address)
|
||||
}
|
||||
|
||||
void
|
||||
DhtRunner::bootstrap(std::vector<NodeExport> nodes)
|
||||
DhtRunner::bootstrap(const std::vector<NodeExport>& nodes)
|
||||
{
|
||||
std::lock_guard<std::mutex> lck(storage_mtx);
|
||||
if (running != State::Running)
|
||||
return;
|
||||
pending_ops_prio.emplace([nodes = std::move(nodes)](SecureDht& dht) {
|
||||
pending_ops_prio.emplace([=](SecureDht& dht) {
|
||||
for (auto& node : nodes)
|
||||
dht.insertNode(node);
|
||||
});
|
||||
|
||||
+10
-10
@@ -116,7 +116,7 @@ NetworkEngine::~NetworkEngine() {
|
||||
}
|
||||
|
||||
void
|
||||
NetworkEngine::tellListener(const Sp<Node>& node, Tid socket_id, const InfoHash& hash, want_t want,
|
||||
NetworkEngine::tellListener(Sp<Node> node, Tid socket_id, const InfoHash& hash, want_t want,
|
||||
const Blob& ntoken, std::vector<Sp<Node>>&& nodes,
|
||||
std::vector<Sp<Node>>&& nodes6, std::vector<Sp<Value>>&& values,
|
||||
const Query& query, int version)
|
||||
@@ -135,7 +135,7 @@ NetworkEngine::tellListener(const Sp<Node>& node, Tid socket_id, const InfoHash&
|
||||
}
|
||||
|
||||
void
|
||||
NetworkEngine::tellListenerRefreshed(const Sp<Node>& n, Tid socket_id, const InfoHash&, const Blob& token, const std::vector<Value::Id>& values, int version)
|
||||
NetworkEngine::tellListenerRefreshed(Sp<Node> n, Tid socket_id, const InfoHash&, const Blob& token, const std::vector<Value::Id>& values, int version)
|
||||
{
|
||||
msgpack::sbuffer buffer;
|
||||
msgpack::packer<msgpack::sbuffer> pk(&buffer);
|
||||
@@ -186,7 +186,7 @@ NetworkEngine::tellListenerRefreshed(const Sp<Node>& n, Tid socket_id, const Inf
|
||||
}
|
||||
|
||||
void
|
||||
NetworkEngine::tellListenerExpired(const Sp<Node>& n, Tid socket_id, const InfoHash&, const Blob& token, const std::vector<Value::Id>& values, int version)
|
||||
NetworkEngine::tellListenerExpired(Sp<Node> n, Tid socket_id, const InfoHash&, const Blob& token, const std::vector<Value::Id>& values, int version)
|
||||
{
|
||||
msgpack::sbuffer buffer;
|
||||
msgpack::packer<msgpack::sbuffer> pk(&buffer);
|
||||
@@ -699,7 +699,7 @@ NetworkEngine::send(const SockAddr& addr, const char *buf, size_t len, bool conf
|
||||
}
|
||||
|
||||
Sp<Request>
|
||||
NetworkEngine::sendPing(const Sp<Node>& node, RequestCb&& on_done, RequestExpiredCb&& on_expired) {
|
||||
NetworkEngine::sendPing(Sp<Node> node, RequestCb&& on_done, RequestExpiredCb&& on_expired) {
|
||||
Tid tid (node->getNewTid());
|
||||
msgpack::sbuffer buffer;
|
||||
msgpack::packer<msgpack::sbuffer> pk(&buffer);
|
||||
@@ -757,7 +757,7 @@ NetworkEngine::sendPong(const SockAddr& addr, Tid tid) {
|
||||
}
|
||||
|
||||
Sp<Request>
|
||||
NetworkEngine::sendFindNode(const Sp<Node>& n, const InfoHash& target, want_t want,
|
||||
NetworkEngine::sendFindNode(Sp<Node> n, const InfoHash& target, want_t want,
|
||||
RequestCb&& on_done, RequestExpiredCb&& on_expired) {
|
||||
Tid tid (n->getNewTid());
|
||||
msgpack::sbuffer buffer;
|
||||
@@ -802,7 +802,7 @@ NetworkEngine::sendFindNode(const Sp<Node>& n, const InfoHash& target, want_t wa
|
||||
|
||||
|
||||
Sp<Request>
|
||||
NetworkEngine::sendGetValues(const Sp<Node>& n, const InfoHash& info_hash, const Query& query, want_t want,
|
||||
NetworkEngine::sendGetValues(Sp<Node> n, const InfoHash& info_hash, const Query& query, want_t want,
|
||||
RequestCb&& on_done, RequestExpiredCb&& on_expired) {
|
||||
Tid tid (n->getNewTid());
|
||||
msgpack::sbuffer buffer;
|
||||
@@ -1073,7 +1073,7 @@ NetworkEngine::bufferNodes(sa_family_t af, const InfoHash& id, want_t want,
|
||||
}
|
||||
|
||||
Sp<Request>
|
||||
NetworkEngine::sendListen(const Sp<Node>& n,
|
||||
NetworkEngine::sendListen(Sp<Node> n,
|
||||
const InfoHash& hash,
|
||||
const Query& query,
|
||||
const Blob& token,
|
||||
@@ -1142,7 +1142,7 @@ NetworkEngine::sendListenConfirmation(const SockAddr& addr, Tid tid) {
|
||||
}
|
||||
|
||||
Sp<Request>
|
||||
NetworkEngine::sendAnnounceValue(const Sp<Node>& n,
|
||||
NetworkEngine::sendAnnounceValue(Sp<Node> n,
|
||||
const InfoHash& infohash,
|
||||
const Sp<Value>& value,
|
||||
time_point created,
|
||||
@@ -1200,7 +1200,7 @@ NetworkEngine::sendAnnounceValue(const Sp<Node>& n,
|
||||
}
|
||||
|
||||
Sp<Request>
|
||||
NetworkEngine::sendUpdateValues(const Sp<Node>& n,
|
||||
NetworkEngine::sendUpdateValues(Sp<Node> n,
|
||||
const InfoHash& infohash,
|
||||
const std::vector<Sp<Value>>& values,
|
||||
time_point created,
|
||||
@@ -1246,7 +1246,7 @@ NetworkEngine::sendUpdateValues(const Sp<Node>& n,
|
||||
}
|
||||
|
||||
Sp<Request>
|
||||
NetworkEngine::sendRefreshValue(const Sp<Node>& n,
|
||||
NetworkEngine::sendRefreshValue(Sp<Node> n,
|
||||
const InfoHash& infohash,
|
||||
const Value::Id& vid,
|
||||
const Blob& token,
|
||||
|
||||
+1
-1
@@ -28,7 +28,7 @@ struct OpCacheValueStorage
|
||||
Sp<Value> data {};
|
||||
unsigned refCount {1};
|
||||
system_clock::time_point updated {system_clock::time_point::min()};
|
||||
explicit OpCacheValueStorage(Sp<Value> val) : data(std::move(val)) {}
|
||||
explicit OpCacheValueStorage(Sp<Value> val) : data(val) {}
|
||||
};
|
||||
|
||||
class OpValueCache {
|
||||
|
||||
+2
-2
@@ -74,14 +74,14 @@ struct Request {
|
||||
Blob&& msg,
|
||||
std::function<void(const Request&, ParsedMessage&&)> on_done,
|
||||
std::function<void(const Request&, bool)> on_expired) :
|
||||
node(std::move(node)), tid(tid), type(type), on_done(std::move(on_done)), on_expired(std::move(on_expired)), msg(std::move(msg)) { }
|
||||
node(node), tid(tid), type(type), on_done(on_done), on_expired(on_expired), msg(std::move(msg)) { }
|
||||
Request(MessageType type, Tid tid,
|
||||
Sp<Node> node,
|
||||
Blob&& msg,
|
||||
std::function<void(const Request&, ParsedMessage&&)> on_done,
|
||||
std::function<bool(const Request&, DhtProtocolException&&)> on_error,
|
||||
std::function<void(const Request&, bool)> on_expired) :
|
||||
node(std::move(node)), tid(tid), type(type), on_done(std::move(on_done)), on_error(std::move(on_error)), on_expired(std::move(on_expired)), msg(std::move(msg)) { }
|
||||
node(node), tid(tid), type(type), on_done(on_done), on_error(on_error), on_expired(on_expired), msg(std::move(msg)) { }
|
||||
|
||||
Tid getTid() const { return tid; }
|
||||
MessageType getType() const { return type; }
|
||||
|
||||
+9
-27
@@ -37,32 +37,18 @@ public:
|
||||
storedValues_.emplace(expiration, std::pair<InfoHash, Value::Id>(id, value.id));
|
||||
}
|
||||
void erase(const InfoHash& id, const Value& value, time_point expiration) {
|
||||
totalSize_ -= value.size();
|
||||
auto range = storedValues_.equal_range(expiration);
|
||||
for (auto rit = range.first; rit != range.second;) {
|
||||
if (rit->second.first == id && rit->second.second == value.id) {
|
||||
totalSize_ -= value.size();
|
||||
storedValues_.erase(rit);
|
||||
return;
|
||||
break;
|
||||
} else
|
||||
++rit;
|
||||
}
|
||||
// printf("StorageBucket::erase can't find value %s %016" PRIx64 "\n", id.to_c_str(), value.id);
|
||||
}
|
||||
void refresh(const InfoHash& id, const Value& value, time_point old_expiration, time_point expiration) {
|
||||
auto range = storedValues_.equal_range(old_expiration);
|
||||
for (auto rit = range.first; rit != range.second;) {
|
||||
if (rit->second.first == id && rit->second.second == value.id) {
|
||||
storedValues_.erase(rit);
|
||||
storedValues_.emplace(expiration, std::pair<InfoHash, Value::Id>(id, value.id));
|
||||
return;
|
||||
} else
|
||||
++rit;
|
||||
}
|
||||
// printf("StorageBucket::refresh can't find value %s %016" PRIx64 "\n", id.to_c_str(), value.id);
|
||||
insert(id, value, expiration);
|
||||
}
|
||||
size_t size() const { return totalSize_; }
|
||||
std::pair<InfoHash, Value::Id> getOldest() const { return storedValues_.empty() ? std::pair<InfoHash, Value::Id>{} : storedValues_.begin()->second; }
|
||||
std::pair<InfoHash, Value::Id> getOldest() const { return storedValues_.begin()->second; }
|
||||
private:
|
||||
std::multimap<time_point, std::pair<InfoHash, Value::Id>> storedValues_;
|
||||
size_t totalSize_ {0};
|
||||
@@ -88,7 +74,7 @@ struct Storage {
|
||||
size_t listener_token {1};
|
||||
|
||||
/* The maximum number of values we store for a given hash. */
|
||||
static constexpr unsigned MAX_VALUES {64 * 1024};
|
||||
static constexpr unsigned MAX_VALUES {1024};
|
||||
|
||||
/**
|
||||
* Changes caused by an operation on the storage.
|
||||
@@ -158,14 +144,11 @@ struct Storage {
|
||||
* @return time of the next expiration, time_point::max() if no expiration
|
||||
*/
|
||||
std::pair<ValueStorage*, time_point>
|
||||
refresh(const InfoHash& id, const time_point& now, const Value::Id& vid, const TypeStore& types) {
|
||||
refresh(const time_point& now, const Value::Id& vid, const TypeStore& types) {
|
||||
for (auto& vs : values)
|
||||
if (vs.data->id == vid) {
|
||||
vs.created = now;
|
||||
auto oldExp = vs.expiration;
|
||||
vs.expiration = std::max(oldExp, now + types.getType(vs.data->type).expiration);
|
||||
if (vs.store_bucket)
|
||||
vs.store_bucket->refresh(id, *vs.data, oldExp, vs.expiration);
|
||||
vs.expiration = std::max(vs.expiration, now + types.getType(vs.data->type).expiration);
|
||||
return {&vs, vs.expiration};
|
||||
}
|
||||
return {nullptr, time_point::max()};
|
||||
@@ -177,7 +160,7 @@ struct Storage {
|
||||
local_listeners.erase(token);
|
||||
}
|
||||
|
||||
Sp<Value> remove(const InfoHash& id, Value::Id);
|
||||
StoreDiff remove(const InfoHash& id, Value::Id);
|
||||
|
||||
std::pair<ssize_t, std::vector<Sp<Value>>> expire(const InfoHash& id, time_point now);
|
||||
|
||||
@@ -246,7 +229,7 @@ Storage::store(const InfoHash& id, const Sp<Value>& value, time_point created, t
|
||||
return std::make_pair(nullptr, StoreDiff{});
|
||||
}
|
||||
|
||||
Sp<Value>
|
||||
Storage::StoreDiff
|
||||
Storage::remove(const InfoHash& id, Value::Id vid)
|
||||
{
|
||||
auto it = std::find_if (values.begin(), values.end(), [&](const ValueStorage& vr) {
|
||||
@@ -260,9 +243,8 @@ Storage::remove(const InfoHash& id, Value::Id vid)
|
||||
if (it->expiration_job)
|
||||
it->expiration_job->cancel();
|
||||
total_size -= size;
|
||||
auto value = it->data;
|
||||
values.erase(it);
|
||||
return value;
|
||||
return {-size, -1, 0};
|
||||
}
|
||||
|
||||
Storage::StoreDiff
|
||||
|
||||
+1
-1
@@ -106,7 +106,7 @@ print_addr(const sockaddr* sa, socklen_t slen)
|
||||
{
|
||||
char hbuf[NI_MAXHOST];
|
||||
char sbuf[NI_MAXSERV];
|
||||
std::ostringstream out;
|
||||
std::stringstream out;
|
||||
if (sa and slen and !getnameinfo(sa, slen, hbuf, sizeof(hbuf), sbuf, sizeof(sbuf), NI_NUMERICHOST | NI_NUMERICSERV)) {
|
||||
if (sa->sa_family == AF_INET6)
|
||||
out << "[" << hbuf << "]";
|
||||
|
||||
+1
-1
@@ -53,7 +53,7 @@ public:
|
||||
CallbackQueue clear() {
|
||||
std::vector<Sp<Value>> expired_values;
|
||||
expired_values.reserve(values.size());
|
||||
for (auto& v : values)
|
||||
for (const auto& v : values)
|
||||
expired_values.emplace_back(std::move(v.second.data));
|
||||
values.clear();
|
||||
CallbackQueue ret;
|
||||
|
||||
+11
-22
@@ -36,8 +36,8 @@ DhtRunnerTester::setUp() {
|
||||
config.dht_config.node_config.max_peer_req_per_sec = -1;
|
||||
config.dht_config.node_config.max_req_per_sec = -1;
|
||||
|
||||
node1.run(0, config);
|
||||
node2.run(0, config);
|
||||
node1.run(42222, config);
|
||||
node2.run(42232, config);
|
||||
node2.bootstrap(node1.getBound());
|
||||
}
|
||||
|
||||
@@ -54,17 +54,15 @@ DhtRunnerTester::tearDown() {
|
||||
node1.shutdown(shutdown);
|
||||
node2.shutdown(shutdown);
|
||||
std::unique_lock<std::mutex> lk(cv_m);
|
||||
CPPUNIT_ASSERT(cv.wait_for(lk, 30s, [&]{ return done == 2u; }));
|
||||
CPPUNIT_ASSERT(cv.wait_for(lk, 12s, [&]{ return done == 2; }));
|
||||
node1.join();
|
||||
node2.join();
|
||||
}
|
||||
|
||||
void
|
||||
DhtRunnerTester::testConstructors() {
|
||||
CPPUNIT_ASSERT(node1.getBoundPort());
|
||||
CPPUNIT_ASSERT_EQUAL(node1.getBoundPort(), node1.getBound().getPort());
|
||||
CPPUNIT_ASSERT(node2.getBoundPort());
|
||||
CPPUNIT_ASSERT_EQUAL(node2.getBoundPort(), node2.getBound().getPort());
|
||||
CPPUNIT_ASSERT(node1.getBoundPort() == 42222);
|
||||
CPPUNIT_ASSERT(node2.getBoundPort() == 42232);
|
||||
|
||||
dht::DhtRunner::Config config {};
|
||||
dht::DhtRunner::Context context {};
|
||||
@@ -94,15 +92,13 @@ DhtRunnerTester::testListen() {
|
||||
std::condition_variable cv;
|
||||
std::atomic_uint valueCount(0);
|
||||
unsigned putCount(0);
|
||||
unsigned putOkCount1(0);
|
||||
unsigned putOkCount2(0);
|
||||
unsigned putOkCount3(0);
|
||||
unsigned putOkCount(0);
|
||||
|
||||
auto a = dht::InfoHash::get("234");
|
||||
auto b = dht::InfoHash::get("2345");
|
||||
auto c = dht::InfoHash::get("23456");
|
||||
auto d = dht::InfoHash::get("234567");
|
||||
constexpr unsigned N = 2048;
|
||||
constexpr unsigned N = 256;
|
||||
constexpr unsigned SZ = 56 * 1024;
|
||||
|
||||
auto ftokena = node1.listen(a, [&](const std::shared_ptr<dht::Value>&) {
|
||||
@@ -134,13 +130,13 @@ DhtRunnerTester::testListen() {
|
||||
node2.put(a, dht::Value("v1"), [&](bool ok) {
|
||||
std::lock_guard<std::mutex> lock(mutex);
|
||||
putCount++;
|
||||
if (ok) putOkCount1++;
|
||||
if (ok) putOkCount++;
|
||||
cv.notify_all();
|
||||
});
|
||||
node2.put(b, dht::Value("v2"), [&](bool ok) {
|
||||
std::lock_guard<std::mutex> lock(mutex);
|
||||
putCount++;
|
||||
if (ok) putOkCount2++;
|
||||
if (ok) putOkCount++;
|
||||
cv.notify_all();
|
||||
});
|
||||
auto bigVal = std::make_shared<dht::Value>();
|
||||
@@ -148,7 +144,7 @@ DhtRunnerTester::testListen() {
|
||||
node2.put(c, bigVal, [&](bool ok) {
|
||||
std::lock_guard<std::mutex> lock(mutex);
|
||||
putCount++;
|
||||
if (ok) putOkCount3++;
|
||||
if (ok) putOkCount++;
|
||||
cv.notify_all();
|
||||
});
|
||||
}
|
||||
@@ -156,9 +152,7 @@ DhtRunnerTester::testListen() {
|
||||
{
|
||||
std::unique_lock<std::mutex> lk(mutex);
|
||||
CPPUNIT_ASSERT(cv.wait_for(lk, 30s, [&]{ return putCount == N * 3u; }));
|
||||
CPPUNIT_ASSERT_EQUAL(N, putOkCount1);
|
||||
CPPUNIT_ASSERT_EQUAL(N, putOkCount2);
|
||||
CPPUNIT_ASSERT_EQUAL(N, putOkCount3);
|
||||
CPPUNIT_ASSERT_EQUAL(N * 3u, putOkCount);
|
||||
}
|
||||
|
||||
CPPUNIT_ASSERT(ftokena.valid());
|
||||
@@ -205,11 +199,6 @@ DhtRunnerTester::testIdOps() {
|
||||
node2.run(42232, config2, std::move(context2));
|
||||
node2.bootstrap(node1.getBound());
|
||||
|
||||
{
|
||||
std::unique_lock<std::mutex> lk(mutex);
|
||||
CPPUNIT_ASSERT(cv.wait_for(lk, 20s, [&]{ return valueCount == 1; }));
|
||||
}
|
||||
|
||||
node1.findCertificate(node2.getId(), [&](const std::shared_ptr<dht::crypto::Certificate>& crt){
|
||||
CPPUNIT_ASSERT(crt);
|
||||
std::lock_guard<std::mutex> lk(mutex);
|
||||
|
||||
@@ -1,9 +1,7 @@
|
||||
if (OPENDHT_SHARED)
|
||||
set (OPENDHT_LIBS opendht)
|
||||
set (OPENDHT_C_LIBS opendht-c)
|
||||
else ()
|
||||
set (OPENDHT_LIBS opendht-static)
|
||||
set (OPENDHT_C_LIBS opendht-c-static)
|
||||
if (MSVC)
|
||||
set (MSC_COMPAT_SOURCES ${MSC_COMPAT_DIR}/wingetopt.c)
|
||||
endif ()
|
||||
@@ -11,8 +9,8 @@ endif ()
|
||||
|
||||
function (configure_tool name extra_files)
|
||||
add_executable (${name} ${name}.cpp ${extra_files})
|
||||
add_dependencies(${name} ${OPENDHT_LIBS})
|
||||
target_link_libraries (${name} LINK_PUBLIC ${OPENDHT_LIBS} ${READLINE_LIBRARIES})
|
||||
target_link_libraries (${name} LINK_PUBLIC ${READLINE_LIBRARIES})
|
||||
target_link_libraries (${name} LINK_PUBLIC ${OPENDHT_LIBS})
|
||||
if (MSVC)
|
||||
target_sources(${name} PRIVATE ${MSC_COMPAT_SOURCES})
|
||||
target_include_directories (${name} PRIVATE ${MSC_COMPAT_DIR})
|
||||
@@ -31,8 +29,7 @@ endif ()
|
||||
|
||||
if (OPENDHT_C)
|
||||
add_executable (dhtcnode dhtcnode.c)
|
||||
add_dependencies(dhtcnode ${OPENDHT_C_LIBS})
|
||||
target_link_libraries (dhtcnode LINK_PUBLIC ${OPENDHT_C_LIBS} ${READLINE_LIBRARIES})
|
||||
target_link_libraries (dhtcnode LINK_PUBLIC opendht-c ${READLINE_LIBRARIES})
|
||||
target_include_directories (dhtcnode SYSTEM PRIVATE ${PROJECT_SOURCE_DIR}/c)
|
||||
endif ()
|
||||
|
||||
|
||||
@@ -11,10 +11,3 @@ dhtchat_LDFLAGS = -lopendht -lreadline -L@top_builddir@/src/.libs @GnuTLS_LIBS@
|
||||
|
||||
dhtscanner_SOURCES = dhtscanner.cpp
|
||||
dhtscanner_LDFLAGS = -lopendht -lreadline -L@top_builddir@/src/.libs @GnuTLS_LIBS@
|
||||
|
||||
if ENABLE_C
|
||||
bin_PROGRAMS += dhtcnode
|
||||
dhtcnode_CFLAGS = -std=c11 -isystem @top_srcdir@/c -isystem @top_srcdir@/include
|
||||
dhtcnode_SOURCES = dhtcnode.c
|
||||
dhtcnode_LDFLAGS = -lopendht-c -lreadline -L@top_builddir@/c/.libs
|
||||
endif
|
||||
+59
-186
@@ -1,41 +1,14 @@
|
||||
/*
|
||||
* Copyright (C) 2014-2022 Savoir-faire Linux Inc.
|
||||
* Author : Adrien Béraud <adrien.beraud@savoirfairelinux.com>
|
||||
*
|
||||
* This program is free software; you can redistribute it and/or modify
|
||||
* it under the terms of the GNU General Public License as published by
|
||||
* the Free Software Foundation; either version 3 of the License, or
|
||||
* (at your option) any later version.
|
||||
*
|
||||
* This program is distributed in the hope that it will be useful,
|
||||
* but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
* GNU General Public License for more details.
|
||||
*
|
||||
* You should have received a copy of the GNU General Public License
|
||||
* along with this program. If not, see <https://www.gnu.org/licenses/>.
|
||||
*/
|
||||
|
||||
#include <opendht_c.h>
|
||||
|
||||
#include <getopt.h>
|
||||
#include <readline/readline.h>
|
||||
#include <readline/history.h>
|
||||
#include <arpa/inet.h>
|
||||
|
||||
#include <stdio.h>
|
||||
#include <string.h>
|
||||
#include <unistd.h>
|
||||
#include <stdlib.h>
|
||||
#include <stdatomic.h>
|
||||
#include <arpa/inet.h>
|
||||
|
||||
struct op_context {
|
||||
dht_runner* runner;
|
||||
atomic_bool stop;
|
||||
};
|
||||
struct listen_context {
|
||||
dht_runner* runner;
|
||||
dht_op_token* token;
|
||||
int d;
|
||||
};
|
||||
|
||||
bool dht_value_callback(const dht_value* value, bool expired, void* user_data)
|
||||
@@ -53,30 +26,16 @@ bool dht_get_callback(const dht_value* value, void* user_data)
|
||||
return true;
|
||||
}
|
||||
|
||||
void dht_get_done_callback(bool ok, void* user_data)
|
||||
void dht_done_callback(bool ok, void* user_data)
|
||||
{
|
||||
dht_runner* runner = (dht_runner*)user_data;
|
||||
printf("Get completed: %s\n", ok ? "success !" : "failure :-(");
|
||||
printf("Done callback. %s\n", ok ? "Success !" : "Failure :-(");
|
||||
}
|
||||
|
||||
void dht_put_done_callback(bool ok, void* user_data)
|
||||
void op_context_free(void* user_data)
|
||||
{
|
||||
dht_runner* runner = (dht_runner*)user_data;
|
||||
printf("Put completed: %s\n", ok ? "success !" : "failure :-(");
|
||||
}
|
||||
|
||||
void dht_shutdown_callback(void* user_data)
|
||||
{
|
||||
printf("Stopped.\n");
|
||||
struct op_context* ctx = (struct op_context*)user_data;
|
||||
atomic_store(&ctx->stop, true);
|
||||
}
|
||||
|
||||
void listen_context_free(void* user_data)
|
||||
{
|
||||
printf("listen_context_free.\n");
|
||||
struct listen_context* ctx = (struct listen_context*)user_data;
|
||||
dht_op_token_delete(ctx->token);
|
||||
printf("op_context_free %d.\n", ctx->d);
|
||||
free(ctx);
|
||||
}
|
||||
|
||||
@@ -101,152 +60,66 @@ char* print_addr(const struct sockaddr* addr) {
|
||||
return s;
|
||||
}
|
||||
|
||||
struct dht_params {
|
||||
bool help;
|
||||
bool version;
|
||||
bool generate_identity;
|
||||
bool service;
|
||||
bool peer_discovery;
|
||||
bool log;
|
||||
const char* bootstrap;
|
||||
unsigned network;
|
||||
in_port_t port;
|
||||
};
|
||||
|
||||
static const struct option long_options[] = {
|
||||
{"help", no_argument , NULL, 'h'},
|
||||
{"port", required_argument, NULL, 'p'},
|
||||
{"net", required_argument, NULL, 'n'},
|
||||
{"bootstrap", required_argument, NULL, 'b'},
|
||||
{"identity", no_argument , NULL, 'i'},
|
||||
{"verbose", no_argument , NULL, 'v'},
|
||||
{"service", no_argument , NULL, 's'},
|
||||
{"peer-discovery", no_argument , NULL, 'D'},
|
||||
{"no-rate-limit", no_argument , NULL, 'U'},
|
||||
{"persist", required_argument, NULL, 'f'},
|
||||
{"logfile", required_argument, NULL, 'l'},
|
||||
{"syslog", no_argument , NULL, 'L'},
|
||||
{NULL, 0 , NULL, 0}
|
||||
};
|
||||
|
||||
struct dht_params
|
||||
parse_args(int argc, char **argv) {
|
||||
struct dht_params params;
|
||||
bzero(¶ms, sizeof params);
|
||||
int opt;
|
||||
while ((opt = getopt_long(argc, argv, "hisvDp:n:b:f:l:", long_options, NULL)) != -1) {
|
||||
switch (opt) {
|
||||
case 'p': {
|
||||
int port_arg = atoi(optarg);
|
||||
if (port_arg >= 0 && port_arg < 0x10000)
|
||||
params.port = port_arg;
|
||||
}
|
||||
break;
|
||||
case 'D':
|
||||
params.peer_discovery = true;
|
||||
break;
|
||||
case 'n':
|
||||
params.network = strtoul(optarg, NULL, 0);
|
||||
break;
|
||||
case 'b':
|
||||
params.bootstrap = (optarg[0] == '=') ? optarg+1 : optarg;
|
||||
break;
|
||||
case 'h':
|
||||
params.help = true;
|
||||
break;
|
||||
case 'v':
|
||||
params.log = true;
|
||||
break;
|
||||
case 'i':
|
||||
params.generate_identity = true;
|
||||
break;
|
||||
case 's':
|
||||
params.service = true;
|
||||
break;
|
||||
default:
|
||||
break;
|
||||
}
|
||||
}
|
||||
return params;
|
||||
}
|
||||
|
||||
int main(int argc, char **argv)
|
||||
int main()
|
||||
{
|
||||
struct dht_params params = parse_args(argc, argv);
|
||||
dht_identity id = dht_identity_generate("testNode", NULL);
|
||||
dht_infohash cert_id = dht_certificate_get_id(id.certificate);
|
||||
printf("Cert ID: %s\n", dht_infohash_print(&cert_id));
|
||||
|
||||
dht_publickey* pk = dht_certificate_get_publickey(id.certificate);
|
||||
dht_infohash pk_id = dht_publickey_get_id(pk);
|
||||
printf("PK ID: %s\n", dht_infohash_print(&pk_id));
|
||||
dht_publickey_delete(pk);
|
||||
|
||||
pk = dht_privatekey_get_publickey(id.privatekey);
|
||||
pk_id = dht_publickey_get_id(pk);
|
||||
printf("Key ID: %s\n", dht_infohash_print(&pk_id));
|
||||
dht_publickey_delete(pk);
|
||||
|
||||
dht_identity_delete(&id);
|
||||
|
||||
dht_runner* runner = dht_runner_new();
|
||||
dht_runner_config dht_config;
|
||||
dht_runner_config_default(&dht_config);
|
||||
dht_config.peer_discovery = params.peer_discovery; // Look for other peers on the network
|
||||
dht_config.peer_publish = params.peer_discovery; // Publish our own peer info
|
||||
dht_config.dht_config.node_config.network = params.network;
|
||||
dht_config.log = params.log;
|
||||
dht_runner_run_config(runner, params.port, &dht_config);
|
||||
dht_runner_run(runner, 4040);
|
||||
|
||||
if (params.bootstrap) {
|
||||
printf("Bootstrap using %s\n", params.bootstrap);
|
||||
dht_runner_bootstrap(runner, params.bootstrap, NULL);
|
||||
dht_infohash h;
|
||||
dht_infohash_random(&h);
|
||||
|
||||
printf("random hash: %s\n", dht_infohash_print(&h));
|
||||
|
||||
// Put data
|
||||
const char* data_str = "yo, this is some data";
|
||||
dht_value* val = dht_value_new_from_string(data_str);
|
||||
dht_runner_put(runner, &h, val, dht_done_callback, runner, false);
|
||||
dht_value_unref(val);
|
||||
|
||||
// Get data
|
||||
dht_runner_get(runner, &h, dht_get_callback, dht_done_callback, runner);
|
||||
|
||||
// Listen for data
|
||||
struct op_context* ctx = malloc(sizeof(struct op_context));
|
||||
ctx->runner = runner;
|
||||
ctx->d = 42;
|
||||
dht_op_token* token = dht_runner_listen(runner, &h, dht_value_callback, op_context_free, ctx);
|
||||
|
||||
sleep(1);
|
||||
|
||||
dht_runner_bootstrap(runner, "bootstrap.jami.net", NULL);
|
||||
|
||||
sleep(2);
|
||||
|
||||
struct sockaddr** addrs = dht_runner_get_public_address(runner);
|
||||
for (struct sockaddr** addrIt = addrs; *addrIt; addrIt++) {
|
||||
struct sockaddr* addr = *addrIt;
|
||||
char* addr_str = print_addr(addr);
|
||||
free(addr);
|
||||
printf("Found public address: %s\n", addr_str);
|
||||
free(addr_str);
|
||||
}
|
||||
free(addrs);
|
||||
|
||||
char cmd[64];
|
||||
char arg[64];
|
||||
char value[256];
|
||||
while (true) {
|
||||
const char* line_read = readline("> ");
|
||||
if (line_read && *line_read)
|
||||
add_history(line_read);
|
||||
if (!line_read)
|
||||
break;
|
||||
if (!strcmp(line_read, "\0"))
|
||||
continue;
|
||||
dht_runner_cancel_listen(runner, &h, token);
|
||||
dht_op_token_delete(token);
|
||||
|
||||
bzero(cmd, sizeof cmd);
|
||||
bzero(arg, sizeof arg);
|
||||
bzero(value, sizeof value);
|
||||
sscanf(line_read, "%64s %64s %256s", cmd, arg, value);
|
||||
|
||||
if (!strcmp(cmd, "la")) {
|
||||
struct sockaddr** addrs = dht_runner_get_public_address(runner);
|
||||
if (addrs) {
|
||||
for (struct sockaddr** addrIt = addrs; *addrIt; addrIt++) {
|
||||
struct sockaddr* addr = *addrIt;
|
||||
char* addr_str = print_addr(addr);
|
||||
free(addr);
|
||||
printf("Found public address: %s\n", addr_str);
|
||||
free(addr_str);
|
||||
}
|
||||
free(addrs);
|
||||
}
|
||||
continue;
|
||||
}
|
||||
|
||||
dht_infohash key;
|
||||
dht_infohash_from_hex(&key, arg);
|
||||
if (dht_infohash_is_zero(&key)) {
|
||||
dht_infohash_get_from_string(&key, arg);
|
||||
printf("Using h(%s) = %s\n", arg, dht_infohash_print(&key));
|
||||
}
|
||||
if (!strcmp(cmd, "g")) {
|
||||
dht_runner_get(runner, &key, dht_get_callback, dht_get_done_callback, runner);
|
||||
} else if (!strcmp(cmd, "l")) {
|
||||
struct listen_context* ctx = malloc(sizeof(struct listen_context));
|
||||
ctx->runner = runner;
|
||||
ctx->token = dht_runner_listen(runner, &key, dht_value_callback, listen_context_free, ctx);
|
||||
} else if (!strcmp(cmd, "p")) {
|
||||
dht_value* val = dht_value_new_from_string(value);
|
||||
dht_runner_put(runner, &key, val, dht_put_done_callback, runner, true);
|
||||
dht_value_unref(val);
|
||||
}
|
||||
}
|
||||
printf("Stopping..\n");
|
||||
|
||||
struct op_context ctx;
|
||||
ctx.runner = runner;
|
||||
atomic_init(&ctx.stop, false);
|
||||
dht_runner_shutdown(runner, dht_shutdown_callback, &ctx);
|
||||
while (!atomic_load(&ctx.stop)) {
|
||||
usleep(10000);
|
||||
}
|
||||
dht_runner_delete(runner);
|
||||
return 0;
|
||||
}
|
||||
Reference in new issue
Block a user