dht: optimize get

CMakeLists.txt

@@ -70,6 +70,8 @@ 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)
     set (CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -fsanitize=address -fstack-protector-strong")
+else ()
+    set (CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -fno-stack-protector")
 endif ()
 set (CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -DMSGPACK_DISABLE_LEGACY_NIL -DMSGPACK_DISABLE_LEGACY_CONVERT")
 if (NOT CMAKE_BUILD_TYPE)
@@ -304,7 +306,7 @@ install (EXPORT opendht DESTINATION ${CMAKE_INSTALL_LIBDIR}/cmake/opendht FILE o
 install (FILES ${CMAKE_CURRENT_BINARY_DIR}/opendhtConfigVersion.cmake DESTINATION ${CMAKE_INSTALL_LIBDIR}/cmake/opendht)
 
 # Unit tests
-IF(OPENDHT_TESTS)
+if (OPENDHT_TESTS)
   FIND_PACKAGE(Cppunit REQUIRED)
   # unit testing
     list (APPEND test_FILES
@@ -325,16 +327,17 @@ IF(OPENDHT_TESTS)
       tests/tests_runner.cpp
       ${test_FILES}
     )
+    target_include_directories(opendht_unit_tests SYSTEM PRIVATE ${CPPUNIT_INCLUDE_DIRS})
     if (OPENDHT_SHARED)
-      TARGET_LINK_LIBRARIES(opendht_unit_tests opendht)
+      target_link_libraries(opendht_unit_tests opendht)
     else ()
-      TARGET_LINK_LIBRARIES(opendht_unit_tests opendht-static)
+      target_link_libraries(opendht_unit_tests opendht-static)
     endif ()
-    TARGET_LINK_LIBRARIES(opendht_unit_tests
+    target_link_libraries(opendht_unit_tests
        ${CMAKE_THREAD_LIBS_INIT}
        ${CPPUNIT_LIBRARIES}
        ${GNUTLS_LIBRARIES}
     )
     enable_testing()
     add_test(TEST opendht_unit_tests)
-ENDIF()
+endif()

include/opendht/dht.h

@@ -163,7 +163,7 @@ public:
     /**
      * Get locally stored data for the given hash.
      */
-    std::vector<Sp<Value>> getLocal(const InfoHash& key, Value::Filter f = Value::AllFilter()) const;
+    std::vector<Sp<Value>> getLocal(const InfoHash& key, Value::Filter f = {}) const;
 
     /**
      * Get locally stored data for the given key and value id.
@@ -469,7 +469,7 @@ private:
     Sp<Search> search(const InfoHash& id, sa_family_t af, GetCallback = {}, QueryCallback = {}, DoneCallback = {}, Value::Filter = {}, const Sp<Query>& q = {});
 
     void announce(const InfoHash& id, sa_family_t af, Sp<Value> value, DoneCallback callback, time_point created=time_point::max(), bool permanent = false);
-    size_t listenTo(const InfoHash& id, sa_family_t af, ValueCallback cb, Value::Filter f = Value::AllFilter(), const Sp<Query>& q = {});
+    size_t listenTo(const InfoHash& id, sa_family_t af, ValueCallback cb, Value::Filter f = {}, const Sp<Query>& q = {});
 
     /**
      * Refill the search with good nodes if possible.

include/opendht/dht_interface.h

@@ -106,7 +106,7 @@ public:
     /**
      * Get locally stored data for the given hash.
      */
-    virtual std::vector<Sp<Value>> getLocal(const InfoHash& key, Value::Filter f = Value::AllFilter()) const = 0;
+    virtual std::vector<Sp<Value>> getLocal(const InfoHash& key, Value::Filter f = {}) const = 0;
 
     /**
      * Get locally stored data for the given key and value id.

include/opendht/dht_proxy_client.h

@@ -41,8 +41,6 @@ namespace Json {
 
 namespace dht {
 
-class SearchCache;
-
 class OPENDHT_PUBLIC DhtProxyClient final : public DhtInterface {
 public:
 
@@ -276,7 +274,6 @@ private:
 
     void opFailed();
 
-    size_t doListen(const InfoHash& key, ValueCallback, Value::Filter);
     bool doCancelListen(const InfoHash& key, size_t token);
 
     struct ListenState;

include/opendht/dhtrunner.h

@@ -63,20 +63,20 @@ public:
     DhtRunner();
     virtual ~DhtRunner();
 
-    void get(InfoHash id, GetCallbackSimple cb, DoneCallback donecb={}, Value::Filter f = Value::AllFilter(), Where w = {}) {
+    void get(InfoHash id, GetCallbackSimple cb, DoneCallback donecb={}, Value::Filter f = {}, Where w = {}) {
         get(id, bindGetCb(cb), donecb, f, w);
     }
 
-    void get(InfoHash id, GetCallbackSimple cb, DoneCallbackSimple donecb={}, Value::Filter f = Value::AllFilter(), Where w = {}) {
+    void get(InfoHash id, GetCallbackSimple cb, DoneCallbackSimple donecb={}, Value::Filter f = {}, Where w = {}) {
         get(id, bindGetCb(cb), donecb, f, w);
     }
 
     void get(InfoHash hash, GetCallback vcb, DoneCallback dcb, Value::Filter f={}, Where w = {});
 
-    void get(InfoHash id, GetCallback cb, DoneCallbackSimple donecb={}, Value::Filter f = Value::AllFilter(), Where w = {}) {
+    void get(InfoHash id, GetCallback cb, DoneCallbackSimple donecb={}, Value::Filter f = {}, Where w = {}) {
         get(id, cb, bindDoneCb(donecb), f, w);
     }
-    void get(const std::string& key, GetCallback vcb, DoneCallbackSimple dcb={}, Value::Filter f = Value::AllFilter(), Where w = {});
+    void get(const std::string& key, GetCallback vcb, DoneCallbackSimple dcb={}, Value::Filter f = {}, Where w = {});
 
     template <class T>
     void get(InfoHash hash, std::function<bool(std::vector<T>&&)> cb, DoneCallbackSimple dcb={})

include/opendht/value.h

@@ -165,8 +165,8 @@ struct OPENDHT_PUBLIC Value
             return chainOr(std::move(f1), std::move(f2));
         }
         static Filter chain(Filter&& f1, Filter&& f2) {
-            if (not f1) return f2;
-            if (not f2) return f1;
+            if (not f1) return std::move(f1);
+            if (not f2) return std::move(f2);
             return [f1,f2](const Value& v) {
                 return f1(v) and f2(v);
             };
@@ -184,7 +184,7 @@ struct OPENDHT_PUBLIC Value
             return chainAll(std::vector<Filter>(l.begin(), l.end()));
         }
         static Filter chainOr(Filter&& f1, Filter&& f2) {
-            if (not f1 or not f2) return AllFilter();
+            if (not f1 or not f2) return {};
             return [f1,f2](const Value& v) {
                 return f1(v) or f2(v);
             };
@@ -990,7 +990,7 @@ template <typename T,
 Value::Filter
 getFilterSet()
 {
-    return Value::AllFilter();
+    return {};
 }
 
 template <class T>

src/dht.cpp

@@ -437,22 +437,18 @@ void Dht::searchSendAnnounceValue(const Sp<Search>& sr) {
             if (not hasValue or seq_no < a.value->seq) {
                 DHT_LOG.d(sr->id, sn->node->id, "[search %s] [node %s] sending 'put' (vid: %d)",
                         sr->id.toString().c_str(), sn->node->toString().c_str(), a.value->id);
-                sn->acked[a.value->id] = std::make_pair(network_engine.sendAnnounceValue(sn->node,
-                                                sr->id,
-                                                a.value,
-                                                a.permanent ? time_point::max() : a.created,
-                                                sn->token,
-                                                onDone,
-                                                onExpired), next_refresh_time);
+                auto created = a.permanent ? time_point::max() : a.created;
+                sn->acked[a.value->id] = {
+                    network_engine.sendAnnounceValue(sn->node, sr->id, a.value, created, sn->token, onDone, onExpired),
+                    next_refresh_time
+                };
             } else if (hasValue and a.permanent) {
                 DHT_LOG.w(sr->id, sn->node->id, "[search %s] [node %s] sending 'refresh' (vid: %d)",
                         sr->id.toString().c_str(), sn->node->toString().c_str(), a.value->id);
-                sn->acked[a.value->id] = std::make_pair(network_engine.sendRefreshValue(sn->node,
-                                                sr->id,
-                                                a.value->id,
-                                                sn->token,
-                                                onDone,
-                                                onExpired), next_refresh_time);
+                sn->acked[a.value->id] = {
+                    network_engine.sendRefreshValue(sn->node, sr->id, a.value->id, sn->token, onDone, onExpired),
+                    next_refresh_time
+                };
             } else {
                 DHT_LOG.w(sr->id, sn->node->id, "[search %s] [node %s] already has value (vid: %d). Aborting.",
                         sr->id.toString().c_str(), sn->node->toString().c_str(), a.value->id);
@@ -540,6 +536,13 @@ Dht::searchSynchedNodeListen(const Sp<Search>& sr, SearchNode& n)
                             l->second.get_cb(l->second.filter.filter(values), expired);
                         }
                     }
+                }, [ws,list_token] (ListenSyncStatus status) {
+                    if (auto sr = ws.lock()) {
+                        auto l = sr->listeners.find(list_token);
+                        if (l != sr->listeners.end()) {
+                            l->second.sync_cb(status);
+                        }
+                    }
                 }
             }).first;
             auto node = n.node;
@@ -557,8 +560,10 @@ Dht::searchSynchedNodeListen(const Sp<Search>& sr, SearchNode& n)
             { /* on done */
                 if (auto sr = ws.lock()) {
                     scheduler.edit(sr->nextSearchStep, scheduler.time());
-                    if (auto sn = sr->getNode(req.node))
+                    if (auto sn = sr->getNode(req.node)) {
                         scheduler.add(sn->getListenTime(query), std::bind(&Dht::searchStep, this, sr));
+                        sn->onListenSynced(query);
+                    }
                     onListenDone(req.node, answer, sr);
                 }
             },
@@ -909,7 +914,7 @@ struct OpStatus {
 
 template <typename T>
 struct GetStatus : public OpStatus {
-    std::vector<Sp<T>> values;
+    T values;
     std::vector<Sp<Node>> nodes;
 };
 
@@ -968,20 +973,16 @@ void doneCallbackWrapper(DoneCallback dcb, const std::vector<Sp<Node>>& nodes, G
     }
 }
 
-template <typename T, typename Cb>
-bool callbackWrapper(Cb get_cb,
-        DoneCallback done_cb,
-        const std::vector<Sp<T>>& values,
-        std::function<std::vector<Sp<T>>(const std::vector<Sp<T>>&)> add_values,
-        Sp<GetStatus<T>> o)
+template <typename T, typename St, typename Cb, typename Av, typename Cv>
+bool callbackWrapper(Cb get_cb, DoneCallback done_cb, const std::vector<Sp<T>>& values,
+    Av add_values, Cv cache_values, GetStatus<St>& op)
 {
-    auto& op = *o;
     if (op.status.done)
         return false;
     auto newvals = add_values(values);
     if (not newvals.empty()) {
         op.status.ok = !get_cb(newvals);
-        op.values.insert(op.values.end(), newvals.begin(), newvals.end());
+        cache_values(newvals);
     }
     doneCallbackWrapper(done_cb, {}, op);
     return !op.status.ok;
@@ -993,23 +994,26 @@ Dht::get(const InfoHash& id, GetCallback getcb, DoneCallback donecb, Value::Filt
     scheduler.syncTime();
 
     auto q = std::make_shared<Query>(Select {}, std::move(where));
-    auto op = std::make_shared<GetStatus<Value>>();
+    auto op = std::make_shared<GetStatus<std::map<Value::Id, Sp<Value>>>>();
     auto f = filter.chain(q->where.getFilter());
 
-    auto add_values = [op,f](const std::vector<Sp<Value>>& values) {
-        std::vector<Sp<Value>> newvals {};
-        for (const auto& v : values) {
-            auto it = std::find_if(op->values.cbegin(), op->values.cend(), [&](const Sp<Value>& sv) {
-                return sv == v or *sv == *v;
-            });
-            if (it == op->values.cend()) {
-               if (not f or f(*v))
-                   newvals.push_back(v);
+    auto gcb = [getcb, donecb, op, f](const std::vector<Sp<Value>>& vals) {
+        auto& o = *op;
+        return callbackWrapper(getcb, donecb, vals, [&o,&f](const std::vector<Sp<Value>>& values) {
+            std::vector<Sp<Value>> newvals {};
+            for (const auto& v : values) {
+                auto it = o.values.find(v->id);
+                if (it == o.values.cend() or (it->second != v && !(*it->second == *v))) {
+                    if (not f or f(*v))
+                        newvals.push_back(v);
+                }
             }
-        }
-        return newvals;
+            return newvals;
+        }, [&o](const std::vector<Sp<Value>>& newvals) {
+            for (const auto& v : newvals)
+                o.values[v->id] = v;
+        }, o);
     };
-    auto gcb = std::bind(callbackWrapper<Value, GetCallback>, getcb, donecb, _1, add_values, op);
 
     /* Try to answer this search locally. */
     gcb(getLocal(id, f));
@@ -1029,36 +1033,39 @@ Dht::get(const InfoHash& id, GetCallback getcb, DoneCallback donecb, Value::Filt
 void Dht::query(const InfoHash& id, QueryCallback cb, DoneCallback done_cb, Query&& q)
 {
     scheduler.syncTime();
-    auto op = std::make_shared<GetStatus<FieldValueIndex>>();
-
+    auto op = std::make_shared<GetStatus<std::vector<Sp<FieldValueIndex>>>>();
     auto f = q.where.getFilter();
-    auto values = getLocal(id, f);
-    auto add_fields = [=](const std::vector<Sp<FieldValueIndex>>& fields) {
-        std::vector<Sp<FieldValueIndex>> newvals {};
-        for (const auto& f : fields) {
-            auto it = std::find_if(op->values.cbegin(), op->values.cend(),
-                [&](const Sp<FieldValueIndex>& sf) {
-                    return sf == f or f->containedIn(*sf);
-                });
-            if (it == op->values.cend()) {
-                auto lesser = std::find_if(op->values.begin(), op->values.end(),
+    auto qcb = [cb, done_cb, op](const std::vector<Sp<FieldValueIndex>>& fields){
+        auto& o = *op;
+        return callbackWrapper(cb, done_cb, fields, [&](const std::vector<Sp<FieldValueIndex>>& fields) {
+            std::vector<Sp<FieldValueIndex>> newvals {};
+            for (const auto& f : fields) {
+                auto it = std::find_if(o.values.cbegin(), o.values.cend(),
                     [&](const Sp<FieldValueIndex>& sf) {
-                        return sf->containedIn(*f);
+                        return sf == f or f->containedIn(*sf);
                     });
-                if (lesser != op->values.end())
-                    op->values.erase(lesser);
-                newvals.push_back(f);
+                if (it == o.values.cend()) {
+                    auto lesser = std::find_if(o.values.begin(), o.values.end(),
+                        [&](const Sp<FieldValueIndex>& sf) {
+                            return sf->containedIn(*f);
+                        });
+                    if (lesser != o.values.end())
+                        o.values.erase(lesser);
+                    newvals.push_back(f);
+                }
             }
-        }
-        return newvals;
+            return newvals;
+        }, [&](const std::vector<Sp<FieldValueIndex>>& fields){
+            o.values.insert(o.values.end(), fields.begin(), fields.end());
+        }, o);
     };
+
+    /* Try to answer this search locally. */
+    auto values = getLocal(id, f);
     std::vector<Sp<FieldValueIndex>> local_fields(values.size());
     std::transform(values.begin(), values.end(), local_fields.begin(), [&q](const Sp<Value>& v) {
         return std::make_shared<FieldValueIndex>(*v, q.select);
     });
-    auto qcb = std::bind(callbackWrapper<FieldValueIndex, QueryCallback>, cb, done_cb, _1, add_fields, op);
-
-    /* Try to answer this search locally. */
     qcb(local_fields);
 
     auto sq = std::make_shared<Query>(std::move(q));
@@ -1166,6 +1173,7 @@ Dht::storageChanged(const InfoHash& id, Storage& st, ValueStorage& v, bool newVa
         if (not st.local_listeners.empty()) {
             DHT_LOG.d(id, "[store %s] %lu local listeners", id.toString().c_str(), st.local_listeners.size());
             std::vector<std::pair<ValueCallback, std::vector<Sp<Value>>>> cbs;
+            cbs.reserve(st.local_listeners.size());
             for (const auto& l : st.local_listeners) {
                 std::vector<Sp<Value>> vals;
                 if (not l.second.filter or l.second.filter(*v.data))
@@ -2275,9 +2283,7 @@ Dht::onListen(Sp<Node> node, const InfoHash& hash, const Blob& token, size_t soc
 }
 
 void
-Dht::onListenDone(const Sp<Node>& node,
-        net::RequestAnswer& answer,
-        Sp<Search>& sr)
+Dht::onListenDone(const Sp<Node>& node, net::RequestAnswer& answer, Sp<Search>& sr)
 {
     // DHT_LOG.d(sr->id, node->id, "[search %s] [node %s] got listen confirmation",
     //            sr->id.toString().c_str(), node->toString().c_str(), answer.values.size());

src/dht_proxy_client.cpp

@@ -527,7 +527,14 @@ DhtProxyClient::getProxyInfos()
     }
     statusThread_ = std::thread([this, serverHost, infoState]{
         try {
-            auto hostAndService = splitPort(serverHost);
+            auto endpointStr = serverHost;
+            auto protocol = std::string(proxy::HTTP_PROTO);
+            auto protocolIdx = serverHost.find("://");
+            if (protocolIdx != std::string::npos) {
+                protocol = endpointStr.substr(0, protocolIdx + 3);
+                endpointStr = endpointStr.substr(protocolIdx + 3);
+            }
+            auto hostAndService = splitPort(endpointStr);
             auto resolved_proxies = SockAddr::resolve(hostAndService.first, hostAndService.second);
             std::vector<std::future<Sp<restbed::Response>>> reqs;
             reqs.reserve(resolved_proxies.size());
@@ -536,9 +543,9 @@ DhtProxyClient::getProxyInfos()
                 if (resolved_proxy.getFamily() == AF_INET6) {
                     // HACK restbed seems to not correctly handle directly http://[ipv6]
                     // See https://github.com/Corvusoft/restbed/issues/290.
-                    server = serverHost;
+                    server = endpointStr;
                 }
-                restbed::Uri uri(proxy::HTTP_PROTO + server + "/");
+                restbed::Uri uri(protocol + server + "/");
                 auto req = std::make_shared<restbed::Request>(uri);
                 if (infoState->cancel)
                     return;
@@ -661,8 +668,93 @@ DhtProxyClient::listen(const InfoHash& key, ValueCallback cb, Value::Filter filt
         it = searches_.emplace(key, ProxySearch{}).first;
     }
     auto query = std::make_shared<Query>(Select{}, where);
-    auto token = it->second.ops.listen(cb, query, filter, [&](Sp<Query> /*q*/, ValueCallback vcb){
-        return doListen(key, vcb, filter);
+    auto token = it->second.ops.listen(cb, query, filter, [this, key, filter](Sp<Query> /*q*/, ValueCallback cb, SyncCallback scb) -> size_t {
+        scheduler.syncTime();
+        restbed::Uri uri(serverHost_ + "/" + key.toString());
+        std::lock_guard<std::mutex> lock(searchLock_);
+        auto search = searches_.find(key);
+        if (search == searches_.end()) {
+            DHT_LOG.e(key, "[search %s] listen: search not found", key.to_c_str());
+            return 0;
+        }
+        DHT_LOG.d(key, "[search %s] sending %s", key.to_c_str(), deviceKey_.empty() ? "listen" : "subscribe");
+
+        auto req = std::make_shared<restbed::Request>(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(token, Listener {
+                ValueCache(cb, std::move(scb)), scheduler.add(time_point::max(), [this, key, token]{
+                    std::lock_guard<std::mutex> lock(searchLock_);
+                    auto s = searches_.find(key);
+                    if (s == searches_.end()) {
+                        return;
+                    }
+                    auto l = s->second.listeners.find(token);
+                    if (l == s->second.listeners.end()) {
+                        return;
+                    }
+                    auto next = l->second.cache.expireValues(scheduler.time());
+                    scheduler.edit(l->second.cacheExpirationJob, next);
+                }), req, std::move(f)
+            }).first;
+        } else {
+            if (l->second.state)
+                l->second.state->cancel = true;
+        }
+
+        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) {
+            if (state->cancel)
+                return false;
+            std::lock_guard<std::mutex> lock(searchLock_);
+            auto s = searches_.find(key);
+            if (s == searches_.end()) {
+                return false;
+            }
+            auto l = s->second.listeners.find(token);
+            if (l == s->second.listeners.end()) {
+                return false;
+            }
+            const std::vector<Sp<Value>> new_values_empty;
+            std::vector<Value::Id> expired_ids;
+            if (expired) {
+                expired_ids.reserve(values.size());
+                for (const auto& v : values)
+                    expired_ids.emplace_back(v->id);
+            }
+            auto next = l->second.cache.onValues(expired ? new_values_empty : values, std::vector<Value::Id>{}, expired_ids, types, scheduler.time());
+            scheduler.edit(l->second.cacheExpirationJob, next);
+            loopSignal_();
+            return true;
+        };
+
+        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)
+            l->second.refreshJob = scheduler.add(scheduler.time() + proxy::OP_TIMEOUT - proxy::OP_MARGIN, [this, key, token, state] {
+                if (state->cancel)
+                    return;
+                std::lock_guard<std::mutex> lock(searchLock_);
+                auto s = searches_.find(key);
+                if (s != searches_.end()) {
+                    auto l = s->second.listeners.find(token);
+                    if (l != s->second.listeners.end()) {
+                        resubscribe(key, l->second);
+                    }
+                }
+            });
+        }
+        l->second.thread = std::thread([this, req, vcb, filter, state]() {
+            sendListen(req, vcb, filter, state,
+                    deviceKey_.empty() ? ListenMethod::LISTEN
+                                    : ListenMethod::SUBSCRIBE);
+        });
+        return token;
     });
     return token;
 }
@@ -738,7 +830,7 @@ void DhtProxyClient::sendListen(const std::shared_ptr<restbed::Request> &req,
                             }
                             auto expired = json.get("expired", Json::Value(false)).asBool();
                             auto value = std::make_shared<Value>(json);
-                            if ((not filter or filter(*value)) and cb)  {
+                            if ((not filter or filter(*value)) and cb) {
                                 std::lock_guard<std::mutex> lock(lockCallbacks);
                                 callbacks_.emplace_back([cb, value, state, expired]() {
                                     if (not state->cancel and not cb({value}, expired))
@@ -766,97 +858,6 @@ void DhtProxyClient::sendListen(const std::shared_ptr<restbed::Request> &req,
         opFailed();
 }
 
-size_t
-DhtProxyClient::doListen(const InfoHash& key, ValueCallback cb, Value::Filter filter/*, Where where*/)
-{
-    scheduler.syncTime();
-    restbed::Uri uri(serverHost_ + "/" + key.toString());
-    std::lock_guard<std::mutex> lock(searchLock_);
-    auto search = searches_.find(key);
-    if (search == searches_.end()) {
-        DHT_LOG.e(key, "[search %s] listen: search not found", key.to_c_str());
-        return 0;
-    }
-    DHT_LOG.d(key, "[search %s] sending %s", key.to_c_str(), deviceKey_.empty() ? "listen" : "subscribe");
-
-    auto req = std::make_shared<restbed::Request>(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(token, Listener {
-            ValueCache(cb), scheduler.add(time_point::max(), [this, key, token]{
-                std::lock_guard<std::mutex> lock(searchLock_);
-                auto s = searches_.find(key);
-                if (s == searches_.end()) {
-                    return;
-                }
-                auto l = s->second.listeners.find(token);
-                if (l == s->second.listeners.end()) {
-                    return;
-                }
-                auto next = l->second.cache.expireValues(scheduler.time());
-                scheduler.edit(l->second.cacheExpirationJob, next);
-            }), req, std::move(f)
-        }).first;
-    } else {
-        if (l->second.state)
-            l->second.state->cancel = true;
-    }
-
-    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) {
-        if (state->cancel)
-            return false;
-        std::lock_guard<std::mutex> lock(searchLock_);
-        auto s = searches_.find(key);
-        if (s == searches_.end()) {
-            return false;
-        }
-        auto l = s->second.listeners.find(token);
-        if (l == s->second.listeners.end()) {
-            return false;
-        }
-        const std::vector<Sp<Value>> new_values_empty;
-        std::vector<Value::Id> expired_ids;
-        if (expired) {
-            expired_ids.reserve(values.size());
-            for (const auto& v : values)
-                expired_ids.emplace_back(v->id);
-        }
-        auto next = l->second.cache.onValues(expired ? new_values_empty : values, std::vector<Value::Id>{}, expired_ids, types, scheduler.time());
-        scheduler.edit(l->second.cacheExpirationJob, next);
-        loopSignal_();
-        return true;
-    };
-
-    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)
-        l->second.refreshJob = scheduler.add(scheduler.time() + proxy::OP_TIMEOUT - proxy::OP_MARGIN, [this, key, token, state] {
-            if (state->cancel)
-                return;
-            std::lock_guard<std::mutex> lock(searchLock_);
-            auto s = searches_.find(key);
-            if (s != searches_.end()) {
-                auto l = s->second.listeners.find(token);
-                if (l != s->second.listeners.end()) {
-                    resubscribe(key, l->second);
-                }
-            }
-        });
-    }
-    l->second.thread = std::thread([this, req, vcb, filter, state]() {
-        sendListen(req, vcb, filter, state,
-                deviceKey_.empty() ? ListenMethod::LISTEN
-                                   : ListenMethod::SUBSCRIBE);
-    });
-    return token;
-}
-
 bool
 DhtProxyClient::doCancelListen(const InfoHash& key, size_t ltoken)
 {

src/dhtrunner.cpp

@@ -521,7 +521,7 @@ DhtRunner::startNetwork(const SockAddr sin4, const SockAddr sin6)
 #ifndef _WIN32
     auto status = pipe(stopfds);
     if (status == -1) {
-        throw DhtException("Can't open pipe");
+        throw DhtException(std::string("Can't open pipe: ") + strerror(errno));
     }
 #else
     udpPipe(stopfds);
@@ -533,15 +533,29 @@ DhtRunner::startNetwork(const SockAddr sin4, const SockAddr sin6)
     s6 = -1;
 
     bound4 = {};
-    if (sin4)
-        s4 = bindSocket(sin4, bound4);
+    if (sin4) {
+        try {
+            s4 = bindSocket(sin4, bound4);
+        } catch (const DhtException& e) {
+            std::cerr << "Can't bind inet socket: " << e.what() << std::endl;
+        }
+    }
 
 #if 1
     bound6 = {};
-    if (sin6)
-        s6 = bindSocket(sin6, bound6);
+    if (sin6) {
+        try {
+            s6 = bindSocket(sin6, bound6);
+        } catch (const DhtException& e) {
+            std::cerr << "Can't bind inet6 socket: " << e.what() << std::endl;
+        }
+    }
 #endif
 
+    if (s4 == -1 && s6 == -1) {
+        throw DhtException("Can't bind socket");
+    }
+
     running_network = true;
     rcv_thread = std::thread([this, stop_readfd]() {
         try {

src/listener.h

@@ -48,11 +48,4 @@ struct LocalListener {
     ValueCallback get_cb;
 };
 
-
-struct SearchListener {
-    Sp<Query> query;
-    Value::Filter filter;
-    ValueCallback get_cb;
-};
-
 }

src/network_engine.cpp

@@ -353,9 +353,8 @@ NetworkEngine::rateLimit(const SockAddr& addr)
         limiter_maintenance = 0;
     }
 
-    auto it = address_rate_limiter.emplace(addr, IpLimiter{});
     // invoke per IP, then global rate limiter
-    return it.first->second.limit(now) and rate_limiter.limit(now);
+    return address_rate_limiter[addr].limit(now) and rate_limiter.limit(now);
 }
 
 bool
@@ -475,15 +474,15 @@ NetworkEngine::processMessage(const uint8_t *buf, size_t buflen, const SockAddr&
         process(std::move(msg), from);
     } else {
         // starting partial message session
-        PartialMessage pmsg;
-        pmsg.from = from;
-        pmsg.msg = std::move(msg);
-        pmsg.start = now;
-        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));
+        auto k = msg->tid;
+        auto& pmsg = partial_messages[k];
+        if (not pmsg.msg) {
+            pmsg.from = from;
+            pmsg.msg = std::move(msg);
+            pmsg.start = now;
+            pmsg.last_part = now;
+            scheduler.add(now + RX_MAX_PACKET_TIME, std::bind(&NetworkEngine::maintainRxBuffer, this, k));
+            scheduler.add(now + RX_TIMEOUT, std::bind(&NetworkEngine::maintainRxBuffer, this, k));
         } else
             DHT_LOG.e("Partial message with given TID already exists");
     }

src/op_cache.cpp

@@ -25,7 +25,7 @@ bool
 OpValueCache::onValuesAdded(const std::vector<Sp<Value>>& vals) {
     std::vector<Sp<Value>> newValues;
     for (const auto& v : vals) {
-        auto viop = values.emplace(v->id, OpCacheValueStorage{v});
+        auto viop = values.emplace(v->id, v);
         if (viop.second) {
             newValues.emplace_back(v);
         } else {
@@ -50,7 +50,7 @@ OpValueCache::onValuesExpired(const std::vector<Sp<Value>>& vals) {
     return expiredValues.empty() ? true : callback(expiredValues, true);
 }
 std::vector<Sp<Value>>
-OpValueCache::get(Value::Filter& filter) const {
+OpValueCache::get(const Value::Filter& filter) const {
     std::vector<Sp<Value>> ret;
     if (not filter)
         ret.reserve(values.size());
@@ -108,26 +108,51 @@ OpCache::getExpiration() const {
     return lastRemoved + EXPIRATION;
 }
 
-size_t
-SearchCache::listen(ValueCallback get_cb, Sp<Query> q, Value::Filter filter, std::function<size_t(Sp<Query>, ValueCallback)> onListen)
+SearchCache::OpMap::iterator
+SearchCache::getOp(const Sp<Query>& q)
 {
     // find exact match
     auto op = ops.find(q);
-    if (op == ops.end()) {
-        // find satisfying query
-        for (auto it = ops.begin(); it != ops.end(); it++) {
-            if (q->isSatisfiedBy(*it->first)) {
-                op = it;
-                break;
-            }
+    if (op != ops.end())
+        return op;
+    // find satisfying query
+    for (auto it = ops.begin(); it != ops.end(); it++) {
+        if (q->isSatisfiedBy(*it->first)) {
+            return it;
         }
     }
+    return ops.end();
+}
+
+SearchCache::OpMap::const_iterator
+SearchCache::getOp(const Sp<Query>& q) const
+{
+    // find exact match
+    auto op = ops.find(q);
+    if (op != ops.cend())
+        return op;
+    // find satisfying query
+    for (auto it = ops.begin(); it != ops.end(); it++) {
+        if (q->isSatisfiedBy(*it->first)) {
+            return it;
+        }
+    }
+    return ops.cend();
+}
+
+size_t
+SearchCache::listen(ValueCallback get_cb, Sp<Query> q, Value::Filter filter, OnListen onListen)
+{
+    // find exact match
+    auto op = getOp(q);
     if (op == ops.end()) {
         // New query
         op = ops.emplace(q, std::unique_ptr<OpCache>(new OpCache)).first;
         auto& cache = *op->second;
         cache.searchToken = onListen(q, [&](const std::vector<Sp<Value>>& values, bool expired){
             return cache.onValue(values, expired);
+        }, [&](ListenSyncStatus status) {
+            cache.onNodeChanged(status);
         });
     }
     auto token = nextToken_++;
@@ -176,8 +201,22 @@ SearchCache::expire(const time_point& now, std::function<void(size_t)> onCancel)
     return ret;
 }
 
+bool
+SearchCache::get(const Value::Filter& f, const Sp<Query>& q, const GetCallback& gcb, const DoneCallback& dcb) const
+{
+    auto op = getOp(q);
+    if (op != ops.end()) {
+        auto vals = op->second->get(f);
+        if ((not vals.empty() and not gcb(vals)) or op->second->isSynced()) {
+            dcb(true, {});
+            return true;
+        }
+    }
+    return false;
+}
+
 std::vector<Sp<Value>>
-SearchCache::get(Value::Filter& filter) const {
+SearchCache::get(const Value::Filter& filter) const {
     if (ops.size() == 1)
         return ops.begin()->second->get(filter);
     std::map<Value::Id, Sp<Value>> c;

src/op_cache.h

@@ -27,7 +27,7 @@ struct OpCacheValueStorage
 {
     Sp<Value> data {};
     unsigned refCount {1};
-    OpCacheValueStorage(Sp<Value> val = {}) : data(val) {}
+    OpCacheValueStorage(Sp<Value> val) : data(val) {}
 };
 
 class OpValueCache {
@@ -57,7 +57,18 @@ public:
     bool onValuesAdded(const std::vector<Sp<Value>>& vals);
     bool onValuesExpired(const std::vector<Sp<Value>>& vals);
 
-    std::vector<Sp<Value>> get(Value::Filter& filter) const;
+    void onNodeChanged(ListenSyncStatus status) {
+        switch (status) {
+            case ListenSyncStatus::ADDED: nodes++; break;
+            case ListenSyncStatus::REMOVED: nodes--; break;
+            case ListenSyncStatus::SYNCED : syncedNodes++; break;
+            case ListenSyncStatus::UNSYNCED: syncedNodes--; break;
+        }
+    }
+
+    bool isSynced() const { return nodes > 0 and syncedNodes == nodes; }
+
+    std::vector<Sp<Value>> get(const Value::Filter& filter) const;
     Sp<Value> get(Value::Id id) const;
     std::vector<Sp<Value>> getValues() const;
 
@@ -65,6 +76,8 @@ private:
     OpValueCache(const OpValueCache&) = delete;
     OpValueCache& operator=(const OpValueCache&) = delete;
 
+    size_t nodes {0};
+    size_t syncedNodes {0};
     std::map<Value::Id, OpCacheValueStorage> values {};
     ValueCallback callback;
 };
@@ -83,6 +96,9 @@ public:
         cache.onValue(vals, expired);
         return not listeners.empty();
     }
+    void onNodeChanged(ListenSyncStatus status) {
+        cache.onNodeChanged(status);
+    }
 
     void onValuesAdded(const std::vector<Sp<Value>>& vals);
     void onValuesExpired(const std::vector<Sp<Value>>& vals);
@@ -112,7 +128,7 @@ public:
         return listeners.empty();
     }
 
-    std::vector<Sp<Value>> get(Value::Filter& filter) const {
+    std::vector<Sp<Value>> get(const Value::Filter& filter) const {
         return cache.get(filter);
     }
 
@@ -120,6 +136,10 @@ public:
         return cache.get(id);
     }
 
+    bool isSynced() const {
+        return cache.isSynced();
+    }
+
     bool isExpired(const time_point& now) const {
         return listeners.empty() and (lastRemoved + EXPIRATION < now);
     }
@@ -140,7 +160,9 @@ class SearchCache {
 public:
     SearchCache() {}
     SearchCache(SearchCache&&) = default;
-    size_t listen(ValueCallback get_cb, Sp<Query> q, Value::Filter filter, std::function<size_t(Sp<Query>, ValueCallback)> onListen);
+
+    using OnListen = std::function<size_t(Sp<Query>, ValueCallback, SyncCallback)>;
+    size_t listen(ValueCallback get_cb, Sp<Query> q, Value::Filter filter, OnListen onListen);
 
     bool cancelListen(size_t gtoken, const time_point& now);
     void cancelAll(std::function<void(size_t)> onCancel);
@@ -150,14 +172,19 @@ public:
         return nextExpiration_;
     }
 
-    std::vector<Sp<Value>> get(Value::Filter& filter) const;
+    bool get(const Value::Filter& f, const Sp<Query>& q, const GetCallback& gcb, const DoneCallback& dcb) const;
+    std::vector<Sp<Value>> get(const Value::Filter& filter) const;
     Sp<Value> get(Value::Id id) const;
 
 private:
     SearchCache(const SearchCache&) = delete;
     SearchCache& operator=(const SearchCache&) = delete;
 
-    std::map<Sp<Query>, std::unique_ptr<OpCache>> ops {};
+    using OpMap = std::map<Sp<Query>, std::unique_ptr<OpCache>>;
+    OpMap ops {};
+    OpMap::iterator getOp(const Sp<Query>& q);
+    OpMap::const_iterator getOp(const Sp<Query>& q) const;
+
     size_t nextToken_ {1};
     time_point nextExpiration_ {time_point::max()};
 };

src/search.h

@@ -65,7 +65,8 @@ struct Dht::SearchNode {
         ValueCache cache;
         Sp<Scheduler::Job> cacheExpirationJob {};
         Sp<net::Request> req {};
-        CachedListenStatus(ValueStateCallback&& cb) : cache(std::forward<ValueStateCallback>(cb)) {}
+        CachedListenStatus(ValueStateCallback&& cb, SyncCallback&& scb)
+         : cache(std::forward<ValueStateCallback>(cb), std::forward<SyncCallback>(scb)) {}
         CachedListenStatus(CachedListenStatus&&) = default;
         CachedListenStatus(const CachedListenStatus&) = delete;
         CachedListenStatus& operator=(const CachedListenStatus&) = delete;
@@ -225,6 +226,13 @@ struct Dht::SearchNode {
         }
     }
 
+    void onListenSynced(const Sp<Query>& q, bool synced = true) {
+        auto l = listenStatus.find(q);
+        if (l != listenStatus.end()) {
+            l->second.cache.onSynced(synced);
+        }
+    }
+
     void expireValues(const Sp<Query>& q, Scheduler& scheduler) {
         auto l = listenStatus.find(q);
         if (l != listenStatus.end()) {
@@ -377,6 +385,12 @@ struct Dht::Search {
     std::multimap<time_point, Get> callbacks {};
 
     /* listeners */
+    struct SearchListener {
+        Sp<Query> query;
+        Value::Filter filter;
+        ValueCallback get_cb;
+        SyncCallback sync_cb;
+    };
     std::map<size_t, SearchListener> listeners {};
     size_t listener_token = 1;
 
@@ -477,21 +491,20 @@ struct Dht::Search {
 
     void get(Value::Filter f, const Sp<Query>& q, const QueryCallback& qcb, const GetCallback& gcb, const DoneCallback& dcb, Scheduler& scheduler) {
         if (gcb or qcb) {
-            const auto& now = scheduler.time();
-            callbacks.emplace(now, Get { now, f, q, qcb, gcb, dcb });
-            auto values = cache.get(f);
-            if (not values.empty())
-                gcb(values);
-            scheduler.edit(nextSearchStep, now);
+            if (not cache.get(f, q, gcb, dcb)) {
+                const auto& now = scheduler.time();
+                callbacks.emplace(now, Get { now, f, q, qcb, gcb, dcb });
+                scheduler.edit(nextSearchStep, now);
+            }
         }
     }
 
     size_t listen(ValueCallback cb, Value::Filter f, const Sp<Query>& q, Scheduler& scheduler) {
         //DHT_LOG.e(id, "[search %s IPv%c] listen", id.toString().c_str(), (af == AF_INET) ? '4' : '6');
-        return cache.listen(cb, q, f, [&](const Sp<Query>& q, ValueCallback vcb){
+        return cache.listen(cb, q, f, [&](const Sp<Query>& q, ValueCallback vcb, SyncCallback scb){
             done = false;
             auto token = ++listener_token;
-            listeners.emplace(token, SearchListener{q, f, vcb});
+            listeners.emplace(token, SearchListener{q, f, vcb, scb});
             scheduler.edit(nextSearchStep, scheduler.time());
             return token;
         });

src/securedht.cpp

@@ -284,6 +284,8 @@ SecureDht::getCallbackFilter(ValueCallback cb, Value::Filter&& filter)
 {
     return [=](const std::vector<Sp<Value>>& values, bool expired) {
         std::vector<Sp<Value>> tmpvals {};
+        if (not filter)
+            tmpvals.reserve(values.size());
         for (const auto& v : values) {
             if (auto nv = checkValue(v))
                 if (not filter or filter(*nv))
@@ -301,6 +303,8 @@ SecureDht::getCallbackFilter(GetCallback cb, Value::Filter&& filter)
 {
     return [=](const std::vector<Sp<Value>>& values) {
         std::vector<Sp<Value>> tmpvals {};
+        if (not filter)
+            tmpvals.reserve(values.size());
         for (const auto& v : values) {
             if (auto nv = checkValue(v))
                 if (not filter or filter(*nv))

src/value.cpp

@@ -288,7 +288,7 @@ FieldValue::getLocalFilter() const
         case Value::Field::UserType:
             return Value::UserTypeFilter(std::string {blobValue.begin(), blobValue.end()});
         default:
-            return Value::AllFilter();
+            return {};
     }
 }
 
@@ -497,8 +497,7 @@ template <typename T>
 bool subset(std::vector<T> fds, std::vector<T> qfds)
 {
     for (auto& fd : fds) {
-        auto correspondance = std::find_if(qfds.begin(), qfds.end(), [&fd](T& _vfd) { return fd == _vfd; });
-        if (correspondance == qfds.end())
+        if (std::find_if(qfds.begin(), qfds.end(), [&fd](T& _vfd) { return fd == _vfd; }) == qfds.end())
             return false;
     }
     return true;
@@ -506,10 +505,9 @@ bool subset(std::vector<T> fds, std::vector<T> qfds)
 
 bool Select::isSatisfiedBy(const Select& os) const {
     /* empty, means all values are selected. */
-    if (fieldSelection_.empty() and not os.fieldSelection_.empty())
-        return false;
-    else
-        return subset(fieldSelection_, os.fieldSelection_);
+    return fieldSelection_.empty() ?
+        os.fieldSelection_.empty() :
+        subset(fieldSelection_, os.fieldSelection_);
 }
 
 bool Where::isSatisfiedBy(const Where& ow) const {

src/value_cache.h

@@ -22,19 +22,32 @@
 namespace dht {
 
 using ValueStateCallback = std::function<void(const std::vector<Sp<Value>>&, bool)>;
+enum class ListenSyncStatus { ADDED, SYNCED, UNSYNCED, REMOVED };
+using SyncCallback = std::function<void(ListenSyncStatus)>;
 using CallbackQueue = std::list<std::function<void()>>;
 
 class ValueCache {
 public:
-    ValueCache(ValueStateCallback&& cb) : callback(std::forward<ValueStateCallback>(cb)) {}
-    ValueCache(ValueCache&& o) : values(std::move(o.values)), callback(std::move(o.callback)) {
+    ValueCache(ValueStateCallback&& cb, SyncCallback&& scb = {})
+        : callback(std::forward<ValueStateCallback>(cb)), syncCallback(std::move(scb))
+    {
+        if (syncCallback)
+            syncCallback(ListenSyncStatus::ADDED);
+    }
+    ValueCache(ValueCache&& o) : values(std::move(o.values)), callback(std::move(o.callback)), syncCallback(std::move(o.syncCallback)) {
         o.callback = {};
+        o.syncCallback = {};
     }
 
     ~ValueCache() {
         auto q = clear();
         for (auto& cb: q)
             cb();
+        if (syncCallback) {
+            if (status == ListenSyncStatus::SYNCED)
+                syncCallback(ListenSyncStatus::UNSYNCED);
+            syncCallback(ListenSyncStatus::REMOVED);
+        }
     }
 
     CallbackQueue clear() {
@@ -120,6 +133,15 @@ public:
         return ret;
     }
 
+    void onSynced(bool synced) {
+        auto newStatus = synced ? ListenSyncStatus::SYNCED : ListenSyncStatus::UNSYNCED;
+        if (status != newStatus) {
+            status = newStatus;
+            if (syncCallback)
+                syncCallback(newStatus);
+        }
+    }
+
 private:
     // prevent copy
     ValueCache(const ValueCache&) = delete;
@@ -141,6 +163,8 @@ private:
 
     std::map<Value::Id, CacheValueStorage> values;
     ValueStateCallback callback;
+    SyncCallback syncCallback;
+    ListenSyncStatus status {ListenSyncStatus::UNSYNCED};
 
     CallbackQueue addValues(const std::vector<Sp<Value>>& new_values, const TypeStore& types, const time_point& now) {
         std::vector<Sp<Value>> nvals;