diff --git a/src/dht.cpp b/src/dht.cpp index 5ae5ec9f..dd57eb27 100644 --- a/src/dht.cpp +++ b/src/dht.cpp @@ -381,7 +381,6 @@ void Dht::searchSendAnnounceValue(const Sp& sr) { if (sr->announce.empty()) return; unsigned i = 0; - auto probe_query = std::make_shared(Select {}.field(Value::Field::Id).field(Value::Field::SeqNum)); std::weak_ptr ws = sr; auto onDone = [this,ws](const net::Request& req, net::RequestAnswer&& answer) @@ -471,25 +470,46 @@ void Dht::searchSendAnnounceValue(const Sp& sr) { } }; + Sp 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(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; } diff --git a/src/search.h b/src/search.h index 6c5086a0..00621809 100644 --- a/src/search.h +++ b/src/search.h @@ -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(); } /**