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

dht: don't send probe query for non permanent puts

Browse Source
This commit is contained in:
Adrien Béraud committed 2018-11-22 15:43:55 -05:00
1 parent 4d26537b93
commit e414ca89c7
2 files changed
+41 -22
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/dht.cpp
+35 -15
View File
@@ -381,7 +381,6 @@ void Dht::searchSendAnnounceValue(const Sp<Search>& sr) {
if (sr->announce.empty())
return;
unsigned i = 0;
auto probe_query = std::make_shared<Query>(Select {}.field(Value::Field::Id).field(Value::Field::SeqNum));
std::weak_ptr<Search> ws = sr;
auto onDone = [this,ws](const net::Request& req, net::RequestAnswer&& answer)
@@ -471,25 +470,46 @@ void Dht::searchSendAnnounceValue(const Sp<Search>& sr) {
}
};
Sp<Query> probe_query {};
const auto& now = scheduler.time();
for (auto& n : sr->nodes) {
if (not n.isSynced(now))
continue;
if (std::find_if(sr->announce.cbegin(), sr->announce.cend(),
[&now,&n](const Announce& a) {
return n.getAnnounceTime(a.value->id) <= now;
}) == sr->announce.cend())
continue;
DHT_LOG.w(sr->id, n.node->id, "[search %s] [node %s] sending %s",
sr->id.toString().c_str(), n.node->toString().c_str(), probe_query->toString().c_str());
n.probe_query = probe_query;
n.getStatus[probe_query] = network_engine.sendGetValues(n.node,
sr->id,
*probe_query,
-1,
onSelectDone,
std::bind(&Dht::searchNodeGetExpired, this, _1, _2, ws, probe_query));
const auto& gs = n.probe_query ? n.getStatus.find(n.probe_query) : n.getStatus.cend();
if (gs != n.getStatus.cend() and gs->second and gs->second->pending()) {
continue;
}
bool sendQuery = false;
for (auto& a : sr->announce) {
if (n.getAnnounceTime(a.value->id) <= now) {
if (a.permanent) {
sendQuery = true;
} else {
DHT_LOG.w(sr->id, n.node->id, "[search %s] [node %s] sending 'put' (vid: %d)",
sr->id.toString().c_str(), n.node->toString().c_str(), a.value->id);
n.acked[a.value->id] = {
network_engine.sendAnnounceValue(n.node, sr->id, a.value, a.created, n.token, onDone, onExpired),
now + getType(a.value->type).expiration
};
}
}
}
if (sendQuery) {
if (not probe_query)
probe_query = std::make_shared<Query>(Select {}.field(Value::Field::Id).field(Value::Field::SeqNum));
DHT_LOG.w(sr->id, n.node->id, "[search %s] [node %s] sending %s",
sr->id.toString().c_str(), n.node->toString().c_str(), probe_query->toString().c_str());
n.probe_query = probe_query;
n.getStatus[probe_query] = network_engine.sendGetValues(n.node,
sr->id,
*probe_query,
-1,
onSelectDone,
std::bind(&Dht::searchNodeGetExpired, this, _1, _2, ws, probe_query));
}
if (not n.candidate and ++i == TARGET_NODES)
break;
}
src/search.h
+6 -7
View File
@@ -324,14 +324,13 @@ struct Dht::SearchNode {
*/
time_point getAnnounceTime(Value::Id vid) const {
const auto& ack = acked.find(vid);
const auto& gs = probe_query ? getStatus.find(probe_query) : getStatus.cend();
if ((ack == acked.cend() or not ack->second.first) and (gs == getStatus.cend()
or not gs->second or not gs->second->pending()))
if (ack == acked.cend() or not ack->second.first) {
return time_point::min();
return ((gs != getStatus.cend() and gs->second and gs->second->pending())
or ack == acked.cend() or not ack->second.first or ack->second.first->pending())
? time_point::max()
: (ack->second.first->completed() ? ack->second.second - REANNOUNCE_MARGIN : time_point::min());
}
if (ack->second.first->completed()) {
return ack->second.second - REANNOUNCE_MARGIN;
}
return ack->second.first->pending() ? time_point::max() : time_point::min();
}
/**