diff --git a/CMakeLists.txt b/CMakeLists.txt index 0cbb6eb6..a8ff9c52 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -24,6 +24,7 @@ option (OPENDHT_DOCUMENTATION "Create and install the HTML based API documentati list (APPEND CMAKE_MODULE_PATH "${PROJECT_SOURCE_DIR}/cmake") find_package (Threads) 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) @@ -31,7 +32,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}) @@ -43,7 +43,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} -Wno-return-type -Wall -Wextra -Wnon-virtual-dtor -pedantic-errors -fvisibility=hidden") if (OPENDHT_SANITIZE) @@ -77,7 +77,10 @@ endif () if (Nettle_INCLUDE_DIRS) include_directories (SYSTEM "${Nettle_INCLUDE_DIRS}") endif () -link_directories (${Nettle_LIBRARY_DIRS}) +if (libuv_INCLUDE_DIRS) + include_directories (SYSTEM "${libuv_INCLUDE_DIRS}") +endif () +link_directories (${libuv_LIBRARY_DIRS} ${Nettle_LIBRARY_DIRS}) include_directories ( ./ include/ @@ -177,7 +180,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 ${CMAKE_THREAD_LIBS_INIT} ${GNUTLS_LIBRARIES} ${Nettle_LIBRARIES}) + target_link_libraries(opendht-static ${CMAKE_THREAD_LIBS_INIT} ${GNUTLS_LIBRARIES} ${Nettle_LIBRARIES} ${libuv_LIBRARIES}) install (TARGETS opendht-static DESTINATION ${CMAKE_INSTALL_LIBDIR} EXPORT opendht) endif () @@ -198,7 +201,7 @@ if (OPENDHT_SHARED) target_link_libraries(opendht PRIVATE ${argon2_LIBRARIES}) target_include_directories(opendht SYSTEM PRIVATE ${argon2_INCLUDE_DIRS}) endif () - target_link_libraries(opendht PRIVATE ${CMAKE_THREAD_LIBS_INIT} ${GNUTLS_LIBRARIES} ${Nettle_LIBRARIES}) + target_link_libraries(opendht PRIVATE ${CMAKE_THREAD_LIBS_INIT} ${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 9c8b38d6..08046df6 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 {}; net::NetworkEngine network_engine; unsigned pending_pings4 {0}; @@ -409,9 +414,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; diff --git a/include/opendht/dhtrunner.h b/include/opendht/dhtrunner.h index fd3346c0..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 getBound(f).getPort(); + 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 1fc5f020..12a195dc 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 @@ -202,8 +204,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, @@ -217,6 +219,9 @@ public: virtual ~NetworkEngine(); void clear(); + void close(OnClose cb) { + sock->close(cb); + } /** * Sends values (with closest nodes) to a listenner. @@ -236,7 +241,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; } void connectivityChanged(sa_family_t); @@ -266,8 +271,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)); } @@ -409,6 +414,10 @@ public: return cache.getCachedNodes(id, sa_f, count); } + SockAddr getBoundAddr() { + return sock->getBoundAddr(); + } + private: struct PartialMessage; @@ -463,7 +472,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, UdpSocket::OnSent&& cb); void sendValueParts(TransId tid, const std::vector& svals, const SockAddr& addr); std::vector packValueHeader(msgpack::sbuffer&, const std::vector>&); @@ -505,10 +515,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 {}; // global limiting should be triggered by at least 8 different IPs 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 3c251941..bb57a54a 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 7f5087d2..fa45005f 100644 --- a/include/opendht/sockaddr.h +++ b/include/opendht/sockaddr.h @@ -219,6 +219,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..ba07f069 --- /dev/null +++ b/include/opendht/uv_utils.h @@ -0,0 +1,324 @@ +#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; + using OnSent = 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 = std::move(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, OnSent&& cb = {}) { + auto req = new SendReq; + req->msg = data; + req->req.data = req; + req->cb = std::move(cb); + 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_storage addr; + int size = sizeof(addr); + uv_udp_getsockname(&sock, (sockaddr*)&addr, &size); + return {(const sockaddr*)&addr, (socklen_t)size}; + } +private: + struct SendReq { + uint8_t* msg; + uv_udp_send_t req; + OnSent cb; + }; + + 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 (req.cb) { + req.cb(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..329bac23 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): @@ -504,9 +508,9 @@ cdef class DhtRunner(_WithID): cpp.enableFileLogging(self.thisptr.get()[0], path.encode()) def isRunning(self): return self.thisptr.get().isRunning() - def getBound(self, cpp.sa_family_t af = 0): + def getBound(self): s = SockAddr() - s._addr = self.thisptr.get().getBound(af) + s._addr = self.thisptr.get().getBound() return s def getStorageLog(self): return self.thisptr.get().getStorageLog().decode() diff --git a/python/opendht_cpp.pxd b/python/opendht_cpp.pxd index 1ad79b56..7d17c419 100644 --- a/python/opendht_cpp.pxd +++ b/python/opendht_cpp.pxd @@ -73,7 +73,7 @@ cdef extern from "opendht/sockaddr.h" namespace "dht": SockAddr() except + string toString() const in_port_t getPort() const - void setPort(in_port_t p) + void setPort(in_port_t p) const sa_family_t getFamily() const void setFamily(sa_family_t f) @@ -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,12 +223,11 @@ 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() - SockAddr getBound(sa_family_t af) const + SockAddr getBound() const string getStorageLog() const string getRoutingTablesLog(sa_family_t af) const string getSearchesLog(sa_family_t af) const 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 b79f4ecf..69bab549 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(); } } @@ -700,7 +707,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; } @@ -961,6 +968,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) { @@ -1300,6 +1321,7 @@ Dht::expireStore() void Dht::connectivityChanged(sa_family_t af) { + updateStatus(); const auto& now = scheduler.time(); scheduler.edit(nextNodesConfirmation, now); buckets(af).connectivityChanged(now); @@ -1324,14 +1346,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; @@ -1340,12 +1361,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 {}; } @@ -1653,14 +1674,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), @@ -1672,23 +1693,12 @@ 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}}; - buckets4.is_client = config.is_bootstrap; - } - if (s6 >= 0) { - buckets6 = {Bucket {AF_INET6}}; - buckets6.is_client = config.is_bootstrap; - } - + buckets4.is_client = config.is_bootstrap; + buckets6.is_client = config.is_bootstrap; 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 { @@ -1735,7 +1745,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; @@ -1788,10 +1804,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; @@ -1863,7 +1878,7 @@ Dht::maintainStorage(decltype(store)::value_type& storage, bool force, DoneCallb return announce_per_af; } -time_point +/*time_point Dht::periodic(const uint8_t *buf, size_t buflen, const SockAddr& from) { scheduler.syncTime(); @@ -1875,19 +1890,37 @@ Dht::periodic(const uint8_t *buf, size_t buflen, const SockAddr& from) } } return scheduler.run(); +}*/ + +void +Dht::updateStatus() +{ + 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); + } } 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 @@ -1897,11 +1930,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); } @@ -1923,9 +1958,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 @@ -2032,13 +2065,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::RequestAnswer&&) { + network_engine.sendPing(sa, [&count,cb](const net::Request&, net::RequestAnswer&&) { count--; if (cb) cb(true); @@ -2207,7 +2240,7 @@ Dht::onListen(Sp node, const InfoHash& hash, const Blob& token, size_t soc net::DhtProtocolException::LISTEN_NO_INFOHASH }; } - if (!tokenMatch(token, node->getAddr())) { + if (not tokenMatch(token, node->getAddr())) { 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}; } @@ -2246,7 +2279,7 @@ Dht::onAnnounce(Sp n, net::DhtProtocolException::PUT_NO_INFOHASH }; } - if (!tokenMatch(token, node.getAddr())) { + if (not tokenMatch(token, node.getAddr())) { 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 66b4ebbd..3af6b734 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,106 +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))); - sockaddr_storage ss; - socklen_t ss_len = sizeof(ss); - getsockname(s4, (sockaddr*)&ss, &ss_len); - bound4 = {ss, ss_len}; - } - } - -#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))); - sockaddr_storage ss; - socklen_t ss_len = sizeof(ss); - getsockname(s6, (sockaddr*)&ss, &ss_len); - bound6 = {ss, ss_len}; - } - } -#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_storage from; - socklen_t from_len = sizeof(from); - - if(s4 >= 0 && FD_ISSET(s4, &readfds)) - rc = recvfrom(s4, (char*)buf.data(), buf.size(), 0, (sockaddr*)&from, &from_len); - else if(s6 >= 0 && FD_ISSET(s6, &readfds)) - rc = recvfrom(s6, (char*)buf.data(), buf.size(), 0, (sockaddr*)&from, &from_len); - else - break; - if (rc > 0) { - { - std::lock_guard lck(sock_mtx); - rcv.emplace_back(Blob {buf.begin(), buf.begin()+rc+1}, SockAddr(from, from_len)); - } - 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 @@ -479,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 @@ -497,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 @@ -510,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(); } @@ -529,7 +363,7 @@ DhtRunner::cancelListen(InfoHash h, size_t token) dht.cancelListen(h, token); }); } - cv.notify_all(); + uv_async_send(&uv_async_); } void @@ -542,7 +376,7 @@ DhtRunner::cancelListen(InfoHash h, std::shared_future token) dht.cancelListen(h, tk); }); } - cv.notify_all(); + uv_async_send(&uv_async_); } void @@ -555,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 @@ -567,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 @@ -585,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 @@ -597,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 @@ -621,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 @@ -636,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() { @@ -693,7 +528,7 @@ DhtRunner::tryBootstrapContinuously() } } while (bootstraping); }); -} +}*/ std::vector> DhtRunner::getAddrInfo(const std::string& host, const std::string& service) @@ -723,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 @@ -743,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; @@ -751,7 +594,7 @@ DhtRunner::bootstrap(const std::vector>& cb(r.second); } : DoneCallbackSimple{}); }); - cv.notify_all(); + uv_async_send(&uv_async_); } void @@ -759,9 +602,9 @@ DhtRunner::bootstrap(const SockAddr& addr, DoneCallbackSimple&& cb) { std::lock_guard lck(storage_mtx); pending_ops_prio.emplace([addr,cb](SecureDht& dht) mutable { - dht.pingNode(addr.get(), addr.getLength(), std::move(cb)); + dht.pingNode(addr, std::move(cb)); }); - cv.notify_all(); + uv_async_send(&uv_async_); } void @@ -774,7 +617,7 @@ DhtRunner::bootstrap(const std::vector& nodes) dht.insertNode(node); }); } - cv.notify_all(); + uv_async_send(&uv_async_); } void @@ -786,7 +629,7 @@ DhtRunner::connectivityChanged() dht.connectivityChanged(); }); } - cv.notify_all(); + uv_async_send(&uv_async_); } void @@ -797,7 +640,7 @@ DhtRunner::findCertificate(InfoHash hash, std::function= 0; -} - -#endif - struct NetworkEngine::PartialMessage { SockAddr from; time_point start; @@ -116,8 +90,8 @@ RequestAnswer::RequestAnswer(ParsedMessage&& msg) : ntoken(std::move(msg.token)), values(std::move(msg.values)), fields(std::move(msg.fields)), nodes4(std::move(msg.nodes4)), nodes6(std::move(msg.nodes6)) {} -NetworkEngine::NetworkEngine(Logger& log, Scheduler& scheduler) : myid(zeroes), DHT_LOG(log), scheduler(scheduler) {} -NetworkEngine::NetworkEngine(InfoHash& myid, NetId net, int s, int s6, Logger& log, Scheduler& scheduler, +NetworkEngine::NetworkEngine(uv_loop_t*, Logger& log, Scheduler& scheduler) : myid(zeroes), DHT_LOG(log), scheduler(scheduler) {} +NetworkEngine::NetworkEngine(uv_loop_t* loop, const NetworkConfig& config, InfoHash& myid, NetId net, Logger& log, Scheduler& scheduler, decltype(NetworkEngine::onError) onError, decltype(NetworkEngine::onNewNode) onNewNode, decltype(NetworkEngine::onReportedAddr) onReportedAddr, @@ -129,20 +103,13 @@ NetworkEngine::NetworkEngine(InfoHash& myid, NetId net, int s, int s6, Logger& l decltype(NetworkEngine::onRefresh) onRefresh) : onError(onError), onNewNode(onNewNode), onReportedAddr(onReportedAddr), onPing(onPing), onFindNode(onFindNode), onGetValues(onGetValues), onListen(onListen), onAnnounce(onAnnounce), onRefresh(onRefresh), myid(myid), - network(net), dht_socket(s), dht_socket6(s6), DHT_LOG(log), scheduler(scheduler) + network(net), DHT_LOG(log), sock(std::make_shared(loop)), scheduler(scheduler) { - if (dht_socket >= 0) { - if (!set_nonblocking(dht_socket, 1)) - throw DhtException("Can't set socket to non-blocking mode"); - } - if (dht_socket6 >= 0) { - if (!set_nonblocking(dht_socket6, 1)) - throw DhtException("Can't set socket to non-blocking mode"); - } + 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 @@ -160,18 +127,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(); } void @@ -203,27 +161,27 @@ NetworkEngine::requestStep(Sp sreq) req.on_expired(req, false); } - auto err = send((char*)req.msg.data(), req.msg.size(), - (node.getReplyTime() >= now - UDP_REPLY_TIME) ? 0 : MSG_CONFIRM, - node.getAddr()); - if (err == ENETUNREACH || - err == EHOSTUNREACH || - err == EAFNOSUPPORT) - { - node.setExpired(); - if (node.id == zeroes) - requests.erase(req.tid); - } else { - if (err != EAGAIN) { - ++req.attempt_count; + std::weak_ptr wreq = sreq; + auto err = send(req.msg, 0, node.getAddr(), [this,wreq](int status) { + if (auto req = wreq.lock()) { + if (status >= 0) { + ++req->attempt_count; + return; + } + if (status == -EAGAIN) + return; + auto& node = *req->node; + DHT_LOG.e(node.id, "[node %s] can't send packet: %d %s", node.toString().c_str(), status, uv_strerror(status)); + node.setExpired(); + if (node.id == zeroes) + requests.erase(req->tid); } - req.last_try = now; - std::weak_ptr wreq = sreq; - scheduler.add(req.last_try + Node::MAX_RESPONSE_TIME, [this,wreq] { - if (auto req = wreq.lock()) - requestStep(req); - }); - } + }); + req.last_try = now; + scheduler.add(req.last_try + Node::MAX_RESPONSE_TIME, [this,wreq] { + if (auto req = wreq.lock()) + requestStep(req); + }); } /** @@ -307,8 +265,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; @@ -358,7 +317,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; } @@ -387,8 +346,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"); } @@ -415,7 +374,8 @@ NetworkEngine::process(std::unique_ptr&& msg, const SockAddr& fro node->received(now, req); if (not node->isClient()) onNewNode(node, 1); - throw DhtProtocolException {DhtProtocolException::UNKNOWN_TID, "Can't find transaction", msg->id}; + DHT_LOG.e(node->id, "[node %s] Can't find transaction", node->toString().c_str()); + return; } } @@ -548,27 +508,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 (not addr) 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, addr.get(), addr.getLength()) == -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, UdpSocket::OnSent&& cb) +{ + auto data = (uint8_t*)malloc(msg.size()); + memcpy(data, msg.data(), msg.size()); + return sock->send(data, msg.size(), addr, std::move(cb)); } Sp @@ -627,7 +583,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 @@ -829,7 +785,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++; @@ -887,7 +843,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()) @@ -1029,7 +985,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 @@ -1162,7 +1118,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 @@ -1193,7 +1149,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/search.h b/src/search.h index e5aeee82..a7828dc3 100644 --- a/src/search.h +++ b/src/search.h @@ -320,7 +320,7 @@ struct Dht::Search { uint16_t tid; time_point refill_time {time_point::min()}; time_point step_time {time_point::min()}; /* the time of the last search step */ - Sp nextSearchStep {}; + Sp nextSearchStep {}; bool expired {false}; /* no node, or all nodes expired */ bool done {false}; /* search is over, cached for later */ diff --git a/src/securedht.cpp b/src/securedht.cpp index ea15e125..19ff12a7 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 cb2fd5b3..ea8a2c77 100644 --- a/src/utils.cpp +++ b/src/utils.cpp @@ -20,12 +20,15 @@ #include "sockaddr.h" #include "default_types.h" #include "net.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 @@ -112,6 +115,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()))