From 86088b043bd156045189e4532b9527de85978fa9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Adrien=20B=C3=A9raud?= Date: Fri, 14 Jun 2019 13:06:06 -0400 Subject: [PATCH 01/18] thread pool: getShared returns shared_future instead of shared_ptr --- include/opendht/thread_pool.h | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) 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(); From 738424b1aec7019ad8bed891d1d82ee3e38699ed Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Adrien=20B=C3=A9raud?= Date: Wed, 19 Jun 2019 19:27:37 -0400 Subject: [PATCH 02/18] dhtrunner: use default values for config --- include/opendht/callbacks.h | 14 +++++++------- include/opendht/dhtrunner.h | 36 ++++++++++++------------------------ 2 files changed, 19 insertions(+), 31 deletions(-) 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/dhtrunner.h b/include/opendht/dhtrunner.h index 32e3974b..8fed0277 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 = {}); From d438d4ef1b648ff2a3b4edcef8ce544de5edcd83 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Adrien=20B=C3=A9raud?= Date: Fri, 21 Jun 2019 11:00:48 -0400 Subject: [PATCH 03/18] sockaddr: add getMappedIPv6 --- include/opendht/sockaddr.h | 1 + src/utils.cpp | 16 ++++++++++++++++ 2 files changed, 17 insertions(+) diff --git a/include/opendht/sockaddr.h b/include/opendht/sockaddr.h index 143a3981..98a82009 100644 --- a/include/opendht/sockaddr.h +++ b/include/opendht/sockaddr.h @@ -249,6 +249,7 @@ public: bool isMappedIPv4() const; SockAddr getMappedIPv4() const; + SockAddr getMappedIPv6() const; /** * A comparator to classify IP addresses, only considering the diff --git a/src/utils.cpp b/src/utils.cpp index 9b7aabbb..3c9bc207 100644 --- a/src/utils.cpp +++ b/src/utils.cpp @@ -215,6 +215,22 @@ SockAddr::getMappedIPv4() const return ret; } +SockAddr +SockAddr::getMappedIPv6() const +{ + auto family = getFamily(); + if (family != AF_INET) + return *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); } From feb498305fdb3343b40b3d78b455d51ddbe10699 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Adrien=20B=C3=A9raud?= Date: Fri, 21 Jun 2019 00:33:12 -0400 Subject: [PATCH 04/18] network engine: convert mapped IPv4 --- include/opendht/dht.h | 2 +- include/opendht/dht_interface.h | 2 +- include/opendht/dht_proxy_client.h | 2 +- include/opendht/network_engine.h | 2 +- include/opendht/securedht.h | 4 ++-- include/opendht/sockaddr.h | 8 ++++---- src/dht.cpp | 4 ++-- src/dht_proxy_client.cpp | 2 +- src/dhtrunner.cpp | 2 +- src/network_engine.cpp | 3 ++- src/utils.cpp | 4 ++-- 11 files changed, 18 insertions(+), 17 deletions(-) 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 39ae912c..5dbd8eb9 100644 --- a/include/opendht/dht_proxy_client.h +++ b/include/opendht/dht_proxy_client.h @@ -180,7 +180,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)); } 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 98a82009..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,8 +248,8 @@ public: bool isUnspecified() const; bool isMappedIPv4() const; - SockAddr getMappedIPv4() const; - SockAddr getMappedIPv6() const; + SockAddr getMappedIPv4(); + SockAddr getMappedIPv6(); /** * A comparator to classify IP addresses, only considering the @@ -280,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/src/dht.cpp b/src/dht.cpp index 90e2eec5..440099aa 100644 --- a/src/dht.cpp +++ b/src/dht.cpp @@ -1901,12 +1901,12 @@ 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()); } diff --git a/src/dht_proxy_client.cpp b/src/dht_proxy_client.cpp index 963201d4..153a2961 100644 --- a/src/dht_proxy_client.cpp +++ b/src/dht_proxy_client.cpp @@ -219,7 +219,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 scheduler.syncTime(); diff --git a/src/dhtrunner.cpp b/src/dhtrunner.cpp index 29ada0ef..af782b3c 100644 --- a/src/dhtrunner.cpp +++ b/src/dhtrunner.cpp @@ -525,7 +525,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 { 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/utils.cpp b/src/utils.cpp index 3c9bc207..29e7173e 100644 --- a/src/utils.cpp +++ b/src/utils.cpp @@ -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()); From 89ff1317506fe746dda9264394dad6bc255c425c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Adrien=20B=C3=A9raud?= Date: Fri, 21 Jun 2019 00:33:49 -0400 Subject: [PATCH 05/18] tools: close stdin on sigint --- tools/tools_common.h | 2 ++ 1 file changed, 2 insertions(+) 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; From fbe8834e047504e579f21e0f67578388c6b74270 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Adrien=20B=C3=A9raud?= Date: Sat, 22 Jun 2019 18:49:50 -0400 Subject: [PATCH 06/18] net: compute selectFd once --- src/network_utils.cpp | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/network_utils.cpp b/src/network_utils.cpp index 3868f0de..2a87d8ef 100644 --- a/src/network_utils.cpp +++ b/src/network_utils.cpp @@ -211,6 +211,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 +223,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) { From 2ac5157ab809a39310b32a57b91af76bbfa05952 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Adrien=20B=C3=A9raud?= Date: Sat, 22 Jun 2019 19:07:05 -0400 Subject: [PATCH 07/18] net: attempt to use the same port for IPv4 and IPv6 if 0 --- src/network_utils.cpp | 22 ++++++++++++++++++---- src/utils.cpp | 4 ++-- 2 files changed, 20 insertions(+), 6 deletions(-) diff --git a/src/network_utils.cpp b/src/network_utils.cpp index 2a87d8ef..ecb0d844 100644 --- a/src/network_utils.cpp +++ b/src/network_utils.cpp @@ -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 diff --git a/src/utils.cpp b/src/utils.cpp index 29e7173e..4c70873d 100644 --- a/src/utils.cpp +++ b/src/utils.cpp @@ -216,11 +216,11 @@ SockAddr::getMappedIPv4() } SockAddr -SockAddr::getMappedIPv6() const +SockAddr::getMappedIPv6() { auto family = getFamily(); if (family != AF_INET) - return *this; + return std::move(*this); SockAddr ret; ret.setFamily(AF_INET6); ret.setPort(getPort()); From be0baefee8bf03d089a7de3cb7bb6650db571b4f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Adrien=20B=C3=A9raud?= Date: Sat, 22 Jun 2019 22:51:56 -0400 Subject: [PATCH 08/18] net: recompute selectFd in case of rebind --- src/network_utils.cpp | 1 + 1 file changed, 1 insertion(+) diff --git a/src/network_utils.cpp b/src/network_utils.cpp index ecb0d844..31492cb1 100644 --- a/src/network_utils.cpp +++ b/src/network_utils.cpp @@ -294,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; } } } From b64820af593f73e96b0afe19c9341de0125597dc Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Adrien=20B=C3=A9raud?= Date: Mon, 1 Jul 2019 00:33:53 -0400 Subject: [PATCH 09/18] proxy client: remove unneeded filter Filters are applied at the SearchCache level --- include/opendht/dht_proxy_client.h | 2 +- src/dht_proxy_client.cpp | 46 +++++++++++++++--------------- 2 files changed, 24 insertions(+), 24 deletions(-) diff --git a/include/opendht/dht_proxy_client.h b/include/opendht/dht_proxy_client.h index 5dbd8eb9..7c06f56c 100644 --- a/include/opendht/dht_proxy_client.h +++ b/include/opendht/dht_proxy_client.h @@ -283,7 +283,7 @@ private: RESUBSCRIBE, }; void sendListen(const std::shared_ptr &request, - const ValueCallback &, const Value::Filter &filter, + const ValueCallback &, const Sp &state, ListenMethod method = ListenMethod::LISTEN); diff --git a/src/dht_proxy_client.cpp b/src/dht_proxy_client.cpp index 153a2961..04505977 100644 --- a/src/dht_proxy_client.cpp +++ b/src/dht_proxy_client.cpp @@ -45,14 +45,13 @@ struct DhtProxyClient::Listener { OpValueCache cache; ValueCallback cb; - Value::Filter filter; Sp req; std::thread thread; unsigned callbackId; Sp state; Sp refreshJob; - Listener(OpValueCache&& c, const Sp& r, Value::Filter&& f) - : cache(std::move(c)), filter(std::move(f)),req(r) {} + Listener(OpValueCache&& c, const Sp& r) + : cache(std::move(c)), req(r) {} }; struct PermanentPut { @@ -690,11 +689,12 @@ size_t DhtProxyClient::listen(const InfoHash& key, ValueCallback cb, Value::Filter filter, Where where) { DHT_LOG.d(key, "[search %s]: listen", 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](Sp /*q*/, ValueCallback cb, SyncCallback /*scb*/) -> size_t { + auto query = std::make_shared(Select{}, std::move(where)); + auto token = search.ops.listen(cb, query, filter, [this, key](Sp /*q*/, ValueCallback cb, SyncCallback /*scb*/) -> size_t { scheduler.syncTime(); restbed::Uri uri(serverHost_ + "/" + key.toString()); std::lock_guard lock(searchLock_); + // Find search auto search = searches_.find(key); if (search == searches_.end()) { DHT_LOG.e(key, "[search %s] listen: search not found", key.to_c_str()); @@ -702,19 +702,21 @@ DhtProxyClient::listen(const InfoHash& key, ValueCallback cb, Value::Filter filt } DHT_LOG.d(key, "[search %s] sending %s", key.to_c_str(), deviceKey_.empty() ? "listen" : "subscribe"); + // Add listener auto req = std::make_shared(uri); 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), req, std::move(f))).first; + std::forward_as_tuple(std::move(cb), req)).first; } else { if (l->second.state) l->second.state->cancel = true; + l->second.req = req; } + // Add callback auto state = std::make_shared(); l->second.state = state; l->second.cb = [this,key,token,state](const std::vector>& values, bool expired) { @@ -722,18 +724,16 @@ 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; - l->second.req = req; if (not deviceKey_.empty()) { // Relaunch push listeners even if a timeout is not received (if the proxy crash for any reason) @@ -750,8 +750,10 @@ DhtProxyClient::listen(const InfoHash& key, ValueCallback cb, Value::Filter filt } }); } - l->second.thread = std::thread([this, req, vcb, filter, state]() { - sendListen(req, vcb, filter, state, + + // Send listen to servers + l->second.thread = std::thread([this, req, vcb, state]() { + sendListen(req, vcb, state, deviceKey_.empty() ? ListenMethod::LISTEN : ListenMethod::SUBSCRIBE); }); return token; @@ -788,7 +790,6 @@ DhtProxyClient::cancelListen(const InfoHash& key, size_t gtoken) { void DhtProxyClient::sendListen(const std::shared_ptr &req, const ValueCallback &cb, - const Value::Filter &filter, const Sp &state, ListenMethod method) { auto settings = std::make_shared(); @@ -805,7 +806,7 @@ void DhtProxyClient::sendListen(const std::shared_ptr &req, fillBody(req, method == ListenMethod::RESUBSCRIBE); #endif restbed::Http::async(req, - [this, filter, cb, state](const std::shared_ptr& req, + [this, cb, state](const std::shared_ptr& req, const std::shared_ptr& reply) { auto code = reply->get_status_code(); @@ -830,7 +831,7 @@ void DhtProxyClient::sendListen(const std::shared_ptr &req, } auto expired = json.get("expired", Json::Value(false)).asBool(); auto value = std::make_shared(json); - if ((not filter or filter(*value)) and cb) { + if (cb) { { std::lock_guard lock(lockCallbacks); callbacks_.emplace_back([cb, value, state, expired]() { @@ -998,14 +999,13 @@ DhtProxyClient::restartListeners() // Redo listen state->cancel = false; state->ok = true; - auto filter = listener.filter; auto cb = listener.cb; restbed::Uri uri(serverHost_ + "/" + search.first.toString()); auto req = std::make_shared(uri); req->set_method("LISTEN"); listener.req = req; - listener.thread = std::thread([this, req, cb, filter, state]() { - sendListen(req, cb, filter, state); + listener.thread = std::thread([this, req, cb, state]() { + sendListen(req, cb, state); }); } } From 4374d206ab7fb9b2d7141d09060773f3f851f659 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Adrien=20B=C3=A9raud?= Date: Mon, 1 Jul 2019 00:27:48 -0400 Subject: [PATCH 10/18] op_cache: simplify addListener --- src/op_cache.h | 9 +++------ 1 file changed, 3 insertions(+), 6 deletions(-) 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; } From bce41ecd692ec59a2f859ffe720b795ea1659f47 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Adrien=20B=C3=A9raud?= Date: Mon, 1 Jul 2019 01:12:21 -0400 Subject: [PATCH 11/18] proxy client: fix build with push notifications --- src/dht_proxy_client.cpp | 8 +++----- 1 file changed, 3 insertions(+), 5 deletions(-) diff --git a/src/dht_proxy_client.cpp b/src/dht_proxy_client.cpp index 04505977..6467fa64 100644 --- a/src/dht_proxy_client.cpp +++ b/src/dht_proxy_client.cpp @@ -1052,14 +1052,13 @@ DhtProxyClient::pushNotificationReceived(const std::map>& vals) { return cb(vals, false); }, [cb, oldValues](bool /*ok*/) { // Decrement old values refcount to expire values not present in the new list cb(oldValues, true); - }, std::move(filter)); + }); } else { std::stringstream ss(expired->second); std::vector ids; @@ -1121,9 +1120,8 @@ DhtProxyClient::resubscribe(const InfoHash& key, Listener& listener) listener.req = req; scheduler.edit(listener.refreshJob, scheduler.time() + proxy::OP_TIMEOUT - proxy::OP_MARGIN); auto vcb = listener.cb; - auto filter = listener.filter; - listener.thread = std::thread([this, req, vcb, filter, state]() { - sendListen(req, vcb, filter, state, ListenMethod::RESUBSCRIBE); + listener.thread = std::thread([this, req, vcb, state]() { + sendListen(req, vcb, state, ListenMethod::RESUBSCRIBE); }); #else (void) key; From 89bdfce8405cc4fc2d535823c2f882109394aac1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Adrien=20B=C3=A9raud?= Date: Mon, 1 Jul 2019 12:10:51 -0400 Subject: [PATCH 12/18] dhtrunner: add proper locks for push notification --- src/dhtrunner.cpp | 2 ++ 1 file changed, 2 insertions(+) diff --git a/src/dhtrunner.cpp b/src/dhtrunner.cpp index af782b3c..2998aacf 100644 --- a/src/dhtrunner.cpp +++ b/src/dhtrunner.cpp @@ -1005,6 +1005,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_) @@ -1022,6 +1023,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_) From 150e62de0b6b6ef26cda8669b1434339af6de111 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Adrien=20B=C3=A9raud?= Date: Mon, 1 Jul 2019 14:08:41 -0400 Subject: [PATCH 13/18] sockaddr: fix npe if address is not set --- src/utils.cpp | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/utils.cpp b/src/utils.cpp index 4c70873d..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 From 149b33ef57018f226fa9592d1b7b72f571d74efd Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Adrien=20B=C3=A9raud?= Date: Mon, 1 Jul 2019 15:06:53 -0400 Subject: [PATCH 14/18] dht: simplify error message to avoid use-after-move --- src/dht.cpp | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/dht.cpp b/src/dht.cpp index 440099aa..da431f19 100644 --- a/src/dht.cpp +++ b/src/dht.cpp @@ -1908,7 +1908,7 @@ Dht::periodic(const uint8_t *buf, size_t buflen, SockAddr from) try { 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(); From 27b3c0d47ee78c067029c999abed68a032c6b68d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Adrien=20B=C3=A9raud?= Date: Mon, 1 Jul 2019 23:20:03 -0400 Subject: [PATCH 15/18] dhtrunner: add appropriate lock to setProxyServer --- src/dhtrunner.cpp | 1 + 1 file changed, 1 insertion(+) diff --git a/src/dhtrunner.cpp b/src/dhtrunner.cpp index 2998aacf..c97dcd7f 100644 --- a/src/dhtrunner.cpp +++ b/src/dhtrunner.cpp @@ -940,6 +940,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; From 0d641161e867fbbfa7c595cd66bd418c6d28753b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Adrien=20B=C3=A9raud?= Date: Mon, 1 Jul 2019 23:22:00 -0400 Subject: [PATCH 16/18] dhtrunner: fix minor warning --- src/dhtrunner.cpp | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/dhtrunner.cpp b/src/dhtrunner.cpp index c97dcd7f..a1f653d8 100644 --- a/src/dhtrunner.cpp +++ b/src/dhtrunner.cpp @@ -225,7 +225,7 @@ DhtRunner::shutdown(ShutdownCallback cb) { 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 { if (dht_) dht_->shutdown(cb); }); From 297a23fb1f709ae661d2560e1c60ba1864e68aab Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Adrien=20B=C3=A9raud?= Date: Wed, 26 Jun 2019 16:36:59 -0400 Subject: [PATCH 17/18] dhtrunner: run shutdown on op queue --- src/dhtrunner.cpp | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/src/dhtrunner.cpp b/src/dhtrunner.cpp index a1f653d8..a04311ba 100644 --- a/src/dhtrunner.cpp +++ b/src/dhtrunner.cpp @@ -220,12 +220,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&) mutable { +#ifdef OPENDHT_PROXY_CLIENT + if (dht_via_proxy_) + dht_via_proxy_->shutdown(cb); +#endif if (dht_) dht_->shutdown(cb); }); From a8718242cda414909207320ce0692fc49d3104fb Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Adrien=20B=C3=A9raud?= Date: Tue, 2 Jul 2019 14:51:28 -0400 Subject: [PATCH 18/18] net: fix build on MSVC2017 --- src/network_utils.cpp | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/network_utils.cpp b/src/network_utils.cpp index 31492cb1..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) {