Merge branch 'master' into proxy_restinio

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};

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));
     }

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;
 
     /**

include/opendht/dht_proxy_client.h

@@ -196,7 +196,7 @@ public:
      */
     void pushNotificationReceived(const std::map<std::string, std::string>& 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<ListenState> &state, Listener &listener,
                     ListenMethod method = ListenMethod::LISTEN);
 

include/opendht/dhtrunner.h

@@ -56,12 +56,12 @@ public:
     using StatusCallback = std::function<void(NodeStatus, NodeStatus)>;
 
     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 = {});
 

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<Node> insertNode(const InfoHash& myid, const SockAddr& addr) {
         auto n = cache.getNode(myid, addr, scheduler.time(), 0);

include/opendht/securedht.h

@@ -195,8 +195,8 @@ public:
     std::vector<SockAddr> 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);

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<sockaddr, free_delete> addr {};
+    socklen_t len {0};
 
     void set(const sockaddr* sa, socklen_t length) {
         if (len != length) {

include/opendht/thread_pool.h

@@ -49,8 +49,8 @@ public:
         return ret->get_future();
     }
     template<class T>
-    std::shared_ptr<std::future<T>> getShared(std::function<T()>&& cb) {
-        return std::make_shared<std::future<T>>(get(std::move(cb)));
+    std::shared_future<T> getShared(std::function<T()>&& cb) {
+        return get(std::move(cb));
     }
 
     void stop();

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();

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<ListenState> state;
     Sp<asio::steady_timer> 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<Query>(Select{}, where);
-    auto token = search.ops.listen(cb, query, filter, [this, key, filter](
+    auto query = std::make_shared<Query>(Select{}, std::move(where));
+    auto token = search.ops.listen(cb, query, filter, [this, key](
                                    Sp<Query>, ValueCallback cb, SyncCallback) -> size_t {
         std::lock_guard<std::mutex> 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<ListenState>();
         l->second.state = state;
         l->second.cb = [this,key,token,state](const std::vector<Sp<Value>>& values, bool expired) {
@@ -809,14 +810,13 @@ DhtProxyClient::listen(const InfoHash& key, ValueCallback cb, Value::Filter filt
                 return false;
             std::lock_guard<std::mutex> 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<ListenState>& 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>();
     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<Value>(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<std::string, std::string
                 auto state = list.second.state;
                 if (expired == notification.end()) {
                     auto cb = list.second.cb;
-                    auto filter = list.second.filter;
                     auto oldValues = list.second.cache.getValues();
                     get(key, [cb](const std::vector<Sp<Value>>& vals) {
                         return cb(vals, false);
@@ -1233,7 +1230,7 @@ DhtProxyClient::pushNotificationReceived(const std::map<std::string, std::string
                         // 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<Value::Id> 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;

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<std::mutex> 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<std::mutex> 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<std::mutex> 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<std::mutex> lck(dht_mtx);
 #if defined(OPENDHT_PROXY_CLIENT) && defined(OPENDHT_PUSH_NOTIFICATIONS)
     pushToken_ = token;
     if (dht_via_proxy_)

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;

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;
                         }
                     }
                 }

src/op_cache.h

@@ -105,14 +105,11 @@ public:
     void onValuesExpired(const std::vector<Sp<Value>>& vals);
 
     bool addListener(size_t token, const ValueCallback& cb, const Sp<Query>& 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;
     }
 

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<const uint8_t*>(&getIPv4().sin_addr);
+    auto addr6 = reinterpret_cast<uint8_t*>(&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);
 }

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;