mirrors/opendht
1
0 Fork 0
mirror of https://github.com/savoirfairelinux/opendht.git synced 2026-11-08 04:21:51 +00:00
Code Issues Packages Projects Releases Wiki Activity

dht: add synced status

Browse Source
This commit is contained in:
Adrien Béraud committed 2019-02-19 01:59:59 -05:00
1 parent 39e69f813c
commit 35c6a57526
7 files changed
+93 -37
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/dht.cpp
+20 -17
View File
@@ -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);
}
},
@@ -2275,9 +2280,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
+1 -1
View File
@@ -668,7 +668,7 @@ 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){
auto token = it->second.ops.listen(cb, query, filter, [&](Sp<Query> /*q*/, ValueCallback vcb, SyncCallback /*scb*/){
return doListen(key, vcb, filter);
});
return token;
src/listener.h
-7
View File
@@ -48,11 +48,4 @@ struct LocalListener {
ValueCallback get_cb;
};
struct SearchListener {
Sp<Query> query;
Value::Filter filter;
ValueCallback get_cb;
};
}
src/op_cache.cpp
+5 -3
View File
@@ -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());
@@ -109,7 +109,7 @@ OpCache::getExpiration() const {
}
size_t
SearchCache::listen(ValueCallback get_cb, Sp<Query> q, Value::Filter filter, std::function<size_t(Sp<Query>, ValueCallback)> onListen)
SearchCache::listen(ValueCallback get_cb, Sp<Query> q, Value::Filter filter, std::function<size_t(Sp<Query>, ValueCallback, SyncCallback)> onListen)
{
// find exact match
auto op = ops.find(q);
@@ -128,6 +128,8 @@ SearchCache::listen(ValueCallback get_cb, Sp<Query> q, Value::Filter filter, std
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_++;
@@ -177,7 +179,7 @@ SearchCache::expire(const time_point& now, std::function<void(size_t)> onCancel)
}
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
+24 -4
View File
@@ -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,7 @@ 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);
size_t listen(ValueCallback get_cb, Sp<Query> q, Value::Filter filter, std::function<size_t(Sp<Query>, ValueCallback, SyncCallback)> onListen);
bool cancelListen(size_t gtoken, const time_point& now);
void cancelAll(std::function<void(size_t)> onCancel);
@@ -150,7 +170,7 @@ public:
return nextExpiration_;
}
std::vector<Sp<Value>> get(Value::Filter& filter) const;
std::vector<Sp<Value>> get(const Value::Filter& filter) const;
Sp<Value> get(Value::Id id) const;
private:
src/search.h
+17 -3
View File
@@ -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;
@@ -488,10 +502,10 @@ struct Dht::Search {
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/value_cache.h
+26 -2
View File
@@ -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;