diff --git a/include/opendht/callbacks.h b/include/opendht/callbacks.h index 6738c307..bbf3a2f1 100644 --- a/include/opendht/callbacks.h +++ b/include/opendht/callbacks.h @@ -89,23 +89,23 @@ struct OPENDHT_PUBLIC NodeInfo { */ struct OPENDHT_PUBLIC Config { /** DHT node ID */ - InfoHash node_id; + InfoHash node_id {}; /** * DHT network ID. A node will only talk with other nodes having * the same network ID. * Network ID 0 (default) represents the main public network. */ - NetId network; + NetId network {0}; /** For testing purposes only, enables bootstrap mode */ - bool is_bootstrap; + bool is_bootstrap {false}; /** Makes the DHT responsible to maintain its stored values. Consumes more ressources. */ - bool maintain_storage; + bool maintain_storage {false}; /** If set, the dht will load its state from this file on start and save its state in this file on shutdown */ - std::string persist_path; + std::string persist_path {}; }; /** @@ -113,8 +113,8 @@ struct OPENDHT_PUBLIC Config { */ struct OPENDHT_PUBLIC SecureDhtConfig { - Config node_config; - crypto::Identity id; + Config node_config {}; + crypto::Identity id {}; }; static constexpr size_t DEFAULT_STORAGE_LIMIT {1024 * 1024 * 64}; diff --git a/include/opendht/dht.h b/include/opendht/dht.h index 4519a1de..37d9b07d 100644 --- a/include/opendht/dht.h +++ b/include/opendht/dht.h @@ -122,7 +122,7 @@ public: void pingNode(const sockaddr*, socklen_t, DoneCallbackSimple&& cb={}) override; - time_point periodic(const uint8_t *buf, size_t buflen, const SockAddr&) override; + time_point periodic(const uint8_t *buf, size_t buflen, SockAddr) override; time_point periodic(const uint8_t *buf, size_t buflen, const sockaddr* from, socklen_t fromlen) override { return periodic(buf, buflen, SockAddr(from, fromlen)); } diff --git a/include/opendht/dht_interface.h b/include/opendht/dht_interface.h index 0cfdf3e1..ff30a635 100644 --- a/include/opendht/dht_interface.h +++ b/include/opendht/dht_interface.h @@ -79,7 +79,7 @@ public: virtual void pingNode(const sockaddr*, socklen_t, DoneCallbackSimple&& cb={}) = 0; - virtual time_point periodic(const uint8_t *buf, size_t buflen, const SockAddr&) = 0; + virtual time_point periodic(const uint8_t *buf, size_t buflen, SockAddr) = 0; virtual time_point periodic(const uint8_t *buf, size_t buflen, const sockaddr* from, socklen_t fromlen) = 0; /** diff --git a/include/opendht/dht_proxy_client.h b/include/opendht/dht_proxy_client.h index ec87f465..52f28231 100644 --- a/include/opendht/dht_proxy_client.h +++ b/include/opendht/dht_proxy_client.h @@ -196,7 +196,7 @@ public: */ void pushNotificationReceived(const std::map& notification); - time_point periodic(const uint8_t*, size_t, const SockAddr&); + time_point periodic(const uint8_t*, size_t, SockAddr); time_point periodic(const uint8_t* buf, size_t buflen, const sockaddr* from, socklen_t fromlen) { return periodic(buf, buflen, SockAddr(from, fromlen)); } @@ -304,7 +304,7 @@ private: * Send Listen with httpClient_ */ void sendListen(const restinio::http_request_header_t header, - const ValueCallback &cb, const Value::Filter &filter, + const ValueCallback &cb, const Sp &state, Listener &listener, ListenMethod method = ListenMethod::LISTEN); diff --git a/include/opendht/dhtrunner.h b/include/opendht/dhtrunner.h index 1b8479b5..3a180ec4 100644 --- a/include/opendht/dhtrunner.h +++ b/include/opendht/dhtrunner.h @@ -56,12 +56,12 @@ public: using StatusCallback = std::function; struct Config { - SecureDhtConfig dht_config; - bool threaded; - std::string proxy_server; - std::string push_node_id; - bool peer_discovery; - bool peer_publish; + SecureDhtConfig dht_config {}; + bool threaded {true}; + std::string proxy_server {}; + std::string push_node_id {}; + bool peer_discovery {false}; + bool peer_publish {false}; }; struct Context { @@ -365,24 +365,12 @@ public: * @param threaded: If false, ::loop() must be called periodically. Otherwise a thread is launched. * @param cb: Optional callback to receive general state information. */ - void run(in_port_t port = 4222, const crypto::Identity identity = {}, bool threaded = true, NetId network = 0) { - run(port, { - /*.dht_config = */{ - /*.node_config = */{ - /*.node_id = */{}, - /*.network = */network, - /*.is_bootstrap = */false, - /*.maintain_storage = */false, - /*.persist_path = */{} - }, - /*.id = */identity - }, - /*.threaded = */threaded, - /*.proxy_server = */"", - /*.push_node_id = */"", - /*.peer_discovery = */true, - /*.peer_publish = */true, - }); + void run(in_port_t port = 4222, const crypto::Identity& identity = {}, bool threaded = true, NetId network = 0) { + Config config; + config.dht_config.node_config.network = network; + config.dht_config.id = identity; + config.threaded = threaded; + run(port, config); } void run(in_port_t port, const Config& config, Context&& context = {}); diff --git a/include/opendht/network_engine.h b/include/opendht/network_engine.h index f3da45f4..1ec4d6fe 100644 --- a/include/opendht/network_engine.h +++ b/include/opendht/network_engine.h @@ -397,7 +397,7 @@ public: * @param fromlen The length of the corresponding sockaddr structure. * @param now The time to adjust the clock in the network engine. */ - void processMessage(const uint8_t *buf, size_t buflen, const SockAddr& addr); + void processMessage(const uint8_t *buf, size_t buflen, SockAddr addr); Sp insertNode(const InfoHash& myid, const SockAddr& addr) { auto n = cache.getNode(myid, addr, scheduler.time(), 0); diff --git a/include/opendht/securedht.h b/include/opendht/securedht.h index 1274825a..2682ce0e 100644 --- a/include/opendht/securedht.h +++ b/include/opendht/securedht.h @@ -195,8 +195,8 @@ public: std::vector getPublicAddress(sa_family_t family = 0) override { return dht_->getPublicAddress(family); } - time_point periodic(const uint8_t *buf, size_t buflen, const SockAddr& sa) override { - return dht_->periodic(buf, buflen, sa); + time_point periodic(const uint8_t *buf, size_t buflen, SockAddr sa) override { + return dht_->periodic(buf, buflen, std::move(sa)); } time_point periodic(const uint8_t *buf, size_t buflen, const sockaddr* from, socklen_t fromlen) override { return dht_->periodic(buf, buflen, from, fromlen); diff --git a/include/opendht/sockaddr.h b/include/opendht/sockaddr.h index 143a3981..57928478 100644 --- a/include/opendht/sockaddr.h +++ b/include/opendht/sockaddr.h @@ -59,7 +59,7 @@ public: SockAddr(const SockAddr& o) { set(o.get(), o.getLength()); } - SockAddr(SockAddr&& o) noexcept : len(o.len), addr(std::move(o.addr)) { + SockAddr(SockAddr&& o) noexcept : addr(std::move(o.addr)), len(o.len) { o.len = 0; } @@ -248,7 +248,8 @@ public: bool isUnspecified() const; bool isMappedIPv4() const; - SockAddr getMappedIPv4() const; + SockAddr getMappedIPv4(); + SockAddr getMappedIPv6(); /** * A comparator to classify IP addresses, only considering the @@ -279,9 +280,9 @@ public: } }; private: - socklen_t len {0}; struct free_delete { void operator()(void* p) { ::free(p); } }; std::unique_ptr addr {}; + socklen_t len {0}; void set(const sockaddr* sa, socklen_t length) { if (len != length) { diff --git a/include/opendht/thread_pool.h b/include/opendht/thread_pool.h index 3c0f932d..c8486fd7 100644 --- a/include/opendht/thread_pool.h +++ b/include/opendht/thread_pool.h @@ -49,8 +49,8 @@ public: return ret->get_future(); } template - std::shared_ptr> getShared(std::function&& cb) { - return std::make_shared>(get(std::move(cb))); + std::shared_future getShared(std::function&& cb) { + return get(std::move(cb)); } void stop(); diff --git a/src/dht.cpp b/src/dht.cpp index 90e2eec5..da431f19 100644 --- a/src/dht.cpp +++ b/src/dht.cpp @@ -1901,14 +1901,14 @@ Dht::maintainStorage(decltype(store)::value_type& storage, bool force, const Don } time_point -Dht::periodic(const uint8_t *buf, size_t buflen, const SockAddr& from) +Dht::periodic(const uint8_t *buf, size_t buflen, SockAddr from) { scheduler.syncTime(); if (buflen) { try { - network_engine.processMessage(buf, buflen, from); + network_engine.processMessage(buf, buflen, std::move(from)); } catch (const std::exception& e) { - DHT_LOG.e("Can't process message from %s: %s", from.toString().c_str(), e.what()); + DHT_LOG.e("Can't process message: %s", e.what()); } } return scheduler.run(); diff --git a/src/dht_proxy_client.cpp b/src/dht_proxy_client.cpp index bd567749..7cbaf0e8 100644 --- a/src/dht_proxy_client.cpp +++ b/src/dht_proxy_client.cpp @@ -37,15 +37,14 @@ struct DhtProxyClient::ListenState { struct DhtProxyClient::Listener { - Listener(OpValueCache&& c, Value::Filter&& f): - cache(std::move(c)), filter(std::move(f)) + Listener(OpValueCache&& c): + cache(std::move(c)) {} uint16_t connId {0}; unsigned callbackId; OpValueCache cache; ValueCallback cb; - Value::Filter filter; Sp state; Sp refreshTimer; @@ -233,7 +232,7 @@ DhtProxyClient::isRunning(sa_family_t af) const } time_point -DhtProxyClient::periodic(const uint8_t*, size_t, const SockAddr&) +DhtProxyClient::periodic(const uint8_t*, size_t, SockAddr) { // Exec all currently stored callbacks decltype(callbacks_) callbacks; @@ -776,10 +775,11 @@ DhtProxyClient::listen(const InfoHash& key, ValueCallback cb, Value::Filter filt logger_->d("[proxy:client] [listen] [search %s]", key.to_c_str()); auto& search = searches_[key]; - auto query = std::make_shared(Select{}, where); - auto token = search.ops.listen(cb, query, filter, [this, key, filter]( + auto query = std::make_shared(Select{}, std::move(where)); + auto token = search.ops.listen(cb, query, filter, [this, key]( Sp, ValueCallback cb, SyncCallback) -> size_t { std::lock_guard lock(searchLock_); + // Find search auto search = searches_.find(key); if (search == searches_.end()) { if (logger_) @@ -790,18 +790,19 @@ DhtProxyClient::listen(const InfoHash& key, ValueCallback cb, Value::Filter filt logger_->d("[proxy:client] [listen] [search %s] sending %s", key.to_c_str(), deviceKey_.empty() ? "listen" : "subscribe"); + // Add listener auto token = ++listenerToken_; auto l = search->second.listeners.find(token); if (l == search->second.listeners.end()) { - auto f = filter; l = search->second.listeners.emplace(std::piecewise_construct, std::forward_as_tuple(token), - std::forward_as_tuple(std::move(cb), std::move(f))).first; + std::forward_as_tuple(std::move(cb))).first; } else { if (l->second.state) l->second.state->cancel = true; } + // Add callback auto state = std::make_shared(); l->second.state = state; l->second.cb = [this,key,token,state](const std::vector>& values, bool expired) { @@ -809,14 +810,13 @@ DhtProxyClient::listen(const InfoHash& key, ValueCallback cb, Value::Filter filt return false; std::lock_guard lock(searchLock_); auto s = searches_.find(key); - if (s == searches_.end()) { - return false; + if (s != searches_.end()) { + auto l = s->second.listeners.find(token); + if (l != s->second.listeners.end()) { + return l->second.cache.onValue(values, expired); + } } - auto l = s->second.listeners.find(token); - if (l == s->second.listeners.end()) { - return false; - } - return l->second.cache.onValue(values, expired); + return false; }; auto vcb = l->second.cb; @@ -862,7 +862,7 @@ DhtProxyClient::listen(const InfoHash& key, ValueCallback cb, Value::Filter filt header.method(restinio::http_method_subscribe()); header.request_target("/" + key.toString()); } - sendListen(header, vcb, filter, state, l->second, method); + sendListen(header, vcb, state, l->second, method); return token; }); return token; @@ -912,7 +912,7 @@ DhtProxyClient::cancelListen(const InfoHash& key, size_t gtoken) { void DhtProxyClient::sendListen(const restinio::http_request_header_t header, - const ValueCallback& cb, const Value::Filter& filter, + const ValueCallback& cb, const Sp& state, Listener& listener, ListenMethod method) { @@ -943,7 +943,6 @@ DhtProxyClient::sendListen(const restinio::http_request_header_t header, return cb(values, expired); }; context->state = state; - context->filter = filter; auto parser = std::make_shared(); http_parser_init(parser.get(), HTTP_RESPONSE); @@ -978,7 +977,7 @@ DhtProxyClient::sendListen(const restinio::http_request_header_t header, } auto value = std::make_shared(json); auto expired = json.get("expired", Json::Value(false)).asBool(); - if ((not context->filter or context->filter(*value)) and context->cb){ + if (context->cb){ context->cb({value}, expired); } } catch(const std::exception& e) { @@ -1167,7 +1166,6 @@ DhtProxyClient::restartListeners() // Redo listen state->cancel = false; state->ok = true; - auto filter = listener.filter; auto cb = listener.cb; // define header restinio::http_request_header_t header; @@ -1175,7 +1173,7 @@ DhtProxyClient::restartListeners() restinio::method_listen.raw_id())); header.request_target("/" + search.first.toString()); // send listen - sendListen(header, cb, filter, state, listener, ListenMethod::LISTEN); + sendListen(header, cb, state, listener, ListenMethod::LISTEN); } } } @@ -1225,7 +1223,6 @@ DhtProxyClient::pushNotificationReceived(const std::map>& vals) { return cb(vals, false); @@ -1233,7 +1230,7 @@ DhtProxyClient::pushNotificationReceived(const std::mapsecond); std::vector ids; @@ -1302,8 +1299,7 @@ DhtProxyClient::resubscribe(const InfoHash& key, Listener& listener) listener.refreshTimer->expires_at(std::chrono::steady_clock::now() + proxy::OP_TIMEOUT - proxy::OP_MARGIN); auto vcb = listener.cb; - auto filter = listener.filter; - sendListen(header, vcb, filter, state, listener, ListenMethod::RESUBSCRIBE); + sendListen(header, vcb, state, listener, ListenMethod::RESUBSCRIBE); #else (void) key; (void) listener; diff --git a/src/dhtrunner.cpp b/src/dhtrunner.cpp index 8d9deb8d..9a7d4318 100644 --- a/src/dhtrunner.cpp +++ b/src/dhtrunner.cpp @@ -223,12 +223,12 @@ DhtRunner::shutdown(ShutdownCallback cb) { cb(); return; } -#ifdef OPENDHT_PROXY_CLIENT - if (dht_via_proxy_) - dht_via_proxy_->shutdown(cb); -#endif std::lock_guard lck(storage_mtx); - pending_ops_prio.emplace([=](SecureDht& dht) mutable { + pending_ops_prio.emplace([=](SecureDht&) mutable { +#ifdef OPENDHT_PROXY_CLIENT + if (dht_via_proxy_) + dht_via_proxy_->shutdown(cb); +#endif if (dht_) dht_->shutdown(cb); }); @@ -528,7 +528,7 @@ DhtRunner::loop_() if (clock::now() - pck->received > RX_QUEUE_MAX_DELAY) dropped++; else - wakeup = dht->periodic(pck->data.data(), pck->data.size(), pck->from); + wakeup = dht->periodic(pck->data.data(), pck->data.size(), std::move(pck->from)); received.pop(); } } else { @@ -943,6 +943,7 @@ void DhtRunner::setProxyServer(const std::string& proxy, const std::string& pushNodeId) { #ifdef OPENDHT_PROXY_CLIENT + std::lock_guard lck(dht_mtx); if (config_.proxy_server == proxy and config_.push_node_id == pushNodeId) return; config_.proxy_server = proxy; @@ -1008,6 +1009,7 @@ DhtRunner::enableProxy(bool proxify) void DhtRunner::forwardAllMessages(bool forward) { + std::lock_guard lck(dht_mtx); #ifdef OPENDHT_PROXY_SERVER #ifdef OPENDHT_PROXY_CLIENT if (dht_via_proxy_) @@ -1025,6 +1027,7 @@ DhtRunner::forwardAllMessages(bool forward) */ void DhtRunner::setPushNotificationToken(const std::string& token) { + std::lock_guard lck(dht_mtx); #if defined(OPENDHT_PROXY_CLIENT) && defined(OPENDHT_PUSH_NOTIFICATIONS) pushToken_ = token; if (dht_via_proxy_) diff --git a/src/network_engine.cpp b/src/network_engine.cpp index 269dafc1..b7736d3d 100644 --- a/src/network_engine.cpp +++ b/src/network_engine.cpp @@ -357,8 +357,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, SockAddr f) { + auto from = f.getMappedIPv4(); if (isMartian(from)) { DHT_LOG.w("Received packet from martian node %s", from.toString().c_str()); return; diff --git a/src/network_utils.cpp b/src/network_utils.cpp index 3868f0de..4e0fc49b 100644 --- a/src/network_utils.cpp +++ b/src/network_utils.cpp @@ -150,7 +150,7 @@ UdpSocket::sendTo(const SockAddr& dest, const uint8_t* data, size_t size, bool r flags |= MSG_NOSIGNAL; #endif - if (sendto(s, data, size, flags, dest.get(), dest.getLength()) == -1) { + if (sendto(s, (const char*)data, size, flags, dest.get(), dest.getLength()) == -1) { int err = errno; logger.d("Can't send message to %s: %s", dest.toString().c_str(), strerror(err)); if (err == EPIPE || err == ENOTCONN || err == ECONNRESET) { @@ -197,10 +197,24 @@ UdpSocket::openSockets(const SockAddr& bind4, const SockAddr& bind6) #if 1 bound6 = {}; if (bind6) { - try { - s6 = bindSocket(bind6, bound6); - } catch (const DhtException& e) { - logger.e("Can't bind inet6 socket: %s", e.what()); + if (bind6.getPort() == 0) { + // Attempt to use the same port as IPv4 with IPv6 + if (auto p4 = bound4.getPort()) { + auto b6 = bind6; + b6.setPort(p4); + try { + s6 = bindSocket(b6, bound6); + } catch (const DhtException& e) { + logger.e("Can't bind inet6 socket: %s", e.what()); + } + } + } + if (s6 == -1) { + try { + s6 = bindSocket(bind6, bound6); + } catch (const DhtException& e) { + logger.e("Can't bind inet6 socket: %s", e.what()); + } } } #endif @@ -211,6 +225,7 @@ UdpSocket::openSockets(const SockAddr& bind4, const SockAddr& bind6) running = true; rcv_thread = std::thread([this, stop_readfd]() { + int selectFd = std::max({s4, s6, stop_readfd}) + 1; try { while (running) { fd_set readfds; @@ -222,7 +237,6 @@ UdpSocket::openSockets(const SockAddr& bind4, const SockAddr& bind6) if(s6 >= 0) FD_SET(s6, &readfds); - int selectFd = std::max({s4, s6, stop_readfd}) + 1; int rc = select(selectFd, &readfds, nullptr, nullptr, nullptr); if (rc < 0) { if (errno != EINTR) { @@ -280,6 +294,7 @@ UdpSocket::openSockets(const SockAddr& bind4, const SockAddr& bind6) } if (s4 < 0 && s6 < 0) break; + selectFd = std::max({s4, s6, stop_readfd}) + 1; } } } diff --git a/src/op_cache.h b/src/op_cache.h index ac64a4be..12e7486a 100644 --- a/src/op_cache.h +++ b/src/op_cache.h @@ -105,14 +105,11 @@ public: void onValuesExpired(const std::vector>& vals); bool addListener(size_t token, const ValueCallback& cb, const Sp& q, const Value::Filter& filter) { - listeners.emplace(token, LocalListener{q, filter, cb}); auto cached = cache.get(filter); - if (not cached.empty()) { - auto ret = cb(cached, false); - if (not ret) - listeners.erase(token); - return ret; + if (not cached.empty() and not cb(cached, false)) { + return false; } + listeners.emplace(token, LocalListener{q, filter, cb}); return true; } diff --git a/src/utils.cpp b/src/utils.cpp index 9b7aabbb..458b044d 100644 --- a/src/utils.cpp +++ b/src/utils.cpp @@ -110,7 +110,7 @@ print_addr(const sockaddr* sa, socklen_t slen) char hbuf[NI_MAXHOST]; char sbuf[NI_MAXSERV]; std::stringstream out; - if (!getnameinfo(sa, slen, hbuf, sizeof(hbuf), sbuf, sizeof(sbuf), NI_NUMERICHOST | NI_NUMERICSERV)) { + if (sa and slen and !getnameinfo(sa, slen, hbuf, sizeof(hbuf), sbuf, sizeof(sbuf), NI_NUMERICHOST | NI_NUMERICSERV)) { if (sa->sa_family == AF_INET6) out << "[" << hbuf << "]"; else @@ -201,10 +201,10 @@ SockAddr::isMappedIPv4() const } SockAddr -SockAddr::getMappedIPv4() const +SockAddr::getMappedIPv4() { if (not isMappedIPv4()) - return *this; + return std::move(*this); SockAddr ret; ret.setFamily(AF_INET); ret.setPort(getPort()); @@ -215,6 +215,22 @@ SockAddr::getMappedIPv4() const return ret; } +SockAddr +SockAddr::getMappedIPv6() +{ + auto family = getFamily(); + if (family != AF_INET) + return std::move(*this); + SockAddr ret; + ret.setFamily(AF_INET6); + ret.setPort(getPort()); + auto addr4 = reinterpret_cast(&getIPv4().sin_addr); + auto addr6 = reinterpret_cast(&ret.getIPv6().sin6_addr); + std::copy(MAPPED_IPV4_PREFIX.begin(), MAPPED_IPV4_PREFIX.end(), addr6); + std::copy_n(addr4, sizeof(in_addr), addr6 + MAPPED_IPV4_PREFIX.size()); + return ret; +} + bool operator==(const SockAddr& a, const SockAddr& b) { return a.equals(b); } diff --git a/tools/tools_common.h b/tools/tools_common.h index a64a5b9f..66cc5792 100644 --- a/tools/tools_common.h +++ b/tools/tools_common.h @@ -303,6 +303,8 @@ void signal_handler(int sig) case SIGHUP: break; case SIGINT: + close(STDIN_FILENO); + // fall through case SIGTERM: runner.kill(); break;