diff --git a/CMakeLists.txt b/CMakeLists.txt index 3eeaf38c..34dd5cfb 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -19,6 +19,7 @@ option (OPENDHT_LTO "Build with LTO" OFF) # Dependencies list (APPEND CMAKE_MODULE_PATH "${PROJECT_SOURCE_DIR}/cmake") find_package (PkgConfig) +pkg_search_module(libuv libuv) find_package (GnuTLS 3.3 REQUIRED) pkg_search_module (Nettle nettle) find_package (Msgpack 1.2 REQUIRED) @@ -26,7 +27,6 @@ if (OPENDHT_TOOLS) find_package (Readline 6 REQUIRED) endif () if (NOT OPENDHT_ARGON2) - find_package(PkgConfig) pkg_search_module(argon2 libargon2) if (argon2_FOUND) message("Argon2 found " ${argon2_req}) @@ -38,7 +38,7 @@ if (NOT OPENDHT_ARGON2) endif () # Build flags -set (CMAKE_CXX_STANDARD 11) +set (CMAKE_CXX_STANDARD 14) set (CMAKE_CXX_STANDARD_REQUIRED on) set (CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -pthread -Wno-return-type -Wall -Wextra -Wnon-virtual-dtor -pedantic-errors -fvisibility=hidden") set (CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -DMSGPACK_DISABLE_LEGACY_NIL -DMSGPACK_DISABLE_LEGACY_CONVERT") @@ -167,7 +167,7 @@ if (OPENDHT_STATIC) target_link_libraries(opendht-static ${argon2_LIBRARIES}) target_include_directories(opendht-static SYSTEM PRIVATE ${argon2_INCLUDE_DIRS}) endif () - target_link_libraries(opendht-static ${GNUTLS_LIBRARIES} ${Nettle_LIBRARIES}) + target_link_libraries(opendht-static ${GNUTLS_LIBRARIES} ${Nettle_LIBRARIES} ${libuv_LIBRARIES}) install (TARGETS opendht-static DESTINATION ${CMAKE_INSTALL_LIBDIR} EXPORT opendht) endif () @@ -188,7 +188,7 @@ if (OPENDHT_SHARED) target_link_libraries(opendht ${argon2_LIBRARIES}) target_include_directories(opendht SYSTEM PRIVATE ${argon2_INCLUDE_DIRS}) endif () - target_link_libraries(opendht ${GNUTLS_LIBRARIES} ${Nettle_LIBRARIES}) + target_link_libraries(opendht ${GNUTLS_LIBRARIES} ${Nettle_LIBRARIES} ${libuv_LIBRARIES}) install (TARGETS opendht DESTINATION ${CMAKE_INSTALL_LIBDIR} EXPORT opendht) endif () diff --git a/include/opendht/callbacks.h b/include/opendht/callbacks.h index 21f2dd0c..f96b2a65 100644 --- a/include/opendht/callbacks.h +++ b/include/opendht/callbacks.h @@ -49,6 +49,12 @@ struct OPENDHT_PUBLIC NodeStats { std::string toString() const; }; + +struct OPENDHT_PUBLIC NetworkConfig { + std::string bind_addr {"::0"}; + in_port_t bind_port {0}; +}; + /** * Dht configuration. */ @@ -68,6 +74,8 @@ struct OPENDHT_PUBLIC Config { /** Makes the DHT responsible to maintain its stored values. Consumes more ressources. */ bool maintain_storage; + + NetworkConfig network_config; }; /** @@ -87,6 +95,7 @@ using QueryCallback = std::function>& values)>; using GetCallbackSimple = std::function value)>; using ShutdownCallback = std::function; +using StatusCallback = std::function; using CertificateStoreQuery = std::function>(const InfoHash& pk_id)>; diff --git a/include/opendht/dht.h b/include/opendht/dht.h index 7167fa1f..6fbdcf1b 100644 --- a/include/opendht/dht.h +++ b/include/opendht/dht.h @@ -54,9 +54,6 @@ struct LocalListener; /** * Main Dht class. * Provides a Distributed Hash Table node. - * - * Must be given open UDP sockets and ::periodic must be - * called regularly. */ class OPENDHT_PUBLIC Dht { public: @@ -73,7 +70,7 @@ public: * Initialise the Dht with two open sockets (for IPv4 and IP6) * and an ID for the node. */ - Dht(int s, int s6, Config config); + Dht(uv_loop_t* loop, Config config); virtual ~Dht(); /** @@ -89,6 +86,9 @@ public: NodeStatus getStatus() const { return std::max(getStatus(AF_INET), getStatus(AF_INET6)); } + void setOnStatusChanged(StatusCallback&& cb) { + statusCb = std::move(cb); + } /** * Performs final operations before quitting. @@ -136,13 +136,13 @@ public: insertNode(n.id, SockAddr(n.ss, n.sslen)); } - void pingNode(const sockaddr*, socklen_t, DoneCallbackSimple&& cb={}); - - 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)); + void bootstrap(const std::string& host, const std::string& service); + void clearBootstrap() { + bootstrap_nodes.clear(); } + void pingNode(const SockAddr&, DoneCallbackSimple&& cb={}); + /** * Get a value by searching on all available protocols (IPv4, IPv6), * and call the provided get callback when values are found at key. @@ -307,6 +307,10 @@ public: std::vector getPublicAddress(sa_family_t family = 0); + SockAddr getBoundAddr() { + return network_engine.getBoundAddr(); + } + protected: Logger DHT_LOG; bool logFilerEnable_ {}; @@ -378,6 +382,8 @@ private: const bool is_bootstrap {false}; const bool maintain_storage {false}; + std::vector> bootstrap_nodes {}; + // the stuff RoutingTable buckets4 {}; RoutingTable buckets6 {}; @@ -400,8 +406,7 @@ private: // timing Scheduler scheduler; - Sp nextNodesConfirmation {}; - Sp nextStorageMaintenance {}; + Sp nextNodesConfirmation {}; time_point mybucket_grow_time {time_point::min()}, mybucket6_grow_time {time_point::min()}; net::NetworkEngine network_engine; @@ -410,9 +415,13 @@ private: using ReportedAddr = std::pair; std::vector reported_addr; - std::mt19937_64 rd {crypto::getSeededRandomEngine()}; + + NodeStatus status4 {NodeStatus::Disconnected}, + status6 {NodeStatus::Disconnected}; + StatusCallback statusCb {nullptr}; + void updateStatus(); void rotateSecrets(); Blob makeToken(const SockAddr&, bool old) const; @@ -554,8 +563,6 @@ private: bool neighbourhoodMaintenance(RoutingTable&); - void processMessage(const uint8_t *buf, size_t buflen, const SockAddr&); - void onError(Sp node, net::DhtProtocolException e); /* when our address is reported by a distant peer. */ void onReportedAddr(const InfoHash& id, const SockAddr&); diff --git a/include/opendht/dhtrunner.h b/include/opendht/dhtrunner.h index 0007c31a..dc8d7ee9 100644 --- a/include/opendht/dhtrunner.h +++ b/include/opendht/dhtrunner.h @@ -26,6 +26,7 @@ #include "sockaddr.h" #include "log_enable.h" #include "def.h" +#include "uv_utils.h" #include #include @@ -51,8 +52,6 @@ struct SecureDhtConfig; class OPENDHT_PUBLIC DhtRunner { public: - typedef std::function StatusCallback; - DhtRunner(); virtual ~DhtRunner(); @@ -241,16 +240,14 @@ 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 { - return (f == AF_INET) ? bound4 : bound6; - } + SockAddr getBound() const; /** * Returns the currently bound port, in host byte order. * @param f: address family of the bound port to retreive. */ - in_port_t getBoundPort(sa_family_t f = AF_INET) const { - return ntohs(((sockaddr_in*)&getBound(f).first)->sin_port); + in_port_t getBoundPort() const { + return getBound().getPort(); } std::pair getStoreSize() const; @@ -306,20 +303,24 @@ public: * @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) { - run(port, { + run({ /*.dht_config = */{ /*.node_config = */{ /*.node_id = */{}, /*.network = */network, /*.is_bootstrap = */false, - /*.maintain_storage*/false + /*.maintain_storage = */false, + /* .network_config = */{ + /*.addr = */ "::0", + /*.port = */ port + } }, /*.id = */identity }, /*.threaded = */threaded }); } - void run(in_port_t port, Config config); + void run(Config config); /** * @param local4: Local IPv4 address and port to bind. Can be null. @@ -329,25 +330,22 @@ 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(const sockaddr_in* local4, const sockaddr_in6* local6, Config config); + //void run(const sockaddr_in* local4, const sockaddr_in6* local6, Config config); /** * Same as @run(sockaddr_in, sockaddr_in6, Identity, bool, StatusCallback), but with string IP addresses and service (port). */ - void run(const char* ip4, const char* ip6, const char* service, Config config); + //void run(const char* ip4, const char* ip6, const char* service, Config config); - void setOnStatusChanged(StatusCallback&& cb) { - statusCb = std::move(cb); - } + void setOnStatusChanged(StatusCallback&& cb); /** * In non-threaded mode, the user should call this method * regularly and everytime a new packet is received. * @return the next op */ - time_point loop() { - std::lock_guard lck(dht_mtx); - return loop_(); + void loop() { + uv_loop_.runOnce(); } /** @@ -373,33 +371,41 @@ private: */ void tryBootstrapContinuously(); - void doRun(const sockaddr_in* sin4, const sockaddr_in6* sin6, SecureDhtConfig config); - time_point loop_(); + void doRun(SecureDhtConfig config); + void loop_(); static std::vector> getAddrInfo(const std::string& host, const std::string& service); - NodeStatus getStatus() const { - return std::max(status4, status6); + void loopCallback() { + std::lock_guard lck(dht_mtx); + if (running) + loop_(); + else { + uv_close((uv_handle_t*)&uv_async_, nullptr); + uv_loop_.stop(); + } + } + + static void loop_callback(uv_async_t* handle) { + if (handle->data) + static_cast(handle->data)->loopCallback(); } std::unique_ptr dht_; + EventLoop uv_loop_; + uv_async_t uv_async_; mutable std::mutex dht_mtx {}; std::thread dht_thread {}; - std::condition_variable cv {}; - - std::thread rcv_thread {}; - std::mutex sock_mtx {}; - std::vector> rcv {}; /** true if currently actively boostraping */ std::atomic_bool bootstraping {false}; /* bootstrap nodes given as (host, service) pairs */ - std::vector> bootstrap_nodes_all {}; + /*std::vector> bootstrap_nodes_all {}; std::vector> bootstrap_nodes {}; - std::thread bootstrap_thread {}; + std::thread bootstrap_thread {};*/ /** protects bootstrap_nodes, bootstrap_thread */ - std::mutex bootstrap_mtx {}; - std::condition_variable bootstrap_cv {}; + /*std::mutex bootstrap_mtx {}; + std::condition_variable bootstrap_cv {};*/ std::queue> pending_ops_prio {}; std::queue> pending_ops {}; @@ -407,12 +413,12 @@ private: std::atomic running {false}; - NodeStatus status4 {NodeStatus::Disconnected}, + /*NodeStatus status4 {NodeStatus::Disconnected}, status6 {NodeStatus::Disconnected}; StatusCallback statusCb {nullptr}; SockAddr bound4 {}; - SockAddr bound6 {}; + SockAddr bound6 {};*/ }; } diff --git a/include/opendht/network_engine.h b/include/opendht/network_engine.h index 8cf6827a..e6e9ddb3 100644 --- a/include/opendht/network_engine.h +++ b/include/opendht/network_engine.h @@ -28,6 +28,8 @@ #include "rng.h" #include "rate_limiter.h" #include "net.h" +#include "callbacks.h" +#include "uv_utils.h" #include #include @@ -210,8 +212,8 @@ public: using RequestCb = std::function; using RequestExpiredCb = std::function; - NetworkEngine(Logger& log, Scheduler& scheduler); - NetworkEngine(InfoHash& myid, NetId net, int s, int s6, Logger& log, Scheduler& scheduler, + NetworkEngine(uv_loop_t*, Logger& log, Scheduler& scheduler); + NetworkEngine(uv_loop_t*, const NetworkConfig& config, InfoHash& myid, NetId net, Logger& log, Scheduler& scheduler, decltype(NetworkEngine::onError) onError, decltype(NetworkEngine::onNewNode) onNewNode, decltype(NetworkEngine::onReportedAddr) onReportedAddr, @@ -225,6 +227,9 @@ public: virtual ~NetworkEngine(); void clear(); + void close(OnClose cb) { + sock->close(cb); + } /** * Sends values (with closest nodes) to a listenner. @@ -244,7 +249,7 @@ public: std::vector>&& values, const Query& q); bool isRunning(sa_family_t af) const; - inline want_t want () const { return dht_socket >= 0 && dht_socket6 >= 0 ? (WANT4 | WANT6) : -1; } + inline want_t want () const { /*return dht_socket >= 0 && dht_socket6 >= 0 ? (WANT4 | WANT6) : -1;*/return WANT4 | WANT6; } /** * Cancel a request. Setting req->cancelled = true is not enough in the case @@ -280,8 +285,8 @@ public: * @return the request with information concerning its success. */ Sp - sendPing(const sockaddr* sa, socklen_t salen, RequestCb&& on_done, RequestExpiredCb&& on_expired) { - return sendPing(std::make_shared(zeroes, sa, salen), + sendPing(const SockAddr& sa, RequestCb&& on_done, RequestExpiredCb&& on_expired) { + return sendPing(std::make_shared(zeroes, sa), std::forward(on_done), std::forward(on_expired)); } @@ -440,6 +445,10 @@ public: return cache.getCachedNodes(id, sa_f, count); } + SockAddr getBoundAddr() { + return sock->getBoundAddr(); + } + private: struct PartialMessage; @@ -505,7 +514,8 @@ private: // basic wrapper for socket sendto function - int send(const char *buf, size_t len, int flags, const SockAddr& addr); + int send(msgpack::sbuffer& msg, const SockAddr& addr); + int send(const Blob& msg, int flags, const SockAddr& addr); void sendValueParts(TransId tid, const std::vector& svals, const SockAddr& addr); std::vector packValueHeader(msgpack::sbuffer&, const std::vector>&); @@ -547,10 +557,9 @@ private: /* DHT info */ const InfoHash& myid; const NetId network {0}; - const int dht_socket {-1}; - const int dht_socket6 {-1}; const Logger& DHT_LOG; + std::shared_ptr sock; NodeCache cache {}; /** diff --git a/include/opendht/scheduler.h b/include/opendht/scheduler.h index 8f61d4cd..23dc4621 100644 --- a/include/opendht/scheduler.h +++ b/include/opendht/scheduler.h @@ -22,6 +22,9 @@ #include "utils.h" #include "log_enable.h" +#include "uv_utils.h" + +#include #include #include @@ -36,12 +39,7 @@ namespace dht { */ class Scheduler { public: - Scheduler(const Logger& l) : DHT_LOG(l) {} - - struct Job { - Job(std::function&& f) : do_(std::move(f)) {} - std::function do_; - }; + Scheduler(uv_loop_t* loop, const Logger& l) : loop_(loop), DHT_LOG(l) {} /** * Adds another job to the queue. @@ -51,10 +49,32 @@ public: * * @return pointer to the newly scheduled job. */ - Sp add(time_point t, std::function&& job_func) { - auto job = std::make_shared(std::move(job_func)); - if (t != time_point::max()) - timers.emplace(std::move(t), job); + Sp add(const time_point& t, std::function&& job_func) { + if (not loop_) + return {}; + auto job = Job::make(loop_, std::move(job_func)); + if (t != time_point::max()) { + garbageCollection(); + pending_jobs.emplace(job); + job->run(t); + } else + job->cancel(); + return job; + } + Sp add(const duration& t, std::function&& job_func) { + if (not loop_) + return {}; + auto job = Job::make(loop_, std::move(job_func)); + garbageCollection(); + pending_jobs.emplace(job); + job->run(t); + return job; + } + Sp run(std::function&& job_func) { + auto job = Job::make(loop_, std::move(job_func)); + garbageCollection(); + pending_jobs.emplace(job); + job->run(); return job; } @@ -66,7 +86,9 @@ public: * * @return pointer to the newly scheduled job. */ - void edit(Sp& job, time_point t) { + void edit(Sp& job, const duration& t = {}) { + if (not loop_) + return; if (not job) { DHT_LOG.ERR("editing an empty job"); return; @@ -75,50 +97,73 @@ public: // Force clearing old value. auto task = std::move(job->do_); job->do_ = {}; + job->cancel(); + job = add(t, std::move(task)); + } + void edit(Sp& job, const time_point& t) { + if (not loop_) + return; + if (not job) { + DHT_LOG.ERR("editing an empty job"); + return; + } + auto task = std::move(job->do_); + job->do_ = {}; + job->cancel(); job = add(t, std::move(task)); } - /** - * Runs the jobs to do up to now. - * - * @return The time for the next job to run. - */ - time_point run() { - syncTime(); - while (not timers.empty()) { - auto timer = timers.begin(); - /* - * Running jobs scheduled before "now" prevents run+rescheduling - * loops before this method ends. It is garanteed by the fact that a - * job will at least be scheduled for "now" and not before. - */ - if (timer->first > now) - break; - - auto job = std::move(timer->second); - timers.erase(timer); - - if (job->do_) - job->do_(); - } - return getNextJobTime(); + void stop() { + loop_ = nullptr; + size_t del {0}; + for (auto& w : pending_jobs) + if (auto job = w.lock()) { + job->cancel(); + del++; + } + pending_jobs.clear(); + DHT_LOG.DEBUG("stopping scheduler, %lu events canceled", del); } - inline time_point getNextJobTime() const { - return timers.empty() ? time_point::max() : timers.begin()->first; + ~Scheduler(){ + stop(); } /** * Accessors for the common time reference used for synchronizing * operations. */ - inline const time_point& time() const { return now; } - inline time_point syncTime() { return (now = clock::now()); } + inline time_point time() const { + return time_point(std::chrono::duration_cast(std::chrono::milliseconds(uv_now(loop_)))); + } + inline time_point syncTime() { + return time(); + } + uv_loop_t* getLoop() const { + return loop_; + } private: - time_point now {clock::now()}; - std::multimap> timers {}; /* the jobs ordered by time */ + std::set, std::owner_less>> pending_jobs {}; + size_t lastSize_ {0}; + uv_loop_t* loop_; const Logger& DHT_LOG; + + void garbageCollection() { + if (pending_jobs.size() > lastSize_ * 2) { + unsigned del {0}; + for (auto it = pending_jobs.begin(); it != pending_jobs.end();) { + if (it->expired()) { + it = pending_jobs.erase(it); + del++; + } else + ++it; + } + //DHT_LOG.DEBUG("pending_jobs cleanup: %u removed, %zu remaining", del, pending_jobs.size()); + lastSize_ = pending_jobs.size(); + } + } + }; } diff --git a/include/opendht/securedht.h b/include/opendht/securedht.h index acae729f..c8c3d2ee 100644 --- a/include/opendht/securedht.h +++ b/include/opendht/securedht.h @@ -44,7 +44,7 @@ public: * id: the identity to use for the crypto layer and to compute * our own hash on the Dht. */ - SecureDht(int s, int s6, Config config); + SecureDht(uv_loop_t* loop, Config config); virtual ~SecureDht(); diff --git a/include/opendht/sockaddr.h b/include/opendht/sockaddr.h index d2b814d1..a0e00fa0 100644 --- a/include/opendht/sockaddr.h +++ b/include/opendht/sockaddr.h @@ -50,6 +50,15 @@ public: throw std::runtime_error("Socket address length is too large"); std::copy_n((uint8_t*)sa, len, (uint8_t*)&first); } + SockAddr(const sockaddr* sa) : pair::pair({},0) { + if (sa) { + if (sa->sa_family == AF_INET) + second = sizeof(sockaddr_in); + else if(sa->sa_family == AF_INET6) + second = sizeof(sockaddr_in6); + } + std::copy_n((uint8_t*)sa, second, (uint8_t*)&first); + } SockAddr(const sockaddr_storage& ss, socklen_t len) : SockAddr((const sockaddr*)&ss, len) {} bool operator<(const SockAddr& o) const { @@ -117,6 +126,12 @@ public: sockaddr_in6& getIPv6() { return *reinterpret_cast(&first); } + sockaddr* get() { + return reinterpret_cast(&first); + } + const sockaddr* get() const { + return reinterpret_cast(&first); + } /** * Return true if address is a loopback IP address. @@ -130,6 +145,9 @@ public: bool isUnspecified() const; + bool isMappedIPv4() const; + SockAddr getMappedIPv4() const; + /** * A comparator to classify IP addresses, only considering the * first 64 bits in IPv6. diff --git a/include/opendht/uv_utils.h b/include/opendht/uv_utils.h new file mode 100644 index 00000000..8f2ef6f3 --- /dev/null +++ b/include/opendht/uv_utils.h @@ -0,0 +1,319 @@ +#pragma once + +#include "utils.h" + +#include + +namespace dht { + +struct EventLoop { + using OnSignal = std::function; + + EventLoop() { + uv_loop_init(&loop); + uv_signal_init(&loop, &sig_handler); + sig_handler.data = this; + uv_signal_start(&sig_handler, &EventLoop::signal_handler, SIGINT); + } + ~EventLoop() { + uv_loop_close(&loop); + } + auto run() { + return uv_run(&loop, UV_RUN_DEFAULT); + } + auto runOnce() { + return uv_run(&loop, UV_RUN_NOWAIT); + } + void stop() { + uv_signal_stop(&sig_handler); + uv_walk(&loop, &EventLoop::close_walk_cb, this); + } + uv_loop_t* get() { return &loop; } + void setOnSignal(OnSignal cb) { + onSignal_ = cb; + } +private: + uv_loop_t loop; + uv_signal_t sig_handler; + OnSignal onSignal_; + void onSignal(int sig) { + if (onSignal_) + onSignal_(sig); + } + static void close_walk_cb(uv_handle_t* handle, void* /*arg*/) { + if (not uv_is_closing(handle)) + uv_close(handle, [](uv_handle_t* handle) { + std::cout << "handle " << handle << " of type " << handle->type << " closed" << std::endl; + }); + } + static void signal_handler(uv_signal_t* handle, int signum) { + if (handle->data) { + static_cast(handle->data)->onSignal(signum); + } + } +}; + +using GetAddrInfoCb = std::function&&)>; + +void getAddrInfo(uv_loop_t* loop, const char* node, const char* service, GetAddrInfoCb&& cb); + +struct Job : std::enable_shared_from_this { + Job(uv_loop_t* loop, std::function&& f = {}) : do_(std::move(f)) { + uv_timer_init(loop, &timer_); + timer_.data = nullptr; + } + void run(uint64_t t = 0) { + uv_timer_start(&timer_, &Job::timer_cb, t, 0); + } + inline void run(const duration& t) { + run(getTimeout(t)); + } + inline void run(const time_point& t) { + run(getTimeout(t)); + } + void cancel() { + if (timer_.data and not uv_is_closing((uv_handle_t*)&timer_)) { + uv_timer_stop(&timer_); + uv_close((uv_handle_t*)&timer_, &Job::close_cb); + } + } + static Sp make(uv_loop_t* loop, std::function&& f = {}) { + auto job = std::make_shared(loop, std::move(f)); + job->timer_.data = new Sp(job); + return job; + } + static uint64_t getTimeout(const duration& d) { + return std::chrono::duration_cast(d).count(); + } + uint64_t getTimeout(const time_point& tp) const { + uint64_t t = getTimeout(tp.time_since_epoch()); + uint64_t now = uv_now(timer_.loop); + return (t < now) ? 0 : t - now; + } + std::function do_; +private: + uv_timer_t timer_; + void onTimer() { + if (do_) + do_(); + cancel(); + } + void onClosed() { + if (timer_.data) { + delete static_cast*>(timer_.data); + timer_.data = nullptr; + } + } + static void timer_cb(uv_timer_t* handle) { + if (handle->data) { + auto timer = *static_cast*>(handle->data); + timer->onTimer(); + } + } + static void close_cb(uv_handle_t* handle) { + if (handle->data) { + auto timer = *static_cast*>(handle->data); + timer->onClosed(); + } else { + std::cerr << "Job::onClosed(): no data! " << handle << std::endl; + } + } +}; + + +static inline void alloc_buffer(uv_handle_t* /*handle*/, size_t suggested_size, uv_buf_t *buf) { + buf->base = (char*)malloc(suggested_size); + buf->len = suggested_size; +} + + +using OnClose = std::function; + +struct UdpSocket : public std::enable_shared_from_this +{ + using OnReceive = std::function; + UdpSocket(uv_loop_t* loop) { + uv_udp_init(loop, &sock); + sock.data = nullptr; + } + void open(const char* bind_addr, in_port_t port, OnReceive cb) { + if (not sock.data) + sock.data = new std::shared_ptr(shared_from_this()); + receive_cb = cb; + struct sockaddr_in6 addr; + uv_ip6_addr(bind_addr, port, &addr); + uv_udp_bind(&sock, (const struct sockaddr*)&addr, 0); + uv_udp_recv_start(&sock, alloc_buffer, &UdpSocket::on_receive); + } + + int send(uint8_t* data, size_t size, const SockAddr& addr) { + auto req = new SendReq; + req->msg = data; + req->req.data = req; + const uv_buf_t buf { + /*.base =*/ (char*)data, + /*.len =*/ size + }; + return uv_udp_send(&req->req, &sock, &buf, 1, addr.get(), &UdpSocket::on_sent); + } + + void close(OnClose cb) { + close_cb = cb; + uv_close((uv_handle_t*) &sock, &UdpSocket::on_close); + } + bool isRunning() const { + return uv_is_active((uv_handle_t*) &sock); + } + SockAddr getBoundAddr() { + SockAddr addr; + int size = sizeof(addr.first); + uv_udp_getsockname(&sock, addr.get(), &size); + addr.second = size; + return addr; + } +private: + struct SendReq { + uint8_t* msg; + uv_udp_send_t req; + }; + + uv_udp_t sock; + OnReceive receive_cb; + OnClose close_cb; + void onReceive(ssize_t nread, const uv_buf_t* buf, const struct sockaddr* addr, unsigned /*flags*/) { + if (nread > 0 and addr and receive_cb) { + receive_cb((const uint8_t*)buf->base, nread, SockAddr(addr)); + } + } + void onClosed() { + if (close_cb) + close_cb(); + if (sock.data) { + delete static_cast*>(sock.data); + sock.data = nullptr; + } + } + void onSent(SendReq& req, int /*status*/) { + // if (status) fprintf(stderr, "onSent %s\n", uv_strerror(status)); + free(req.msg); + } + static std::shared_ptr get(uv_handle_t* handle) { + return handle->data ? *static_cast*>(handle->data) : std::shared_ptr(); + } + static void on_close(uv_handle_t* handle) { + if (auto node = get(handle)) + node->onClosed(); + } + static void on_sent(uv_udp_send_t* req, int status) { + if (auto data = static_cast(req->data)) { + if (auto node = get((uv_handle_t*)req->handle)) + node->onSent(*data, status); + delete data; + } + } + static void on_receive(uv_udp_t* handle, ssize_t nread, const uv_buf_t* buf, const struct sockaddr* addr, unsigned flags) { + if (auto node = get((uv_handle_t*)handle)) + node->onReceive(nread, buf, addr, flags); + if (buf->base) + free(buf->base); + } +}; + +struct TcpSocket : public std::enable_shared_from_this +{ + using OnConnect = std::function& new_socket)>; + + TcpSocket(uv_loop_t* loop) { + uv_tcp_init(loop, &sock); + sock.data = nullptr; + } + void listen(in_port_t port, OnConnect cb) { + if (not sock.data) + sock.data = new std::shared_ptr(shared_from_this()); + connect_cb = cb; + struct sockaddr_in6 addr; + uv_ip6_addr("::0", port, &addr); + uv_tcp_bind(&sock, (const struct sockaddr*)&addr, 0); + if (int r = uv_listen((uv_stream_t*)&sock, 128, &TcpSocket::on_new_connection)) { + fprintf(stderr, "Listen error %s\n", uv_strerror(r)); + } + } + void accept(TcpSocket& server) { + if (not sock.data) + sock.data = new std::shared_ptr(shared_from_this()); + if (uv_accept((uv_stream_t*)&server.sock, (uv_stream_t*) &sock) == 0) { + fprintf(stderr, "uv_read_start\n"); + uv_read_start((uv_stream_t*)&sock, alloc_buffer, &TcpSocket::echo_read); + } else { + close(); + } + } + void close(OnClose onclose = {}) { + fprintf(stderr, "uv_close\n"); + close_cb = onclose; + uv_close((uv_handle_t*) &sock, &TcpSocket::on_close); + } +private: + uv_tcp_t sock; + OnConnect connect_cb; + OnClose close_cb; + void onNewConnection(int status) { + if (status < 0) { + fprintf(stderr, "New connection error %s\n", uv_strerror(status)); + return; + } + auto s = std::make_shared(sock.loop); + s->accept(*this); + connect_cb(s); + fprintf(stderr, "on_new_connection end\n"); + } + void onRead(ssize_t nread, const uv_buf_t *buf) { + if (nread < 0) { + if (nread != UV_EOF) { + fprintf(stderr, "Read error %s\n", uv_err_name(nread)); + } else { + fprintf(stderr, "EOF\n"); + } + close(); + } else if (nread > 0) { + uv_write_t *req = (uv_write_t *) malloc(sizeof(uv_write_t)); + uv_buf_t wrbuf = uv_buf_init(buf->base, nread); + fprintf(stderr, "uv_write\n"); + uv_write(req, (uv_stream_t*) &sock, &wrbuf, 1, &TcpSocket::echo_write); + } + } + void onClosed() { + if (close_cb) + close_cb(); + if (sock.data) { + delete static_cast*>(sock.data); + sock.data = nullptr; + } + } + static std::shared_ptr get(uv_handle_t* handle) { + return handle->data ? *static_cast*>(handle->data) : std::shared_ptr(); + } + static void on_new_connection(uv_stream_t* handle, int status) { + if (auto node = get((uv_handle_t*)handle)) + node->onNewConnection(status); + } + static void on_close(uv_handle_t* handle) { + if (auto node = get(handle)) + node->onClosed(); + } + static void echo_read(uv_stream_t* client, ssize_t nread, const uv_buf_t *buf) { + if (auto node = get((uv_handle_t*)client)) + node->onRead(nread, buf); + if (buf->base) + free(buf->base); + } + static void echo_write(uv_write_t *req, int status) { + if (status) { + fprintf(stderr, "Write error %s\n", uv_strerror(status)); + } + free(req); + } +}; + + +} diff --git a/python/opendht.pyx b/python/opendht.pyx index 87cb1c19..abc7bc2b 100644 --- a/python/opendht.pyx +++ b/python/opendht.pyx @@ -300,6 +300,8 @@ cdef class NodeSet(object): return s def __iter__(self): return NodeSetIter(self) + def __len__(self): + return self.size() cdef class PrivateKey(_WithID): cdef shared_ptr[cpp.PrivateKey] _key @@ -442,6 +444,10 @@ cdef class DhtConfig(object): self._config.dht_config.node_config.node_id = id._infohash def setNetwork(self, unsigned netid): self._config.dht_config.node_config.network = netid + def setPort(self, cpp.in_port_t port): + self._config.dht_config.node_config.network_config.bind_port = port + def setBindAddr(self, string addr): + self._config.dht_config.node_config.network_config.bind_addr = addr def setMaintainStorage(self, bool maintain_storage): self._config.dht_config.node_config.maintain_storage = maintain_storage @@ -484,12 +490,10 @@ cdef class DhtRunner(_WithID): def run(self, Identity id=None, is_bootstrap=False, cpp.in_port_t port=0, str ipv4="", str ipv6="", DhtConfig config=DhtConfig()): if id: config.setIdentity(id) - if ipv4 or ipv6: - bind4 = ipv4.encode() if ipv4 else b'' - bind6 = ipv6.encode() if ipv6 else b'' - self.thisptr.get().run(bind4, bind6, str(port).encode(), config._config) - else: - self.thisptr.get().run(port, config._config) + config.setPort(port) + if ipv6: + config.setBindAddr(ipv6) + self.thisptr.get().run(config._config) def join(self): self.thisptr.get().join() def shutdown(self, shutdown_cb=None): diff --git a/python/opendht_cpp.pxd b/python/opendht_cpp.pxd index 1ad79b56..c80c5a3f 100644 --- a/python/opendht_cpp.pxd +++ b/python/opendht_cpp.pxd @@ -198,11 +198,15 @@ cdef extern from "opendht/callbacks.h" namespace "dht": cdef DoneCallback bindDoneCb(DoneCallbackRaw cb, void *user_data) cdef DoneCallbackSimple bindDoneCbSimple(DoneCallbackSimpleRaw cb, void *user_data) + cppclass NetworkConfig: + string bind_addr + in_port_t bind_port cppclass Config: InfoHash node_id uint32_t network bool is_bootstrap bool maintain_storage + NetworkConfig network_config cppclass SecureDhtConfig: Config node_config Identity id @@ -219,8 +223,7 @@ cdef extern from "opendht/dhtrunner.h" namespace "dht": InfoHash getNodeId() const void bootstrap(const_char*, const_char*) void bootstrap(const SockAddr&, DoneCallbackSimple done_cb) - void run(in_port_t, Config config) - void run(const_char*, const_char*, const_char*, Config config) + void run(Config config) void join() void shutdown(ShutdownCallback) bool isRunning() diff --git a/python/setup.py.in b/python/setup.py.in index 1dd73655..59d05fa9 100644 --- a/python/setup.py.in +++ b/python/setup.py.in @@ -50,8 +50,8 @@ setup(name="opendht", ["@CURRENT_SOURCE_DIR@/opendht.pyx"], include_dirs = ['@PROJECT_SOURCE_DIR@/include'], language="c++", - extra_compile_args=["-std=c++11"], - extra_link_args=["-std=c++11"], + extra_compile_args=["-std=c++14"], + extra_link_args=["-std=c++14"], libraries=["opendht"], library_dirs = ['@CURRENT_BINARY_DIR@', '@PROJECT_BINARY_DIR@'] )) diff --git a/src/dht.cpp b/src/dht.cpp index 49e21a66..122003bb 100644 --- a/src/dht.cpp +++ b/src/dht.cpp @@ -62,8 +62,16 @@ Dht::getStatus(sa_family_t af) const void Dht::shutdown(ShutdownCallback cb) { + auto final_shutdown = [this,cb](){ + scheduler.stop(); + DHT_LOG.w("closing socket..."); + network_engine.close([=](){ + if (cb) + cb(); + }); + }; if (not maintain_storage) { - if (cb) cb(); + final_shutdown(); return; } @@ -73,7 +81,7 @@ Dht::shutdown(ShutdownCallback cb) auto str_donecb = [=](bool, const std::vector>&) { --*remaining; DHT_LOG.w("shuting down node: %u ops remaining", *remaining); - if (!*remaining && cb) { cb(); } + if (!*remaining) { final_shutdown(); } }; for (auto& str : store) @@ -81,8 +89,7 @@ Dht::shutdown(ShutdownCallback cb) if (!*remaining) { DHT_LOG.w("shuting down node: %u ops remaining", *remaining); - if (cb) - cb(); + final_shutdown(); } } @@ -241,8 +248,11 @@ Dht::onNewNode(const Sp& node, int confirm) if (confirm or not b->cached) b->cached = node; } else { + auto empty = b->nodes.empty(); /* Create a new node. */ b->nodes.emplace_front(node); + if (empty) + updateStatus(); } } @@ -767,7 +777,7 @@ Dht::search(const InfoHash& id, sa_family_t af, GetCallback gcb, QueryCallback q if (sr->nextSearchStep) scheduler.edit(sr->nextSearchStep, sr->getNextStepTime(scheduler.time())); else - sr->nextSearchStep = scheduler.add(scheduler.time(), std::bind(&Dht::searchStep, this, sr)); + sr->nextSearchStep = scheduler.run(std::bind(&Dht::searchStep, this, sr)); return sr; } @@ -1048,6 +1058,20 @@ bool callbackWrapper(Cb get_cb, return !op->status.ok; } + +void +Dht::bootstrap(const std::string& host, const std::string& service) +{ + for (const auto& bn : bootstrap_nodes) + if (bn.first == host and bn.second == service) + return; + bootstrap_nodes.emplace_back(host, service); + getAddrInfo(scheduler.getLoop(), host.c_str(), service.c_str(), [this](std::vector&& res) { + for (const auto& addr : res) + pingNode(addr); + }); +} + void Dht::get(const InfoHash& id, GetCallback getcb, DoneCallback donecb, Value::Filter&& filter, Where&& where) { @@ -1391,6 +1415,7 @@ Dht::expireStore() void Dht::connectivityChanged(sa_family_t af) { + updateStatus(); const auto& now = scheduler.time(); scheduler.edit(nextNodesConfirmation, now); auto& bucket_grow_time = (af == AF_INET) ? mybucket_grow_time : mybucket6_grow_time; @@ -1418,14 +1443,13 @@ Dht::rotateSecrets() secret = std::uniform_int_distribution{}(rdev); } uniform_duration_distribution<> time_dist(std::chrono::minutes(15), std::chrono::minutes(45)); - auto rotate_secrets_time = scheduler.time() + time_dist(rd); - scheduler.add(rotate_secrets_time, std::bind(&Dht::rotateSecrets, this)); + scheduler.add(time_dist(rd), std::bind(&Dht::rotateSecrets, this)); } Blob Dht::makeToken(const SockAddr& addr, bool old) const { - const void *ip; + const void* ip; size_t iplen; in_port_t port; @@ -1434,12 +1458,12 @@ Dht::makeToken(const SockAddr& addr, bool old) const const auto& sin = addr.getIPv4(); ip = &sin.sin_addr; iplen = 4; - port = sin.sin_port; + port = ntohs(sin.sin_port); } else if (family == AF_INET6) { const auto& sin6 = addr.getIPv6(); ip = &sin6.sin6_addr; iplen = 16; - port = sin6.sin6_port; + port = ntohs(sin6.sin6_port); } else { return {}; } @@ -1745,14 +1769,14 @@ Dht::~Dht() s.second->clear(); } -Dht::Dht() : store(), scheduler(DHT_LOG), network_engine(DHT_LOG, scheduler) {} +Dht::Dht() : store(), scheduler(nullptr, DHT_LOG), network_engine(nullptr, DHT_LOG, scheduler) {} -Dht::Dht(int s, int s6, Config config) +Dht::Dht(uv_loop_t* loop, Config config) : myid(config.node_id != zeroes ? config.node_id : InfoHash::getRandom()), - is_bootstrap(config.is_bootstrap), - maintain_storage(config.maintain_storage), store(), store_quota(), - scheduler(DHT_LOG), - network_engine(myid, config.network, s, s6, DHT_LOG, scheduler, + is_bootstrap(config.is_bootstrap), maintain_storage(config.maintain_storage), + buckets4{Bucket {AF_INET}},buckets6{Bucket {AF_INET6}}, store(), store_quota(), + scheduler(loop, DHT_LOG), + network_engine(loop, config.network_config, myid, config.network, DHT_LOG, scheduler, std::bind(&Dht::onError, this, _1, _2), std::bind(&Dht::onNewNode, this, _1, _2), std::bind(&Dht::onReportedAddr, this, _1, _2), @@ -1764,19 +1788,10 @@ Dht::Dht(int s, int s6, Config config) std::bind(&Dht::onRefresh, this, _1, _2, _3, _4)) { scheduler.syncTime(); - if (s < 0 && s6 < 0) - return; - if (s >= 0) - buckets4 = {Bucket {AF_INET}}; - if (s6 >= 0) - buckets6 = {Bucket {AF_INET6}}; - search_id = std::uniform_int_distribution{}(rd); uniform_duration_distribution<> time_dis {std::chrono::seconds(3), std::chrono::seconds(5)}; - auto confirm_nodes_time = scheduler.time() + time_dis(rd); - DHT_LOG.d(myid, "Scheduling %s", myid.toString().c_str()); - nextNodesConfirmation = scheduler.add(confirm_nodes_time, std::bind(&Dht::confirmNodes, this)); + nextNodesConfirmation = scheduler.add(time_dis(rd), std::bind(&Dht::confirmNodes, this)); // Fill old secret { @@ -1823,7 +1838,13 @@ Dht::neighbourhoodMaintenance(RoutingTable& list) n->toString().c_str(), id.toString().c_str()); /* Since our node-id is the same in both DHTs, it's probably profitable to query both families. */ - network_engine.sendFindNode(n, id, network_engine.want(), nullptr, nullptr); + network_engine.sendFindNode(n, id, network_engine.want(), nullptr, [this,n](const net::Request&, bool over) { + if (over) { + DHT_LOG.d(n->id, "[node %s] neighborhood maintenance op expired", n->toString().c_str()); + scheduler.edit(nextNodesConfirmation, Node::MAX_RESPONSE_TIME); + updateStatus(); + } + }); } return true; @@ -1876,10 +1897,9 @@ Dht::bucketMaintenance(RoutingTable& list) auto start = scheduler.time(); network_engine.sendFindNode(n, id, want, nullptr, [this,start,n](const net::Request&, bool over) { if (over) { - const auto& end = scheduler.time(); - using namespace std::chrono; - DHT_LOG.d(n->id, "[node %s] bucket maintenance op expired after %llu ms", n->toString().c_str(), duration_cast(end-start).count()); - scheduler.edit(nextNodesConfirmation, end + Node::MAX_RESPONSE_TIME); + DHT_LOG.d(n->id, "[node %s] bucket maintenance op expired", n->toString().c_str()); + scheduler.edit(nextNodesConfirmation, Node::MAX_RESPONSE_TIME); + updateStatus(); } }); sent = true; @@ -1952,45 +1972,34 @@ Dht::maintainStorage(decltype(store)::value_type& storage, bool force, DoneCallb } void -Dht::processMessage(const uint8_t *buf, size_t buflen, const SockAddr& from) +Dht::updateStatus() { - if (buflen == 0) - return; - - try { - network_engine.processMessage(buf, buflen, from); - } catch (const std::exception& e) { - DHT_LOG.e("Can't parse message from %s: %s", from.toString().c_str(), e.what()); - //auto code = e.getCode(); - //if (code == net::DhtProtocolException::INVALID_TID_SIZE or code == net::DhtProtocolException::WRONG_NODE_INFO_BUF_LEN) { - /* This is really annoying, as it means that we will - time-out all our searches that go through this node. - Kill it. */ - //const auto& id = e.getNodeId(); - //blacklistNode(&id, from, fromlen); - ///} + NodeStatus nstatus4 = getStatus(AF_INET); + NodeStatus nstatus6 = getStatus(AF_INET6); + 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. + DHT_LOG.e(myid, "DHT disconnected", myid.toString().c_str()); + } else { + //bootstrap_nodes.clear(); + } + if (statusCb) + statusCb(status4, status6); } } -time_point -Dht::periodic(const uint8_t *buf, size_t buflen, const SockAddr& from) -{ - scheduler.syncTime(); - processMessage(buf, buflen, from); - return scheduler.run(); -} - void Dht::expire() { uniform_duration_distribution<> time_dis(std::chrono::minutes(2), std::chrono::minutes(6)); - auto expire_stuff_time = scheduler.time() + duration(time_dis(rd)); expireBuckets(buckets4); expireBuckets(buckets6); expireStore(); expireSearches(); - scheduler.add(expire_stuff_time, std::bind(&Dht::expire, this)); + scheduler.add(time_dis(rd), std::bind(&Dht::expire, this)); } void @@ -2000,11 +2009,13 @@ Dht::confirmNodes() bool soon = false; const auto& now = scheduler.time(); - if (searches4.empty() and getStatus(AF_INET) == NodeStatus::Connected) { + updateStatus(); + + if (searches4.empty() and status4 == NodeStatus::Connected) { DHT_LOG.d(myid, "[confirm nodes] initial IPv4 'get' for my id (%s)", myid.toString().c_str()); search(myid, AF_INET); } - if (searches6.empty() and getStatus(AF_INET6) == NodeStatus::Connected) { + if (searches6.empty() and status6 == NodeStatus::Connected) { DHT_LOG.d(myid, "[confirm nodes] initial IPv6 'get' for my id (%s)", myid.toString().c_str()); search(myid, AF_INET6); } @@ -2026,9 +2037,7 @@ Dht::confirmNodes() auto time_dis = soon ? uniform_duration_distribution<> {seconds(5) , seconds(25)} : uniform_duration_distribution<> {seconds(60), seconds(180)}; - auto confirm_nodes_time = now + time_dis(rd); - - scheduler.edit(nextNodesConfirmation, confirm_nodes_time); + scheduler.edit(nextNodesConfirmation, time_dis(rd)); } std::vector @@ -2135,13 +2144,13 @@ Dht::insertNode(const InfoHash& id, const SockAddr& addr) } void -Dht::pingNode(const sockaddr* sa, socklen_t salen, DoneCallbackSimple&& cb) +Dht::pingNode(const SockAddr& sa, DoneCallbackSimple&& cb) { scheduler.syncTime(); - DHT_LOG.d("Sending ping to %s", print_addr(sa, salen).c_str()); - auto& count = sa->sa_family == AF_INET ? pending_pings4 : pending_pings6; + DHT_LOG.d("Sending ping to %s", sa.toString().c_str()); + auto& count = sa.getFamily() == AF_INET ? pending_pings4 : pending_pings6; count++; - network_engine.sendPing(sa, salen, [&count,cb](const net::Request&, net::NetworkEngine::RequestAnswer&&) { + network_engine.sendPing(sa, [&count,cb](const net::Request&, net::NetworkEngine::RequestAnswer&&) { count--; if (cb) cb(true); @@ -2311,7 +2320,7 @@ Dht::onListen(Sp node, const InfoHash& hash, const Blob& token, size_t soc net::DhtProtocolException::LISTEN_NO_INFOHASH }; } - if (!tokenMatch(token, node->addr)) { + if (not tokenMatch(token, node->addr)) { DHT_LOG.w(hash, node->id, "[node %s] incorrect token %s for 'listen'", node->toString().c_str(), hash.toString().c_str()); throw net::DhtProtocolException {net::DhtProtocolException::UNAUTHORIZED, net::DhtProtocolException::LISTEN_WRONG_TOKEN}; } @@ -2349,7 +2358,7 @@ Dht::onAnnounce(Sp node, net::DhtProtocolException::PUT_NO_INFOHASH }; } - if (!tokenMatch(token, node->addr)) { + if (not tokenMatch(token, node->addr)) { DHT_LOG.w(hash, node->id, "[node %s] incorrect token %s for 'put'", node->toString().c_str(), hash.toString().c_str()); throw net::DhtProtocolException {net::DhtProtocolException::UNAUTHORIZED, net::DhtProtocolException::PUT_WRONG_TOKEN}; } diff --git a/src/dhtrunner.cpp b/src/dhtrunner.cpp index d8903274..514821e1 100644 --- a/src/dhtrunner.cpp +++ b/src/dhtrunner.cpp @@ -20,98 +20,32 @@ #include "dhtrunner.h" #include "securedht.h" -#ifndef _WIN32 -#include -#else -#include -#endif - -#ifndef _WIN32 -#include -#else -#include -#include -#define close(x) closesocket(x) -#endif - namespace dht { constexpr std::chrono::seconds DhtRunner::BOOTSTRAP_PERIOD; DhtRunner::DhtRunner() : dht_() { -#ifdef _WIN32 - WSADATA wsd; - if (WSAStartup(MAKEWORD(2,2), &wsd) != 0) - throw DhtException("Can't initialize Winsock2"); -#endif + uv_async_.data = this; + uv_async_init(uv_loop_.get(), &uv_async_, &DhtRunner::loop_callback); } DhtRunner::~DhtRunner() { join(); -#ifdef _WIN32 - WSACleanup(); -#endif } void -DhtRunner::run(in_port_t port, DhtRunner::Config config) -{ - sockaddr_in sin4; - std::fill_n((uint8_t*)&sin4, sizeof(sin4), 0); - sin4.sin_family = AF_INET; - sin4.sin_port = htons(port); - sockaddr_in6 sin6; - std::fill_n((uint8_t*)&sin6, sizeof(sin6), 0); - sin6.sin6_family = AF_INET6; - sin6.sin6_port = htons(port); - run(&sin4, &sin6, config); -} - -void -DhtRunner::run(const char* ip4, const char* ip6, const char* service, DhtRunner::Config config) -{ - auto res4 = getAddrInfo(ip4, service); - auto res6 = getAddrInfo(ip6, service); - run(res4.empty() ? nullptr : (sockaddr_in*) &res4.front().first, - res6.empty() ? nullptr : (sockaddr_in6*)&res6.front().first, config); -} - -void -DhtRunner::run(const sockaddr_in* local4, const sockaddr_in6* local6, DhtRunner::Config config) +DhtRunner::run(DhtRunner::Config config) { if (running) return; - if (rcv_thread.joinable()) - rcv_thread.join(); running = true; - doRun(local4, local6, config.dht_config); + doRun(config.dht_config); if (not config.threaded) return; dht_thread = std::thread([this]() { - while (running) { - std::unique_lock lk(dht_mtx); - auto wakeup = loop_(); - cv.wait_until(lk, wakeup, [this]() { - if (not running) - return true; - { - std::lock_guard lck(sock_mtx); - if (not rcv.empty()) - return true; - } - { - std::lock_guard lck(storage_mtx); - if (not pending_ops_prio.empty()) - return true; - auto s = getStatus(); - if (not pending_ops.empty() and (s == NodeStatus::Connected or (s == NodeStatus::Disconnected and not bootstraping))) - return true; - } - return false; - }); - } + uv_loop_.run(); }); } @@ -121,21 +55,19 @@ DhtRunner::shutdown(ShutdownCallback cb) { pending_ops_prio.emplace([=](SecureDht& dht) mutable { dht.shutdown(cb); }); - cv.notify_all(); + uv_async_send(&uv_async_); } void DhtRunner::join() { running = false; - cv.notify_all(); - bootstrap_cv.notify_all(); + uv_async_send(&uv_async_); + //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(); + /*if (bootstrap_thread.joinable()) + bootstrap_thread.join();*/ { std::lock_guard lck(storage_mtx); @@ -145,10 +77,10 @@ DhtRunner::join() { std::lock_guard lck(dht_mtx); dht_.reset(); - status4 = NodeStatus::Disconnected; + /*status4 = NodeStatus::Disconnected; status6 = NodeStatus::Disconnected; bound4 = {}; - bound6 = {}; + bound6 = {};*/ } } @@ -175,6 +107,14 @@ DhtRunner::getNodeId() const return dht_->getNodeId(); } +void +DhtRunner::setOnStatusChanged(StatusCallback&& cb) +{ + std::lock_guard lck(dht_mtx); + if (not dht_) + throw std::runtime_error("dht is not running"); + dht_->setOnStatusChanged(std::move(cb)); +} std::pair DhtRunner::getStoreSize() const { @@ -307,6 +247,11 @@ DhtRunner::getPublicAddressStr(sa_family_t af) return ret; } +SockAddr +DhtRunner::getBound() const { + return dht_->getBoundAddr(); +} + void DhtRunner::registerCertificate(std::shared_ptr cert) { std::lock_guard lck(dht_mtx); @@ -318,42 +263,12 @@ DhtRunner::setLocalCertificateStore(CertificateStoreQuery&& query_method) { dht_->setLocalCertificateStore(std::forward(query_method)); } -time_point +void DhtRunner::loop_() { if (!dht_) - return {}; - - decltype(pending_ops) ops {}; - { - std::lock_guard lck(storage_mtx); - auto s = getStatus(); - ops = (pending_ops_prio.empty() && (s == NodeStatus::Connected or (s == NodeStatus::Disconnected and not bootstraping))) ? - std::move(pending_ops) : std::move(pending_ops_prio); - } - while (not ops.empty()) { - ops.front()(*dht_); - ops.pop(); - } - - time_point wakeup {}; - decltype(rcv) received {}; - { - std::lock_guard lck(sock_mtx); - // move to stack - received = std::move(rcv); - } - if (not received.empty()) { - for (const auto& pck : received) { - auto& buf = pck.first; - auto& from = pck.second; - wakeup = dht_->periodic(buf.data(), buf.size()-1, from); - } - received.clear(); - } else { - wakeup = dht_->periodic(nullptr, 0, nullptr, 0); - } - + return; +/* NodeStatus nstatus4 = dht_->getStatus(AF_INET); NodeStatus nstatus6 = dht_->getStatus(AF_INET6); if (nstatus4 != status4 || nstatus6 != status6) { @@ -371,102 +286,24 @@ DhtRunner::loop_() if (statusCb) statusCb(status4, status6); } - - return wakeup; +*/ + decltype(pending_ops) ops {}; + { + std::lock_guard lck(storage_mtx); + auto s = dht_->getStatus(); + ops = (pending_ops_prio.empty() && (s == NodeStatus::Connected or (s == NodeStatus::Disconnected and not bootstraping))) ? + std::move(pending_ops) : std::move(pending_ops_prio); + } + while (not ops.empty()) { + ops.front()(*dht_); + ops.pop(); + } } void -DhtRunner::doRun(const sockaddr_in* sin4, const sockaddr_in6* sin6, SecureDht::Config config) +DhtRunner::doRun(SecureDht::Config config) { - dht_.reset(); - - int s4 = -1, - s6 = -1; - - bound4 = {}; - if (sin4) { - s4 = socket(PF_INET, SOCK_DGRAM, 0); - if(s4 >= 0) { - int rc = bind(s4, (sockaddr*)sin4, sizeof(sockaddr_in)); - if(rc < 0) - throw DhtException("Can't bind IPv4 socket on " + dht::print_addr((sockaddr*)sin4, sizeof(sockaddr_in))); - bound4.second = sizeof(bound4.first); - getsockname(s4, (sockaddr*)&bound4.first, &bound4.second); - } - } - -#if 1 - bound6 = {}; - if (sin6) { - s6 = socket(PF_INET6, SOCK_DGRAM, 0); - if(s6 >= 0) { - int val = 1; - int rc = setsockopt(s6, IPPROTO_IPV6, IPV6_V6ONLY, (char *)&val, sizeof(val)); - if(rc < 0) - throw DhtException("Can't set IPV6_V6ONLY"); - - rc = bind(s6, (sockaddr*)sin6, sizeof(sockaddr_in6)); - if(rc < 0) - throw DhtException("Can't bind IPv6 socket on " + dht::print_addr((sockaddr*)sin6, sizeof(sockaddr_in6))); - bound6.second = sizeof(bound6.first); - getsockname(s6, (sockaddr*)&bound6.first, &bound6.second); - } - } -#endif - - dht_ = std::unique_ptr(new SecureDht {s4, s6, config}); - - rcv_thread = std::thread([this,s4,s6]() { - try { - while (true) { - struct timeval tv {/*.tv_sec = */0, /*.tv_usec = */250000}; - fd_set readfds; - - FD_ZERO(&readfds); - if(s4 >= 0) - FD_SET(s4, &readfds); - if(s6 >= 0) - FD_SET(s6, &readfds); - - int rc = select(s4 > s6 ? s4 + 1 : s6 + 1, &readfds, nullptr, nullptr, &tv); - if(rc < 0) { - if(errno != EINTR) { - perror("select"); - std::this_thread::sleep_for( std::chrono::seconds(1) ); - } - } - - if(!running) - break; - - if(rc > 0) { - std::array buf; - SockAddr from; - from.second = sizeof(from.first); - - if(s4 >= 0 && FD_ISSET(s4, &readfds)) - rc = recvfrom(s4, (char*)buf.data(), buf.size(), 0, (struct sockaddr*)&from.first, &from.second); - else if(s6 >= 0 && FD_ISSET(s6, &readfds)) - rc = recvfrom(s6, (char*)buf.data(), buf.size(), 0, (struct sockaddr*)&from.first, &from.second); - else - break; - if (rc > 0) { - { - std::lock_guard lck(sock_mtx); - rcv.emplace_back(Blob {buf.begin(), buf.begin()+rc+1}, from); - } - cv.notify_all(); - } - } - } - } catch (const std::exception& e) { - std::cerr << "Error in DHT networking thread: " << e.what() << std::endl; - } - if (s4 >= 0) - close(s4); - if (s6 >= 0) - close(s6); - }); + dht_ = std::unique_ptr(new SecureDht {uv_loop_.get(), config}); } void @@ -475,10 +312,11 @@ DhtRunner::get(InfoHash hash, GetCallback vcb, DoneCallback dcb, Value::Filter f { std::lock_guard lck(storage_mtx); pending_ops.emplace([=](SecureDht& dht) mutable { + std::cout << "DhtRunner dht.get " << hash << std::endl; dht.get(hash, vcb, dcb, std::move(f), std::move(w)); }); } - cv.notify_all(); + uv_async_send(&uv_async_); } void @@ -493,7 +331,7 @@ void DhtRunner::query(const InfoHash& hash, QueryCallback cb, DoneCallback done_ dht.query(hash, cb, done_cb, std::move(q)); }); } - cv.notify_all(); + uv_async_send(&uv_async_); } std::future @@ -506,7 +344,7 @@ DhtRunner::listen(InfoHash hash, GetCallback vcb, Value::Filter f, Where w) ret_token->set_value(dht.listen(hash, vcb, std::move(f), std::move(w))); }); } - cv.notify_all(); + uv_async_send(&uv_async_); return ret_token->get_future(); } @@ -525,7 +363,7 @@ DhtRunner::cancelListen(InfoHash h, size_t token) dht.cancelListen(h, token); }); } - cv.notify_all(); + uv_async_send(&uv_async_); } void @@ -538,7 +376,7 @@ DhtRunner::cancelListen(InfoHash h, std::shared_future token) dht.cancelListen(h, tk); }); } - cv.notify_all(); + uv_async_send(&uv_async_); } void @@ -551,7 +389,7 @@ DhtRunner::put(InfoHash hash, Value&& value, DoneCallback cb, time_point created dht.put(hash, sv, cb, created, permanent); }); } - cv.notify_all(); + uv_async_send(&uv_async_); } void @@ -563,7 +401,7 @@ DhtRunner::put(InfoHash hash, std::shared_ptr value, DoneCallback cb, tim dht.put(hash, value, cb, created, permanent); }); } - cv.notify_all(); + uv_async_send(&uv_async_); } void @@ -581,7 +419,7 @@ DhtRunner::cancelPut(const InfoHash& h , const Value::Id& id) dht.cancelPut(h, id); }); } - cv.notify_all(); + uv_async_send(&uv_async_); } void @@ -593,7 +431,7 @@ DhtRunner::putSigned(InfoHash hash, std::shared_ptr value, DoneCallback c dht.putSigned(hash, value, cb); }); } - cv.notify_all(); + uv_async_send(&uv_async_); } void @@ -617,7 +455,7 @@ DhtRunner::putEncrypted(InfoHash hash, InfoHash to, std::shared_ptr value dht.putEncrypted(hash, to, value, cb); }); } - cv.notify_all(); + uv_async_send(&uv_async_); } void @@ -632,6 +470,7 @@ DhtRunner::putEncrypted(const std::string& key, InfoHash to, Value&& value, Done putEncrypted(InfoHash::get(key), to, std::forward(value), cb); } +/* void DhtRunner::tryBootstrapContinuously() { @@ -689,7 +528,7 @@ DhtRunner::tryBootstrapContinuously() } } while (bootstraping); }); -} +}*/ std::vector> DhtRunner::getAddrInfo(const std::string& host, const std::string& service) @@ -719,17 +558,25 @@ DhtRunner::getAddrInfo(const std::string& host, const std::string& service) void DhtRunner::bootstrap(const std::string& host, const std::string& service) { - std::lock_guard lck(bootstrap_mtx); + /*std::lock_guard lck(bootstrap_mtx); bootstrap_nodes_all.emplace_back(host, service); bootstrap_nodes.emplace_back(host, service); - tryBootstrapContinuously(); + tryBootstrapContinuously();*/ + pending_ops_prio.emplace([=](SecureDht& dht) mutable { + dht.bootstrap(host, service); + }); + uv_async_send(&uv_async_); } void DhtRunner::clearBootstrap() { - std::lock_guard lck(bootstrap_mtx); - bootstrap_nodes_all.clear(); + //std::lock_guard lck(bootstrap_mtx); + //bootstrap_nodes_all.clear(); + pending_ops_prio.emplace([=](SecureDht& dht) mutable { + dht.clearBootstrap(); + }); + uv_async_send(&uv_async_); } void @@ -739,7 +586,7 @@ DhtRunner::bootstrap(const std::vector>& pending_ops_prio.emplace([=](SecureDht& dht) mutable { auto rem = cb ? std::make_shared>(nodes.size(), false) : nullptr; for (auto& node : nodes) - dht.pingNode((sockaddr*)&node.first, node.second, cb ? [rem,cb](bool ok) { + dht.pingNode(SockAddr((sockaddr*)&node.first, node.second), cb ? [rem,cb](bool ok) { auto& r = *rem; r.first--; r.second |= ok; @@ -747,7 +594,7 @@ DhtRunner::bootstrap(const std::vector>& cb(r.second); } : DoneCallbackSimple{}); }); - cv.notify_all(); + uv_async_send(&uv_async_); } void @@ -770,7 +617,7 @@ DhtRunner::bootstrap(const std::vector& nodes) dht.insertNode(node); }); } - cv.notify_all(); + uv_async_send(&uv_async_); } void @@ -782,7 +629,7 @@ DhtRunner::connectivityChanged() dht.connectivityChanged(); }); } - cv.notify_all(); + uv_async_send(&uv_async_); } void @@ -793,7 +640,7 @@ DhtRunner::findCertificate(InfoHash hash, std::function(loop)), scheduler(scheduler) { if (dht_socket >= 0) { if (!set_nonblocking(dht_socket, 1)) @@ -197,10 +197,11 @@ NetworkEngine::NetworkEngine(InfoHash& myid, NetId net, int s, int s6, Logger& l throw DhtException("Can't set socket to non-blocking mode"); } transaction_id = std::uniform_int_distribution{1}(rd_device); + using namespace std::placeholders; + sock->open(config.bind_addr.c_str(), config.bind_port, std::bind(&NetworkEngine::processMessage, this, _1, _2, _3)); } NetworkEngine::~NetworkEngine() { - clear(); } void @@ -218,18 +219,9 @@ NetworkEngine::tellListener(Sp node, uint32_t socket_id, const InfoHash& h } bool -NetworkEngine::isRunning(sa_family_t af) const +NetworkEngine::isRunning(sa_family_t /*af*/) const { - switch (af) { - case 0: - return dht_socket >= 0 || dht_socket6 >= 0; - case AF_INET: - return dht_socket >= 0; - case AF_INET6: - return dht_socket6 >= 0; - default: - return false; - } + return sock->isRunning(); } Sp @@ -294,7 +286,7 @@ NetworkEngine::requestStep(Sp sreq) req.on_expired(req, false); } - auto err = send((char*)req.msg.data(), req.msg.size(), + auto err = send(req.msg, req.msg.size(), (node.reply_time >= now - UDP_REPLY_TIME) ? 0 : MSG_CONFIRM, node.addr); if (err == ENETUNREACH || @@ -410,8 +402,9 @@ NetworkEngine::isNodeBlacklisted(const SockAddr& addr) const } void -NetworkEngine::processMessage(const uint8_t *buf, size_t buflen, const SockAddr& from) +NetworkEngine::processMessage(const uint8_t *buf, size_t buflen, const SockAddr& from_raw) { + SockAddr from = from_raw.isMappedIPv4() ? from_raw.getMappedIPv4() : from_raw; if (isMartian(from)) { DHT_LOG.w("Received packet from martian node %s", from.toString().c_str()); return; @@ -461,7 +454,7 @@ NetworkEngine::processMessage(const uint8_t *buf, size_t buflen, const SockAddr& process(std::move(pmsg_it->second.msg), from); partial_messages.erase(pmsg_it); } else - scheduler.add(now + RX_TIMEOUT, std::bind(&NetworkEngine::maintainRxBuffer, this, msg->tid)); + scheduler.add(RX_TIMEOUT, std::bind(&NetworkEngine::maintainRxBuffer, this, msg->tid)); } return; } @@ -490,8 +483,8 @@ NetworkEngine::processMessage(const uint8_t *buf, size_t buflen, const SockAddr& pmsg.last_part = now; auto wmsg = partial_messages.emplace(pmsg.msg->tid, std::move(pmsg)); if (wmsg.second) { - scheduler.add(now + RX_MAX_PACKET_TIME, std::bind(&NetworkEngine::maintainRxBuffer, this, wmsg.first->first)); - scheduler.add(now + RX_TIMEOUT, std::bind(&NetworkEngine::maintainRxBuffer, this, wmsg.first->first)); + scheduler.add(RX_MAX_PACKET_TIME, std::bind(&NetworkEngine::maintainRxBuffer, this, wmsg.first->first)); + scheduler.add(RX_TIMEOUT, std::bind(&NetworkEngine::maintainRxBuffer, this, wmsg.first->first)); } else DHT_LOG.e("Partial message with given TID already exists"); } @@ -508,8 +501,10 @@ NetworkEngine::process(std::unique_ptr&& msg, const SockAddr& fro auto rsocket_it = opened_sockets.end(); if (req_it == requests.end()) rsocket_it = opened_sockets.find(msg->tid); - if (req_it == requests.end() and rsocket_it == opened_sockets.end()) - throw DhtProtocolException {DhtProtocolException::UNKNOWN_TID, "Can't find transaction", msg->id}; + if (req_it == requests.end() and rsocket_it == opened_sockets.end()) { + DHT_LOG.e("Can't find transaction."); + return; + } auto req = req_it != requests.end() ? req_it->second : nullptr; auto rsocket = rsocket_it != opened_sockets.end() ? rsocket_it->second : nullptr; @@ -661,27 +656,23 @@ insertAddr(msgpack::packer& pk, const SockAddr& addr) } int -NetworkEngine::send(const char *buf, size_t len, int flags, const SockAddr& addr) +NetworkEngine::send(msgpack::sbuffer& msg, const SockAddr& addr) { if (addr.second == 0) return EFAULT; - int s; - if (addr.getFamily() == AF_INET) - s = dht_socket; - else if (addr.getFamily() == AF_INET6) - s = dht_socket6; - else - s = -1; + // move data + size_t size = msg.size(); + uint8_t* data = (uint8_t*)msg.release(); + return sock->send(data, size, addr); +} - if (s < 0) - return EAFNOSUPPORT; - if (sendto(s, buf, len, flags, (const sockaddr*)&addr.first, addr.second) == -1) { - int err = errno; - DHT_LOG.e("Can't send message to %s: %s", addr.toString().c_str(), strerror(err)); - return err; - } - return 0; +int +NetworkEngine::send(const Blob& msg, int /*flags*/, const SockAddr& addr) +{ + auto data = (uint8_t*)malloc(msg.size()); + memcpy(data, msg.data(), msg.size()); + return sock->send(data, msg.size(), addr); } Sp @@ -740,7 +731,7 @@ NetworkEngine::sendPong(const SockAddr& addr, TransId tid) { pk.pack(std::string("n")); pk.pack(network); } - send(buffer.data(), buffer.size(), 0, addr); + send(buffer, addr); } Sp @@ -945,7 +936,7 @@ NetworkEngine::sendValueParts(TransId tid, const std::vector& svals, const pk.pack(std::string("o")); pk.pack(start); pk.pack(std::string("d")); pk.pack_bin(end-start); pk.pack_bin_body((const char*)v.data()+start, end-start); - send(buffer.data(), buffer.size(), 0, addr); + send(buffer, addr); start = end; } while (start != v.size()); i++; @@ -1003,7 +994,7 @@ NetworkEngine::sendNodesValues(const SockAddr& addr, TransId tid, const Blob& no } // send response - send(buffer.data(), buffer.size(), 0, addr); + send(buffer, addr); // send parts if (not svals.empty()) @@ -1149,7 +1140,7 @@ NetworkEngine::sendListenConfirmation(const SockAddr& addr, TransId tid) { pk.pack(std::string("n")); pk.pack(network); } - send(buffer.data(), buffer.size(), 0, addr); + send(buffer, addr); } Sp @@ -1282,7 +1273,7 @@ NetworkEngine::sendValueAnnounced(const SockAddr& addr, TransId tid, Value::Id v pk.pack(std::string("n")); pk.pack(network); } - send(buffer.data(), buffer.size(), 0, addr); + send(buffer, addr); } void @@ -1313,7 +1304,7 @@ NetworkEngine::sendError(const SockAddr& addr, pk.pack(std::string("n")); pk.pack(network); } - send(buffer.data(), buffer.size(), 0, addr); + send(buffer, addr); } void diff --git a/src/securedht.cpp b/src/securedht.cpp index d2c36008..312e37de 100644 --- a/src/securedht.cpp +++ b/src/securedht.cpp @@ -40,10 +40,10 @@ Config& getConfig(SecureDht::Config& conf) return c; } -SecureDht::SecureDht(int s, int s6, SecureDht::Config conf) -: Dht(s, s6, getConfig(conf)), key_(conf.id.first), certificate_(conf.id.second) +SecureDht::SecureDht(uv_loop_t* loop, SecureDht::Config conf) +: Dht(loop, getConfig(conf)), key_(conf.id.first), certificate_(conf.id.second) { - if (s < 0 && s6 < 0) + if (not loop) return; for (const auto& type : DEFAULT_TYPES) diff --git a/src/utils.cpp b/src/utils.cpp index afc09f04..99c83ea1 100644 --- a/src/utils.cpp +++ b/src/utils.cpp @@ -19,12 +19,15 @@ #include "utils.h" #include "sockaddr.h" #include "default_types.h" +#include "uv_utils.h" /* An IPv4 equivalent to IN6_IS_ADDR_UNSPECIFIED */ #ifndef IN_IS_ADDR_UNSPECIFIED #define IN_IS_ADDR_UNSPECIFIED(a) (((long int) (a)->s_addr) == 0x00000000) #endif /* IN_IS_ADDR_UNSPECIFIED */ +static constexpr std::array MAPPED_IPV4_PREFIX {{0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0xff, 0xff}}; + namespace dht { std::string @@ -116,6 +119,57 @@ SockAddr::isPrivate() const } } +bool +SockAddr::isMappedIPv4() const +{ + if (getFamily() != AF_INET6) + return false; + const uint8_t* addr6 = reinterpret_cast(&getIPv6().sin6_addr); + return std::equal(MAPPED_IPV4_PREFIX.begin(), MAPPED_IPV4_PREFIX.end(), addr6); +} + +SockAddr +SockAddr::getMappedIPv4() const +{ + if (getFamily() != AF_INET6) + return *this; + SockAddr ret; + ret.setFamily(AF_INET); + ret.setPort(getPort()); + auto addr6 = reinterpret_cast(&getIPv6().sin6_addr); + auto addr4 = reinterpret_cast(&ret.getIPv4().sin_addr); + addr6 += MAPPED_IPV4_PREFIX.size(); + std::copy_n(addr6, sizeof(in_addr), addr4); + return ret; +} + +void get_addr_info_cb(uv_getaddrinfo_t* req, int status, struct addrinfo* res) { + if (auto cb = static_cast(req->data)) { + size_t count {0}; + for (auto addr = res; addr; addr = addr->ai_next) + count++; + std::vector ret; + ret.reserve(count); + for (auto addr = res; addr; addr = addr->ai_next) + ret.emplace_back(addr->ai_addr, addr->ai_addrlen); + (*cb)(std::move(ret)); + delete cb; + } + delete req; + uv_freeaddrinfo(res); +} + +void getAddrInfo(uv_loop_t* loop, const char* node, const char* service, GetAddrInfoCb&& cb) +{ + auto req = new uv_getaddrinfo_t; + req->data = new GetAddrInfoCb(std::move(cb)); + addrinfo hints {}; + hints.ai_flags = AI_PASSIVE | AI_ADDRCONFIG; + hints.ai_family = AF_UNSPEC; + hints.ai_socktype = SOCK_DGRAM; + uv_getaddrinfo(loop, req, get_addr_info_cb, node, service, &hints); +} + bool operator==(const SockAddr& a, const SockAddr& b) { return a.equals(b); } diff --git a/tools/dhtnode.cpp b/tools/dhtnode.cpp index 7e42c641..41809b61 100644 --- a/tools/dhtnode.cpp +++ b/tools/dhtnode.cpp @@ -39,7 +39,7 @@ void print_id_req() { } void print_node_info(const std::shared_ptr& dht, const dht_params& params) { - std::cout << "OpenDht node " << dht->getNodeId() << " running on port " << dht->getBoundPort() << std::endl; + std::cout << "OpenDht node " << dht->getNodeId() << " running on port " << dht->getBound().toString() << std::endl; if (params.generate_identity) std::cout << "Public key ID " << dht->getId() << std::endl; } @@ -370,6 +370,9 @@ main(int argc, char **argv) } dht->run(params.port, crt, true, params.network); + dht->setOnStatusChanged([](dht::NodeStatus status4, dht::NodeStatus status6){ + std::cout << "Node status changed : IPv4:" << (unsigned)status4 << " IPv6:" << (unsigned)status6 << std::endl; + }); if (params.log) { if (params.syslog or (params.daemonize and params.logfile.empty()))