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
Files
9b9056c0c8d21c62743d2d75f468fde26c48e0fc
opendht/src/dht.cpp
T
Cyrille BéraudandCopilot 9b9056c0c8 dht: do not report unfiltered answers to filtered queries 
Query::isSatisfiedBy() considers an unfiltered answer sufficient for a
filtered query, which is true for a vanilla get: onGetValuesDone() applies
the operation filter to the values before calling back.

A projected query is different. It receives the fields another query selected
and can no longer evaluate its own filters on them, all the more so as those
filters may target fields that were not selected at all. Two concurrent
queries such as "SELECT id" and "SELECT id WHERE value_type=1" therefore made
the filtered callback report identifiers whose type does not match, and
searchNodeGetDone() additionally cancelled its own request so it never got a
correctly filtered answer.

Add Query::isProjectionSatisfiedBy(), which additionally requires both queries
to apply the same filters, and use it for operations with a query callback.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
2026-10-26 09:21:10 -04:00

2965 lines
106 KiB
C++
Raw Blame History

// Copyright (c) 2014-2026 Savoir-faire Linux Inc.
// SPDX-License-Identifier: MIT
#include "dht.h"
#include "rng.h"
#include "search.h"
#include "storage.h"
#include "request.h"
#include <msgpack.hpp>
#include <algorithm>
#include <random>
#include <sstream>
#include <fstream>
#include <inttypes.h>
namespace dht {
using namespace std::placeholders;
constexpr std::chrono::minutes Dht::MAX_STORAGE_MAINTENANCE_EXPIRE_TIME;
constexpr std::chrono::minutes Dht::SEARCH_EXPIRE_TIME;
constexpr std::chrono::seconds Dht::BOOTSTRAP_PERIOD;
constexpr duration Dht::LISTEN_EXPIRE_TIME;
constexpr duration Dht::LISTEN_EXPIRE_TIME_PUBLIC;
constexpr duration Dht::REANNOUNCE_MARGIN;
static constexpr size_t MAX_REQUESTS_PER_SEC {8 * 1024};
static constexpr duration BOOTSTRAP_PERIOD_MAX {std::chrono::hours(24)};
NodeStatus
Dht::updateStatus(sa_family_t af)
{
auto& d = dht(af);
auto old = d.status;
d.status = d.getStatus(scheduler.time());
if (d.status == NodeStatus::Disconnected && bootstrap_pending && bootstrapJob && !bootstrap_nodes.empty())
d.status = NodeStatus::Connecting;
if (d.status != old) {
auto& other = dht(af == AF_INET ? AF_INET6 : AF_INET);
if (other.status == NodeStatus::Disconnected && d.status == NodeStatus::Disconnected) {
onDisconnected();
} else if (other.status == NodeStatus::Connected || d.status == NodeStatus::Connected) {
onConnected();
}
}
return d.status;
}
NodeStatus
Dht::Kad::getStatus(time_point now) const
{
unsigned dubious = 0;
for (const auto& b : buckets) {
for (auto& n : b.nodes) {
if (n->isGood(now)) {
return NodeStatus::Connected;
} else if (not n->isExpired())
dubious++;
}
}
auto& ping = pending_pings;
if (dubious or ping)
return NodeStatus::Connecting;
return NodeStatus::Disconnected;
}
void
Dht::shutdown(ShutdownCallback cb, bool stop)
{
if (not persistPath.empty())
saveState(persistPath);
if (stop) {
for (auto dht : {&dht4, &dht6}) {
for (auto& sr : dht->searches) {
for (const auto& r : sr.second->callbacks)
r.second.done_cb(false, {});
sr.second->callbacks.clear();
for (const auto& a : sr.second->announce) {
for (auto& cb : a.callbacks)
cb(false, {});
}
sr.second->announce.clear();
sr.second->listeners.clear();
}
}
network_engine.clear();
}
if (not maintain_storage) {
if (cb)
cb();
return;
}
// Last store maintenance
scheduler.syncTime();
auto remaining = std::make_shared<int>(0);
auto str_donecb = [=](bool, const std::vector<Sp<Node>>&) {
--*remaining;
if (logger_)
logger_->warn("Shutting down node: {} ops remaining", *remaining);
if (!*remaining && cb) {
cb();
}
};
for (auto& str : store)
*remaining += maintainStorage(str, true, str_donecb);
if (logger_)
logger_->warn("Shutting down node: after storage, {} ops", *remaining);
if (!*remaining) {
if (cb)
cb();
}
}
bool
Dht::isRunning(sa_family_t af) const
{
return network_engine.isRunning(af);
}
/* Every bucket contains an unordered list of nodes. */
const Sp<Node>
Dht::findNode(const InfoHash& id, sa_family_t af) const
{
if (const Bucket* b = findBucket(id, af))
for (const auto& n : b->nodes)
if (n->id == id)
return n;
return {};
}
/* Every bucket caches the address of a likely node. Ping it. */
void
Dht::sendCachedPing(Bucket& b)
{
if (b.cached)
if (logger_)
logger_->debug("[node {}] Sending ping to cached node", b.cached->toString());
b.sendCachedPing(network_engine);
}
std::vector<SockAddr>
Dht::getPublicAddress(sa_family_t family)
{
std::vector<SockAddr> ret;
if (family == AF_UNSPEC) {
auto& d4 = dht(AF_INET).reported_addr;
auto& d6 = dht(AF_INET6).reported_addr;
ret.reserve(d4.size() + d6.size());
for (const auto& a : d4)
ret.emplace_back(a.second);
for (const auto& a : d6)
ret.emplace_back(a.second);
return ret;
}
auto& reported_addr = dht(family).reported_addr;
ret.reserve(!family ? reported_addr.size() : reported_addr.size() / 2);
for (const auto& addr : reported_addr)
if (!family || family == addr.second.getFamily())
ret.emplace_back(addr.second);
return ret;
}
std::pair<size_t, size_t>
Dht::getLocalStoreSize() const
{
return {local_store_quota->size(), local_store_quota->valueCount()};
}
bool
Dht::trySearchInsert(const Sp<Node>& node)
{
const auto& now = scheduler.time();
if (not node)
return false;
auto& srs = searches(node->getFamily());
auto closest = srs.lower_bound(node->id);
bool inserted {false};
// insert forward
for (auto it = closest; it != srs.end(); it++) {
auto& s = *it->second;
if (s.insertNode(node, now)) {
inserted = true;
scheduler.edit(s.nextSearchStep, now);
} else if (not s.expired and not s.done)
break;
}
// insert backward
for (auto it = closest; it != srs.begin();) {
--it;
auto& s = *it->second;
if (s.insertNode(node, now)) {
inserted = true;
scheduler.edit(s.nextSearchStep, now);
} else if (not s.expired and not s.done)
break;
}
return inserted;
}
void
Dht::reportedAddr(const SockAddr& addr)
{
auto& reported_addr = dht(addr.getFamily()).reported_addr;
auto firstBefore = reported_addr.empty() ? nullptr : reported_addr.begin()->second.get();
auto it = std::find_if(reported_addr.begin(), reported_addr.end(), [&](const ReportedAddr& a) {
return a.second == addr;
});
if (it == reported_addr.end()) {
if (reported_addr.size() < 32)
reported_addr.emplace_back(1, addr);
} else
it->first++;
std::sort(reported_addr.begin(), reported_addr.end(), [](const ReportedAddr& a, const ReportedAddr& b) {
return a.first > b.first;
});
if (publicAddressChangedCb_) {
auto firstAfter = reported_addr.begin()->second.get();
if (firstBefore != firstAfter) {
auto& otherDht = dht(addr.getFamily() == AF_INET ? AF_INET6 : AF_INET).reported_addr;
std::vector<SockAddr> v;
v.reserve(otherDht.empty() ? 1 : 2);
v.emplace_back(reported_addr.begin()->second);
if (not otherDht.empty())
v.emplace_back(otherDht.begin()->second);
publicAddressChangedCb_(std::move(v));
}
}
}
/* We just learnt about a node, not necessarily a new one. Confirm is 1 if
the node sent a message, 2 if it sent us a reply. */
void
Dht::onNewNode(const Sp<Node>& node, int confirm)
{
const auto& now = scheduler.time();
auto& b = buckets(node->getFamily());
auto wasEmpty = confirm < 2 && b.grow_time < now - std::chrono::minutes(5);
if (b.onNewNode(node, confirm, now, myid, network_engine) or confirm) {
trySearchInsert(node);
if (wasEmpty) {
scheduler.edit(nextNodesConfirmation, now + std::chrono::seconds(1));
}
}
}
/* Called periodically to purge known-bad nodes. Note that we're very
conservative here: broken nodes in the table don't do much harm, we'll
recover as soon as we find better ones. */
void
Dht::expireBuckets(RoutingTable& list)
{
for (auto& b : list) {
bool changed = false;
b.nodes.remove_if([&changed](const Sp<Node>& n) {
if (n->isExpired()) {
changed = true;
return true;
}
return false;
});
if (changed)
sendCachedPing(b);
}
}
void
Dht::expireSearches()
{
auto t = scheduler.time() - SEARCH_EXPIRE_TIME;
auto expired = [&](std::pair<const InfoHash, Sp<Search>>& srp) {
auto& sr = *srp.second;
auto b = sr.callbacks.empty() && sr.announce.empty() && sr.listeners.empty() && sr.step_time < t;
if (b) {
if (logger_)
logger_->debug("[search {}] Removing search", srp.first.to_view());
sr.clear();
return b;
} else {
return false;
}
};
erase_if(dht4.searches, expired);
erase_if(dht6.searches, expired);
}
void
Dht::searchNodeGetDone(const net::Request& req, net::RequestAnswer&& answer, std::weak_ptr<Search> ws, Sp<Query> query)
{
const auto& now = scheduler.time();
if (auto sr = ws.lock()) {
sr->insertNode(req.node, now, answer.ntoken);
if (auto srn = sr->getNode(req.node)) {
/* all other get requests which are satisfied by this answer
should not be sent anymore */
for (auto& g : sr->callbacks) {
auto& q = g.second.query;
if (q != query and g.second.isSatisfiedBy(*query)) {
auto dummy_req = std::make_shared<net::Request>();
dummy_req->cancel();
srn->getStatus[q] = std::move(dummy_req);
}
}
auto syncTime = srn->getSyncTime(scheduler.time());
if (srn->syncJob)
scheduler.edit(srn->syncJob, syncTime);
else
srn->syncJob = scheduler.add(syncTime, std::bind(&Dht::searchStep, this, ws));
}
onGetValuesDone(req.node, answer, sr, query);
}
}
void
Dht::searchNodeGetExpired(const net::Request& status, bool over, std::weak_ptr<Search> ws, Sp<Query> query)
{
if (auto sr = ws.lock()) {
if (auto srn = sr->getNode(status.node)) {
srn->candidate = not over;
if (over)
srn->getStatus.erase(query);
}
scheduler.edit(sr->nextSearchStep, scheduler.time());
}
}
void
Dht::paginate(std::weak_ptr<Search> ws, Sp<Query> query, SearchNode* n)
{
auto sr = ws.lock();
if (not sr)
return;
auto select_q = std::make_shared<Query>(Select {}.field(Value::Field::Id), query ? query->where : Where {});
auto onSelectDone = [this, ws, query](const net::Request& status, net::RequestAnswer&& answer) mutable {
// Retrieve search
auto sr = ws.lock();
if (not sr)
return;
const auto& id = sr->id;
// Retrieve search node
auto sn = sr->getNode(status.node);
if (not sn)
return;
// backward compatibility
if (answer.fields.empty()) {
searchNodeGetDone(status, std::move(answer), ws, query);
return;
}
for (const auto& fvi : answer.fields) {
try {
auto vid = fvi->index.at(Value::Field::Id).getInt();
if (vid == Value::INVALID_ID)
continue;
auto query_for_vid = std::make_shared<Query>(Select {}, Where {}.id(vid));
sn->pagination_queries[query].push_back(query_for_vid);
if (logger_)
logger_->debug("[search {}] [node {}] Sending {}",
id.toString(),
sn->node->toString(),
query_for_vid->toString());
sn->getStatus[query_for_vid] = network_engine.sendGetValues(
status.node,
id,
*query_for_vid,
-1,
std::bind(&Dht::searchNodeGetDone, this, _1, _2, ws, query),
std::bind(&Dht::searchNodeGetExpired, this, _1, _2, ws, query_for_vid));
} catch (const std::out_of_range&) {
if (logger_)
logger_->error("[search {}] [node {}] Received non-id field in response to "
"'SELECT id' request…",
id.toString(),
sn->node->toString());
}
}
};
/* add pagination query key for tracking ongoing requests. */
n->pagination_queries[query].push_back(select_q);
if (logger_)
logger_->debug("[search {}] [node {}] Sending {}", sr->id.toString(), n->node->toString(), select_q->toString());
n->getStatus[select_q] = network_engine.sendGetValues(
n->node, sr->id, *select_q, -1, onSelectDone, std::bind(&Dht::searchNodeGetExpired, this, _1, _2, ws, select_q));
}
Dht::SearchNode*
Dht::searchSendGetValues(Sp<Search> sr, SearchNode* pn, bool update)
{
if (sr->done or sr->currentlySolicitedNodeCount() >= MAX_REQUESTED_SEARCH_NODES)
return nullptr;
const auto& now = scheduler.time();
std::weak_ptr<Search> ws = sr;
auto cb = sr->callbacks.begin();
static const auto ANY_QUERY = std::make_shared<Query>(Select {}, Where {}, true);
do { /* for all requests to send */
SearchNode* n = nullptr;
auto& query = sr->callbacks.empty() ? ANY_QUERY : cb->second.query;
const time_point up = (not sr->callbacks.empty() and update) ? sr->getLastGetTime(*query) : time_point::min();
if (pn and pn->canGet(now, up, query)) {
n = pn;
} else {
for (auto& sn : sr->nodes) {
if (sn->canGet(now, up, query)) {
n = sn.get();
break;
}
}
}
if (sr->callbacks.empty()) { /* 'find_node' request */
if (not n)
return nullptr;
/* if (logger_)
logger_->d(sr->id, n->node->id, "[search %s] [node %s] Sending 'find_node'",
sr->id.toString().c_str(), n->node->toString().c_str());*/
n->getStatus[query] = network_engine
.sendFindNode(n->node,
sr->id,
-1,
std::bind(&Dht::searchNodeGetDone, this, _1, _2, ws, query),
std::bind(&Dht::searchNodeGetExpired, this, _1, _2, ws, query));
} else { /* 'get' request */
if (not n)
continue;
if (query and not query->select.empty()) {
/* The request contains a select. No need to paginate… */
/* if (logger_)
logger_->d(sr->id, n->node->id, "[search %s] [node %s] Sending 'get'",
sr->id.toString().c_str(), n->node->toString().c_str());*/
n->getStatus[query]
= network_engine.sendGetValues(n->node,
sr->id,
*query,
-1,
std::bind(&Dht::searchNodeGetDone, this, _1, _2, ws, query),
std::bind(&Dht::searchNodeGetExpired, this, _1, _2, ws, query));
} else
paginate(ws, query, n);
}
/* We only attempt to send one request. Return. */
return n;
} while (++cb != sr->callbacks.end());
/* no request were sent */
return nullptr;
}
void
Dht::searchSendAnnounceValue(const Sp<Search>& sr, unsigned syncLevel)
{
if (sr->announce.empty())
return;
std::weak_ptr<Search> ws = sr;
auto onDone = [this, ws](const net::Request& req,
net::RequestAnswer&& answer) { /* when put done */
if (auto sr = ws.lock()) {
onAnnounceDone(req.node, answer, sr);
scheduler.edit(sr->nextSearchStep, scheduler.time());
}
};
auto onExpired = [this, ws](const net::Request&, bool over) { /* when put expired */
if (over)
if (auto sr = ws.lock())
scheduler.edit(sr->nextSearchStep,
scheduler.time());
};
auto onSelectDone = [this, ws, onDone, onExpired](const net::Request& req, net::RequestAnswer&& answer) mutable {
/* on probing done */
auto sr = ws.lock();
if (not sr)
return;
const auto& now = scheduler.time();
sr->insertNode(req.node, now, answer.ntoken);
auto sn = sr->getNode(req.node);
if (not sn)
return;
if (not sn->isSynced(now)) {
/* Search is now unsynced. Let's call searchStep to sync again. */
scheduler.edit(sr->nextSearchStep, now);
return;
}
for (auto& a : sr->announce) {
if (sn->getAnnounceTime(a.value->id) > now)
continue;
bool hasValue {false};
uint16_t seq_no = 0;
try {
const auto& f = std::find_if(answer.fields.cbegin(),
answer.fields.cend(),
[&a](const Sp<FieldValueIndex>& i) {
return i->index.at(Value::Field::Id).getInt() == a.value->id;
});
if (f != answer.fields.cend() and *f) {
hasValue = true;
seq_no = static_cast<uint16_t>((*f)->index.at(Value::Field::SeqNum).getInt());
}
} catch (std::out_of_range&) {
}
auto next_refresh_time = now + getType(a.value->type).expiration;
auto& acked = sn->acked[a.value->id];
scheduler.cancel(acked.refresh);
/* only put the value if the node doesn't already have it */
if (not hasValue or seq_no < a.value->seq) {
if (logger_)
logger_->debug("[search {}] [node {}] Sending 'put' (vid: {:016x})",
sr->id.toString(),
sn->node->toString(),
a.value->id);
auto created = a.permanent ? time_point::max() : a.created;
acked = {network_engine
.sendAnnounceValue(sn->node, sr->id, a.value, created, sn->token, onDone, onExpired),
next_refresh_time};
} else if (hasValue and a.permanent) {
if (logger_)
logger_->warn("[search {}] [node {}] Sending 'refresh' (vid: {:016x})",
sr->id.toString(),
sn->node->toString(),
a.value->id);
acked = {network_engine.sendRefreshValue(
sn->node,
sr->id,
a.value->id,
sn->token,
onDone,
[this,
ws,
node = sn->node,
v = a.value,
onDone,
onExpired,
created = a.permanent ? time_point::max() : a.created,
next_refresh_time](const net::Request& /*req*/, net::DhtProtocolException&& e) {
if (e.getCode() == net::DhtProtocolException::NOT_FOUND) {
if (logger_)
logger_->error("[node {}] Returned error 404: storage not found",
node->toString());
if (auto sr = ws.lock()) {
if (auto sn = sr->getNode(node)) {
sn->acked[v->id] = {network_engine.sendAnnounceValue(sn->node,
sr->id,
v,
created,
sn->token,
onDone,
onExpired),
next_refresh_time};
scheduler.edit(sr->nextSearchStep, scheduler.time());
return true;
}
}
}
return false;
},
onExpired),
next_refresh_time};
} else {
if (logger_)
logger_->warn("[search {}] [node {}] Already has value (vid: {:016x}). Aborting.",
sr->id.toString(),
sn->node->toString(),
a.value->id);
auto ack_req = std::make_shared<net::Request>(net::Request::State::COMPLETED);
ack_req->reply_time = now;
acked = {std::move(ack_req), next_refresh_time};
/* step to clear announces */
scheduler.edit(sr->nextSearchStep, now);
}
if (a.permanent) {
acked.refresh = scheduler.add(next_refresh_time - REANNOUNCE_MARGIN,
std::bind(&Dht::searchStep, this, ws));
}
}
};
static const auto PROBE_QUERY = std::make_shared<Query>(
Select {}.field(Value::Field::Id).field(Value::Field::SeqNum));
const auto& now = scheduler.time();
unsigned i = 0;
for (auto& np : sr->nodes) {
auto& n = *np;
if (not n.isSynced(now))
continue;
auto gs = n.probe_query ? n.getStatus.find(n.probe_query) : n.getStatus.end();
if (gs != n.getStatus.end() 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 {
if (logger_)
logger_->warn("[search {}] [node {}] Sending 'put' (vid: {:016x})",
sr->id.toString(),
n.node->toString(),
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) {
n.probe_query = PROBE_QUERY;
if (logger_)
logger_->debug("[search {}] [node {}] Sending {}",
sr->id.toString(),
n.node->toString(),
n.probe_query->toString());
auto req = network_engine.sendGetValues(n.node,
sr->id,
*PROBE_QUERY,
-1,
onSelectDone,
std::bind(&Dht::searchNodeGetExpired, this, _1, _2, ws, PROBE_QUERY));
n.getStatus[PROBE_QUERY] = std::move(req);
}
if (not n.candidate and ++i == syncLevel)
break;
}
}
void
Dht::searchSynchedNodeListen(const Sp<Search>& sr, SearchNode& n)
{
const auto& listenExp = getListenExpiration();
std::weak_ptr<Search> ws = sr;
for (const auto& l : sr->listeners) {
const auto& query = l.second.query;
auto openListenSocket = [this, ws, query, node = n.node] {
return node->openSocket([this, ws, query](const Sp<Node>& node, net::RequestAnswer&& answer) mutable {
/* on new values */
if (auto sr = ws.lock()) {
scheduler.edit(sr->nextSearchStep, scheduler.time());
if (auto sn = sr->getNode(node)) {
if (not answer.ntoken.empty())
sn->refreshToken(answer.ntoken);
sn->onValues(query, std::move(answer), types, scheduler);
}
}
});
};
auto r = n.listenStatus.find(query);
if (n.getListenTime(r, listenExp) > scheduler.time())
continue;
// if (logger_)
// logger_->d(sr->id, n.node->id, "[search %s] [node %s] Sending 'listen'",
// sr->id.toString().c_str(), n.node->toString().c_str());
if (r == n.listenStatus.end()) {
r = n.listenStatus
.emplace(std::piecewise_construct,
std::forward_as_tuple(query),
std::forward_as_tuple(l.second.get_cb, l.second.sync_cb, openListenSocket()))
.first;
r->second.cacheExpirationJob = scheduler.add(time_point::max(), [this, ws, query, node = n.node] {
if (auto sr = ws.lock()) {
if (auto sn = sr->getNode(node)) {
sn->expireValues(query, scheduler);
}
}
});
} else if (not n.node->getSocket(r->second.socketId)) {
r->second.socketId = openListenSocket();
}
auto new_req = network_engine.sendListen(
n.node,
sr->id,
*query,
n.token,
r->second.socketId,
[this, ws, query](const net::Request& req, net::RequestAnswer&& answer) mutable {
/* on done */
if (auto sr = ws.lock()) {
scheduler.edit(sr->nextSearchStep, scheduler.time());
if (auto sn = sr->getNode(req.node)) {
auto job = scheduler.add(sn->getListenTime(query, getListenExpiration()),
std::bind(&Dht::searchStep, this, ws));
sn->onListenSynced(query, true, std::move(job));
}
onListenDone(req.node, answer, sr);
}
},
[this, ws, query](const net::Request& req, bool over) mutable {
/* on request expired */
if (auto sr = ws.lock()) {
scheduler.edit(sr->nextSearchStep, scheduler.time());
if (over)
if (auto sn = sr->getNode(req.node))
sn->listenStatus.erase(query);
}
});
// Here the request may have failed and the CachedListenStatus removed
r = n.listenStatus.find(query);
if (r != n.listenStatus.end()) {
r->second.req = new_req;
}
}
}
/* When a search is in progress, we periodically call search_step to send
further requests. */
void
Dht::searchStep(std::weak_ptr<Search> ws)
{
auto sr = ws.lock();
if (not sr or sr->expired or sr->done)
return;
const auto& now = scheduler.time();
auto level = sr->syncLevel(now);
constexpr auto MARGIN = 1;
bool preSynced = level > MARGIN;
auto syncLevel = preSynced ? level - MARGIN : 0;
/*if (auto req_count = sr->currentlySolicitedNodeCount())
if (logger_)
logger_->d(sr->id, "[search %s IPv%c] Step (%d requests)",
sr->id.toString().c_str(), sr->af == AF_INET ? '4' : '6', req_count);*/
sr->step_time = now;
if (sr->refill_time + Node::NODE_EXPIRE_TIME < now and sr->nodes.size() - sr->getNumberOfBadNodes() < SEARCH_NODES)
refill(*sr);
/* Check if the first TARGET_NODES (8) live nodes have replied. */
if (sr->isSynced(now)) {
if (not(sr->callbacks.empty() and sr->announce.empty())) {
// search is synced but some (newer) get operations are not complete
// Call callbacks when done
std::vector<Get> completed_gets;
for (auto b = sr->callbacks.begin(); b != sr->callbacks.end();) {
if (sr->isDone(b->second)) {
sr->setDone(b->second);
completed_gets.emplace_back(std::move(b->second));
b = sr->callbacks.erase(b);
} else
++b;
}
// clear corresponding queries
for (const auto& get : completed_gets)
for (auto& sn : sr->nodes) {
sn->getStatus.erase(get.query);
sn->pagination_queries.erase(get.query);
}
/* clearing callbacks for announced values */
sr->checkAnnounced();
if (sr->callbacks.empty() && sr->announce.empty() && sr->listeners.empty())
sr->setDone();
}
// true if this node is part of the target nodes cluter.
/*bool in = sr->id.xorCmp(myid, sr->nodes.back().node->id) < 0;
logger__DBG("[search %s IPv%c] synced%s",
sr->id.toString().c_str(), sr->af == AF_INET ? '4' : '6', in ? ", in" : "");*/
if (not sr->listeners.empty()) {
unsigned i = 0;
for (auto& n : sr->nodes) {
if (not n->isSynced(now))
continue;
searchSynchedNodeListen(sr, *n);
if (not n->candidate and ++i == LISTEN_NODES)
break;
}
}
// Announce requests
searchSendAnnounceValue(sr);
if (sr->callbacks.empty() && sr->announce.empty() && sr->listeners.empty())
sr->setDone();
} else if (preSynced) {
if (not sr->listeners.empty()) {
auto nl = std::min(syncLevel, LISTEN_NODES);
unsigned i = 0;
for (auto& n : sr->nodes) {
if (not n->isSynced(now))
break;
searchSynchedNodeListen(sr, *n);
if (++i == nl)
break;
}
}
// Announce requests
searchSendAnnounceValue(sr, syncLevel);
}
while (sr->currentlySolicitedNodeCount() < MAX_REQUESTED_SEARCH_NODES and searchSendGetValues(sr))
;
if (sr->getNumberOfConsecutiveBadNodes() >= std::min<size_t>(sr->nodes.size(), SEARCH_MAX_BAD_NODES)) {
if (logger_)
logger_->warn("[search {} IPv{}] Expired", sr->id.to_view(), sr->af == AF_INET ? '4' : '6');
sr->expire();
if (not public_stable)
connectivityChanged(sr->af);
}
/* dumpSearch(*sr, std::cout); */
}
unsigned
Dht::refill(Dht::Search& sr)
{
const auto& now = scheduler.time();
sr.refill_time = now;
/* we search for up to SEARCH_NODES good nodes. */
auto cached_nodes = network_engine.getCachedNodes(sr.id, sr.af, SEARCH_NODES);
if (cached_nodes.empty()) {
if (logger_)
logger_->error("[search {} IPv{}] No nodes from cache while refilling search",
sr.id.toString(),
(sr.af == AF_INET) ? '4' : '6');
return 0;
}
unsigned inserted = 0;
for (auto& i : cached_nodes) {
/* try to insert the nodes. Search::insertNode will know how many to insert. */
if (sr.insertNode(i, now))
++inserted;
}
if (logger_)
logger_->debug("[search {} IPv{}] Refilled search with {} nodes from node cache",
sr.id.toString(),
(sr.af == AF_INET) ? '4' : '6',
inserted);
return inserted;
}
/* Start a search. */
Sp<Dht::Search>
Dht::search(const InfoHash& id,
sa_family_t af,
GetCallback gcb,
QueryCallback qcb,
DoneCallback dcb,
Value::Filter f,
const Sp<Query>& q)
{
if (!isRunning(af)) {
if (logger_)
logger_->error("[search {} IPv{}] Unsupported protocol", id.to_view(), (af == AF_INET) ? '4' : '6');
if (dcb)
dcb(false, {});
return {};
}
auto& srs = searches(af);
const auto& srp = srs.find(id);
Sp<Search> sr {};
if (srp != srs.end()) {
sr = srp->second;
sr->done = false;
sr->expired = false;
} else {
if (srs.size() < max_searches) {
sr = std::make_shared<Search>();
srs.emplace(id, sr);
} else {
for (auto it = srs.begin(); it != srs.end(); ++it) {
auto& s = *it->second;
if ((s.done or s.expired) and s.announce.empty() and s.listeners.empty()) {
sr = it->second;
if (sr->nextSearchStep)
scheduler.cancel(sr->nextSearchStep);
if (sr->opExpirationJob)
scheduler.cancel(sr->opExpirationJob);
sr->clear();
srs.erase(it);
break;
}
}
if (not sr) {
if (logger_)
logger_->error("[search {} IPv{}] Maximum number of searches reached!",
id.toString(),
(af == AF_INET) ? '4' : '6');
if (dcb)
dcb(false, {});
return {};
}
srs.emplace(id, sr);
}
sr->af = af;
sr->tid = search_id++;
sr->step_time = time_point::min();
sr->id = id;
sr->done = false;
sr->expired = false;
sr->nodes.clear();
sr->nodes.reserve(SEARCH_NODES + 1);
sr->nextSearchStep = scheduler.add(time_point::max(),
std::bind(&Dht::searchStep, this, std::weak_ptr<Search>(sr)));
if (logger_)
logger_->warn("[search {} IPv{}] New search", id.to_view(), (af == AF_INET) ? '4' : '6');
if (search_id == 0)
search_id++;
}
sr->get(f, q, qcb, gcb, dcb, scheduler);
refill(*sr);
return sr;
}
void
Dht::announce(
const InfoHash& id, sa_family_t af, Sp<Value> value, DoneCallback callback, time_point created, bool permanent)
{
auto& srs = searches(af);
auto srp = srs.find(id);
if (auto sr = srp == srs.end() ? search(id, af) : srp->second) {
sr->put(value, callback, created, permanent);
scheduler.edit(sr->nextSearchStep, scheduler.time());
} else if (callback) {
callback(false, {});
}
}
size_t
Dht::listenTo(const InfoHash& id, sa_family_t af, ValueCallback cb, Value::Filter f, const Sp<Query>& q)
{
if (!isRunning(af))
return 0;
// logger__ERR("[search %s IPv%c] Search_time is now in %lfs", sr->id.toString().c_str(), (sr->af == AF_INET) ? '4'
// : '6', print_duration(tm-clock::now()));
// logger__WARN("listenTo %s", id.toString().c_str());
auto& srs = searches(af);
auto srp = srs.find(id);
Sp<Search> sr = (srp == srs.end()) ? search(id, af) : srp->second;
if (!sr)
throw DhtException("Unable to create search");
if (logger_)
logger_->warn("[search {} IPv{}] Listen", id.to_view(), (af == AF_INET) ? '4' : '6');
return sr->listen(cb, std::move(f), q, scheduler);
}
size_t
Dht::listen(const InfoHash& id, ValueCallback cb, Value::Filter f, Where where)
{
if (not id) {
if (logger_)
logger_->warn("Listen called with invalid key");
return 0;
}
scheduler.syncTime();
auto token = ++listener_token;
auto gcb = OpValueCache::cacheCallback(std::move(cb), [this, id, token] { cancelListen(id, token); });
auto query = std::make_shared<Query>(Select {}, std::move(where));
auto filter = Value::Filter::chain(std::move(f), query->where.getFilter());
auto st = store.find(id);
if (st == store.end() && store.size() < max_store_keys)
st = store.emplace(id, scheduler.time() + MAX_STORAGE_MAINTENANCE_EXPIRE_TIME).first;
size_t tokenlocal = 0;
if (st != store.end()) {
tokenlocal = st->second.listen(gcb, filter, query);
if (tokenlocal == 0)
return 0;
}
auto token4 = Dht::listenTo(id, AF_INET, gcb, filter, query);
auto token6 = token4 == 0 ? 0 : Dht::listenTo(id, AF_INET6, gcb, filter, query);
if (token6 == 0 && st != store.end()) {
st->second.cancelListen(tokenlocal);
return 0;
}
listeners.emplace(token, std::make_tuple(tokenlocal, token4, token6));
return token;
}
bool
Dht::cancelListen(const InfoHash& id, size_t token)
{
scheduler.syncTime();
auto it = listeners.find(token);
if (it == listeners.end()) {
if (logger_)
logger_->warn("Listen token not found: {}", token);
return false;
}
if (logger_)
logger_->debug("cancelListen {} with token {}", id.to_view(), token);
if (auto tokenlocal = std::get<0>(it->second)) {
auto st = store.find(id);
if (st != store.end())
st->second.cancelListen(tokenlocal);
}
auto searches_cancel_listen = [this, &id](std::map<InfoHash, Sp<Search>>& srs, size_t token) {
if (token) {
auto srp = srs.find(id);
if (srp != srs.end())
srp->second->cancelListen(token, scheduler);
}
};
searches_cancel_listen(dht4.searches, std::get<1>(it->second));
searches_cancel_listen(dht6.searches, std::get<2>(it->second));
listeners.erase(it);
return true;
}
struct OpStatus
{
struct Status
{
bool done {false};
bool ok {false};
Status(bool done = false, bool ok = false)
: done(done)
, ok(ok)
{}
};
Status status;
Status status4;
Status status6;
};
template<typename T>
struct GetStatus : public OpStatus
{
T values;
std::vector<Sp<Node>> nodes;
};
void
Dht::put(const InfoHash& id, Sp<Value> val, DoneCallback callback, time_point created, bool permanent)
{
if (not id or not val) {
if (logger_)
logger_->warn("Put called with invalid key or value");
if (callback)
callback(false, {});
return;
}
if (val->id == Value::INVALID_ID)
val->id = std::uniform_int_distribution<Value::Id> {1}(rd);
scheduler.syncTime();
const auto& now = scheduler.time();
created = std::min(now, created);
storageStore(id, val, created, {}, permanent);
if (logger_)
logger_->debug("put: adding {} → {}", id.to_view(), val->toString());
auto op = std::make_shared<OpStatus>();
auto donecb = [callback](const std::vector<Sp<Node>>& nodes, OpStatus& op) {
// Callback as soon as the value is announced on one of the available networks
if (callback and not op.status.done and (op.status4.done && op.status6.done)) {
callback(op.status4.ok or op.status6.ok, nodes);
op.status.done = true;
}
};
announce(
id,
AF_INET,
val,
[=](bool ok4, const std::vector<Sp<Node>>& nodes) {
if (logger_)
logger_->debug("Announce done IPv4 {}", ok4);
auto& o = *op;
o.status4 = {true, ok4};
donecb(nodes, o);
},
created,
permanent);
announce(
id,
AF_INET6,
val,
[=](bool ok6, const std::vector<Sp<Node>>& nodes) {
if (logger_)
logger_->debug("Announce done IPv6 {}", ok6);
auto& o = *op;
o.status6 = {true, ok6};
donecb(nodes, o);
},
created,
permanent);
}
template<typename T>
void
doneCallbackWrapper(DoneCallback dcb, const std::vector<Sp<Node>>& nodes, GetStatus<T>& op)
{
if (op.status.done)
return;
op.nodes.insert(op.nodes.end(), nodes.begin(), nodes.end());
if (op.status.ok or (op.status4.done and op.status6.done)) {
bool ok = op.status.ok or op.status4.ok or op.status6.ok;
op.status.done = true;
if (dcb)
dcb(ok, op.nodes);
}
}
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)
{
if (op.status.done)
return false;
auto newvals = add_values(values);
if (not newvals.empty()) {
op.status.ok = !get_cb(newvals);
cache_values(newvals);
}
doneCallbackWrapper(done_cb, {}, op);
return !op.status.ok;
}
void
Dht::get(const InfoHash& id, GetCallback getcb, DoneCallback donecb, Value::Filter&& filter, Where&& where)
{
if (not id) {
if (logger_)
logger_->warn("Get called with invalid key");
if (donecb)
donecb(false, {});
return;
}
scheduler.syncTime();
auto op = std::make_shared<GetStatus<std::map<Value::Id, Sp<Value>>>>();
auto gcb = [getcb, donecb, op](const std::vector<Sp<Value>>& vals) {
auto& o = *op;
return callbackWrapper(
getcb,
donecb,
vals,
[&o](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))) {
newvals.push_back(v);
}
}
return newvals;
},
[&o](const std::vector<Sp<Value>>& newvals) {
for (const auto& v : newvals)
o.values[v->id] = v;
},
o);
};
auto q = std::make_shared<Query>(Select {}, std::move(where));
auto f = filter.chain(q->where.getFilter());
/* Try to answer this search locally. */
gcb(getLocal(id, f));
Dht::search(
id,
AF_INET,
gcb,
{},
[=](bool ok, const std::vector<Sp<Node>>& nodes) {
// logger__WARN("DHT done IPv4");
op->status4 = {true, ok};
doneCallbackWrapper(donecb, nodes, *op);
},
f,
q);
Dht::search(
id,
AF_INET6,
gcb,
{},
[=](bool ok, const std::vector<Sp<Node>>& nodes) {
// logger__WARN("DHT done IPv6");
op->status6 = {true, ok};
doneCallbackWrapper(donecb, nodes, *op);
},
f,
q);
}
void
Dht::query(const InfoHash& id, QueryCallback cb, DoneCallback done_cb, Query&& q)
{
if (not id) {
if (logger_)
logger_->warn("Query called with invalid key");
if (done_cb)
done_cb(false, {});
return;
}
scheduler.syncTime();
auto op = std::make_shared<GetStatus<std::vector<Sp<FieldValueIndex>>>>();
auto f = q.where.getFilter();
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 == f or f->containedIn(*sf);
});
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;
},
[&](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);
});
qcb(local_fields);
auto sq = std::make_shared<Query>(std::move(q));
Dht::search(
id,
AF_INET,
{},
qcb,
[=](bool ok, const std::vector<Sp<Node>>& nodes) {
// logger__WARN("DHT done IPv4");
op->status4 = {true, ok};
doneCallbackWrapper(done_cb, nodes, *op);
},
f,
sq);
Dht::search(
id,
AF_INET6,
{},
qcb,
[=](bool ok, const std::vector<Sp<Node>>& nodes) {
// logger__WARN("DHT done IPv6");
op->status6 = {true, ok};
doneCallbackWrapper(done_cb, nodes, *op);
},
f,
sq);
}
std::vector<Sp<Value>>
Dht::getLocal(const InfoHash& id, const Value::Filter& f) const
{
auto s = store.find(id);
if (s == store.end())
return {};
return s->second.get(f);
}
Sp<Value>
Dht::getLocalById(const InfoHash& id, Value::Id vid) const
{
auto s = store.find(id);
if (s != store.end())
return s->second.getById(vid);
return {};
}
std::vector<Sp<Value>>
Dht::getPut(const InfoHash& id) const
{
std::vector<Sp<Value>> ret;
auto find_values = [&](const std::map<InfoHash, Sp<Search>>& srs) {
auto srp = srs.find(id);
if (srp == srs.end())
return;
auto vals = srp->second->getPut();
ret.insert(ret.end(), vals.begin(), vals.end());
};
find_values(dht4.searches);
find_values(dht6.searches);
return ret;
}
Sp<Value>
Dht::getPut(const InfoHash& id, const Value::Id& vid) const
{
auto find_value = [&](const std::map<InfoHash, Sp<Search>>& srs) {
auto srp = srs.find(id);
return (srp != srs.end()) ? srp->second->getPut(vid) : Sp<Value> {};
};
if (auto v4 = find_value(dht4.searches))
return v4;
if (auto v6 = find_value(dht6.searches))
return v6;
return {};
}
bool
Dht::cancelPut(const InfoHash& id, const Value::Id& vid)
{
bool canceled {false};
auto sr_cancel_put = [&](std::map<InfoHash, Sp<Search>>& srs) {
auto srp = srs.find(id);
return (srp != srs.end()) ? srp->second->cancelPut(vid) : false;
};
canceled |= sr_cancel_put(dht4.searches);
canceled |= sr_cancel_put(dht6.searches);
if (canceled) {
auto st = store.find(id);
if (st != store.end()) {
if (auto value = st->second.remove(id, vid))
storageRemoved(id, st->second, {value}, value->size());
}
}
return canceled;
}
// Storage
void
Dht::storageChanged(const InfoHash& id, Storage& st, const Sp<Value>& v, bool newValue)
{
if (newValue) {
if (not st.local_listeners.empty()) {
if (logger_)
logger_->debug("[store {}] {} local listeners", id.to_view(), 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))
vals.push_back(v);
if (not vals.empty()) {
if (logger_)
logger_->debug("[store {}] Sending update local listener with token {}", id.to_view(), l.first);
cbs.emplace_back(l.second.get_cb, std::move(vals));
}
}
// listeners are copied: they may be deleted by the callback
for (auto& cb : cbs)
cb.first(cb.second, false);
}
}
if (not st.listeners.empty()) {
if (logger_)
logger_->debug("[store {}] {} remote listeners", id.to_view(), st.listeners.size());
for (const auto& node_listeners : st.listeners) {
for (const auto& l : node_listeners.second) {
auto f = l.second.query.where.getFilter();
if (f and not f(*v))
continue;
if (logger_)
logger_->debug("[store {}] [node {}] Sending update",
id.to_view(),
node_listeners.first->toString());
std::vector<Sp<Value>> vals;
vals.push_back(v);
Blob ntoken = makeToken(node_listeners.first->getAddr(), false);
network_engine.tellListener(node_listeners.first,
l.first,
id,
0,
ntoken,
{},
{},
std::move(vals),
l.second.query,
l.second.version);
}
}
}
}
bool
Dht::storageStore(const InfoHash& id,
const Sp<Value>& value,
time_point created,
const SockAddr& sa,
bool permanent,
time_point expiration)
{
const auto& now = scheduler.time();
created = std::min(created, now);
if (expiration == time_point::min())
expiration = permanent ? time_point::max() : created + getType(value->type).expiration;
else if (not permanent)
expiration = std::min(expiration, created + getType(value->type).expiration);
if (expiration < now)
return false;
auto st = store.find(id);
if (st == store.end()) {
if (store.size() >= max_store_keys)
return false;
auto st_i = store.emplace(id, now);
st = st_i.first;
if (maintain_storage and st_i.second)
scheduler.add(st->second.maintenance_time, std::bind(&Dht::dataPersistence, this, id));
}
StorageBucket* store_bucket {nullptr};
bool is_local = !sa;
if (sa)
store_bucket = &store_quota.try_emplace(sa, sa).first->second;
else
store_bucket = local_store_quota.get();
// Reject new local values if local storage limit is exceeded
if (is_local && max_local_store_size != STORAGE_LIMIT_UNLIMITED
&& local_store_quota->size() + value->size() > max_local_store_size) {
if (logger_)
logger_->warn("[store {}] Rejecting local put: local storage limit exceeded ({}/{})",
id.to_view(),
local_store_quota->size(),
max_local_store_size);
return false;
}
auto store = st->second.store(id, value, created, expiration, store_bucket);
if (auto vs = store.first) {
total_store_size += store.second.size_diff;
total_values += store.second.values_diff;
scheduler.cancel(vs->expiration_job);
if (not permanent) {
vs->expiration_job = scheduler.add(expiration, std::bind(&Dht::expireStorage, this, id));
}
if (total_store_size - local_store_quota->size() > max_store_size) {
auto value = vs->data;
auto value_diff = store.second.values_diff;
auto value_edit = store.second.edited_values;
expireStore();
storageChanged(id, st->second, value, value_diff > 0 || value_edit > 0);
} else {
storageChanged(id, st->second, vs->data, store.second.values_diff > 0 || store.second.edited_values > 0);
}
}
return std::get<0>(store);
}
void
Dht::storageAddListener(const InfoHash& id, const Sp<Node>& node, size_t socket_id, Query&& query, int version)
{
const auto& now = scheduler.time();
auto st = store.find(id);
if (st == store.end()) {
if (store.size() >= max_store_keys)
return;
st = store.emplace(id, now).first;
}
auto& node_listeners = st->second.listeners[node];
auto l = node_listeners.find(socket_id);
if (l == node_listeners.end()) {
auto vals = st->second.get(query.where.getFilter());
if (not vals.empty()) {
network_engine.tellListener(node,
socket_id,
id,
WANT4 | WANT6,
makeToken(node->getAddr(), false),
dht4.buckets.findClosestNodes(id, now, TARGET_NODES),
dht6.buckets.findClosestNodes(id, now, TARGET_NODES),
std::move(vals),
query,
version);
}
node_listeners.emplace(socket_id, Listener {now, std::forward<Query>(query), version});
} else
l->second.refresh(now, std::forward<Query>(query));
}
void
Dht::expireStore(decltype(store)::iterator i)
{
const auto& id = i->first;
auto& st = i->second;
auto stats = st.expire(id, scheduler.time());
if (not stats.expired_values.empty()) {
storageRemoved(id, st, stats.expired_values, -stats.size_diff);
}
}
void
Dht::expireStorage(InfoHash h)
{
auto i = store.find(h);
if (i != store.end())
expireStore(i);
}
void
Dht::storageRemoved(const InfoHash& id, Storage& st, const std::vector<Sp<Value>>& values, size_t totalSize)
{
if (logger_)
logger_->debug("[store {}] Discarded {} values ({} bytes)", id.to_view(), values.size(), totalSize);
total_store_size -= totalSize;
total_values -= values.size();
if (not st.listeners.empty()) {
if (logger_)
logger_->debug("[store {}] {} remote listeners", id.to_view(), st.listeners.size());
std::vector<Value::Id> ids;
ids.reserve(values.size());
for (const auto& v : values)
ids.emplace_back(v->id);
for (const auto& node_listeners : st.listeners) {
for (const auto& l : node_listeners.second) {
if (logger_)
logger_->warn("[store {}] [node {}] Sending expired",
id.to_view(),
node_listeners.first->toString());
Blob ntoken = makeToken(node_listeners.first->getAddr(), false);
network_engine.tellListenerExpired(node_listeners.first, l.first, id, ntoken, ids, l.second.version);
}
}
}
for (const auto& local_listeners : st.local_listeners) {
local_listeners.second.get_cb(values, true);
}
}
void
Dht::expireStore()
{
// removing expired values
for (auto i = store.begin(); i != store.end();) {
expireStore(i);
if (i->second.empty() && i->second.listeners.empty() && i->second.local_listeners.empty()) {
if (logger_)
logger_->debug("[store {}] Discarding empty storage", i->first.to_view());
i = store.erase(i);
} else
++i;
}
// remove more values if remote storage limit is exceeded
while (total_store_size - local_store_quota->size() > max_store_size) {
// find IP using the most storage
if (store_quota.empty()) {
if (logger_)
logger_->warn("No space left: local data consumes all the quota!");
break;
}
auto largest = store_quota.begin();
for (auto it = std::next(largest); it != store_quota.end(); ++it) {
if (it->second.size() > largest->second.size())
largest = it;
}
if (largest != store_quota.end() && !largest->second.empty()) {
while (true) {
auto exp_value = largest->second.getOldest();
auto storage = store.find(exp_value.first);
if (storage != store.end()) {
if (logger_)
logger_->warn("Storage quota full: discarding value from {} at {} {:016x}",
largest->first.toString(),
exp_value.first.to_view(),
exp_value.second);
if (auto value = storage->second.remove(exp_value.first, exp_value.second)) {
storageRemoved(storage->first, storage->second, {value}, value->size());
break;
}
}
}
} else {
if (logger_)
logger_->warn("No space left: remote data consumes all the quota!");
break;
}
}
// remove unused quota entires
for (auto i = store_quota.begin(); i != store_quota.end();) {
if (i->second.empty())
i = store_quota.erase(i);
else
++i;
}
}
void
Dht::connectivityChanged(sa_family_t af)
{
const auto& now = scheduler.time();
scheduler.edit(nextNodesConfirmation, now);
auto& dht = this->dht(af);
dht.buckets.connectivityChanged(now);
dht.reported_addr.clear();
network_engine.connectivityChanged(af);
startBootstrap(); // will only happen if disconnected
}
void
Dht::rotateSecrets()
{
oldsecret = secret;
secret = std::uniform_int_distribution<uint64_t> {}(rd);
uniform_duration_distribution<> time_dist(std::chrono::minutes(15), std::chrono::minutes(45));
auto rotate_secrets_time = scheduler.time() + time_dist(rd);
scheduler.add(rotate_secrets_time, std::bind(&Dht::rotateSecrets, this));
}
Blob
Dht::makeToken(const SockAddr& addr, bool old) const
{
const void* ip;
size_t iplen;
in_port_t port;
auto family = addr.getFamily();
if (family == AF_INET) {
const auto& sin = addr.getIPv4();
ip = &sin.sin_addr;
iplen = 4;
port = sin.sin_port;
} else if (family == AF_INET6) {
const auto& sin6 = addr.getIPv6();
ip = &sin6.sin6_addr;
iplen = 16;
port = sin6.sin6_port;
} else {
return {};
}
const auto& c1 = old ? oldsecret : secret;
Blob data;
data.reserve(sizeof(secret) + sizeof(in_port_t) + iplen);
data.insert(data.end(), (uint8_t*) &c1, ((uint8_t*) &c1) + sizeof(c1));
data.insert(data.end(), (uint8_t*) ip, (uint8_t*) ip + iplen);
data.insert(data.end(), (uint8_t*) &port, ((uint8_t*) &port) + sizeof(in_port_t));
return crypto::hash(data, TOKEN_SIZE);
}
bool
Dht::tokenMatch(const Blob& token, const SockAddr& addr) const
{
if (not addr or token.size() != TOKEN_SIZE)
return false;
if (token == makeToken(addr, false))
return true;
if (token == makeToken(addr, true))
return true;
return false;
}
NodeStats
Dht::getNodesStats(sa_family_t af) const
{
NodeStats stats = dht(af).getNodesStats(scheduler.time(), myid);
stats.node_cache_size = network_engine.getNodeCacheSize(af);
return stats;
}
NodeStats
Dht::Kad::getNodesStats(time_point now, const InfoHash& myid) const
{
NodeStats stats {};
for (const auto& b : buckets) {
for (auto& n : b.nodes) {
if (n->isGood(now)) {
stats.good_nodes++;
if (n->isIncoming())
stats.incoming_nodes++;
} else if (not n->isExpired())
stats.dubious_nodes++;
}
if (b.cached)
stats.cached_nodes++;
}
stats.table_depth = buckets.depth(buckets.findBucket(myid));
stats.searches = searches.size();
return stats;
}
void
Dht::dumpBucket(const Bucket& b, std::ostream& out) const
{
const auto& now = scheduler.time();
using namespace std::chrono;
out << b.first << " count: " << b.nodes.size() << " updated: " << print_time_relative(now, b.time);
if (b.cached)
out << " (cached)";
out << std::endl;
for (auto& n : b.nodes) {
out << " Node " << n->toString();
const auto& t = n->getTime();
const auto& r = n->getReplyTime();
if (t != r)
out << " updated: " << print_time_relative(now, t) << ", replied: " << print_time_relative(now, r);
else
out << " updated: " << print_time_relative(now, t);
if (n->isExpired())
out << " [expired]";
else if (n->isGood(now))
out << " [good]";
out << std::endl;
}
}
void
Dht::dumpSearch(const Search& sr, std::ostream& out) const
{
const auto& now = scheduler.time();
const auto& listen_expire = getListenExpiration();
using namespace std::chrono;
fmt::print(out,
"Search IPv{} {} gets: {} last step: {}{}{}{}{}\n",
(sr.af == AF_INET6 ? '6' : '4'),
sr.id,
sr.callbacks.size(),
print_time_relative(now, sr.step_time),
(sr.done ? " [done]"sv : ""sv),
(sr.expired ? " [expired]"sv : ""sv),
(sr.isSynced(now) ? " [synced]"sv : " [not synced]"sv),
(sr.isListening(now, listen_expire) ? " [listening]"sv : ""sv));
// printing the queries
if (sr.callbacks.size() + sr.listeners.size() > 0)
fmt::print(out, "Queries:\n");
for (const auto& cb : sr.callbacks) {
out << *cb.second.query << std::endl;
}
for (const auto& l : sr.listeners) {
out << *l.second.query << std::endl;
}
for (const auto& a : sr.announce) {
bool announced = sr.isAnnounced(a.value->id);
out << "Announcement: " << *a.value << (announced ? " [announced]" : "") << std::endl;
}
fmt::print(out, " Common bits InfoHash Conn. Get Ops IP\n");
auto last_get = sr.getLastGetTime();
for (const auto& np : sr.nodes) {
auto& n = *np;
auto listen = (not sr.listeners.empty()
? (n.listenStatus.empty() ? " "
: fmt::format("[{}] ",
(n.isListening(now, listen_expire)
? 'l'
: (n.pending(n.listenStatus) ? 'f' : ' '))))
: "");
std::string announce;
if (not sr.announce.empty()) {
if (n.acked.empty()) {
announce = std::string(sr.announce.size() + 3, ' ');
} else {
announce = "[";
for (const auto& a : sr.announce) {
auto ack = n.acked.find(a.value->id);
if (ack == n.acked.end() or not ack->second.req) {
announce += ' ';
} else {
announce += ack->second.req->getStateChar();
}
}
announce += "] ";
}
}
fmt::print(out,
"{:3} {} {} [{}{}] {}{}{}\n",
InfoHash::commonBits(sr.id, n.node->id),
n.node->id,
(findNode(n.node->id, sr.af) ? '*' : ' '),
(n.pending(n.getStatus) ? (n.candidate ? 'c' : 'f') : ' '),
(n.isSynced(now) ? (n.last_get_reply > last_get ? 'u' : 's') : '-'),
listen,
announce,
n.node->getAddrStr());
}
}
void
Dht::dumpTables() const
{
std::ostringstream out;
out << "My id " << myid << std::endl;
out << "Buckets IPv4 :" << std::endl;
for (const auto& b : dht4.buckets)
dumpBucket(b, out);
out << "Buckets IPv6 :" << std::endl;
for (const auto& b : dht6.buckets)
dumpBucket(b, out);
auto dump_searches = [&](std::map<InfoHash, Sp<Search>> srs) {
for (auto& srp : srs)
dumpSearch(*srp.second, out);
};
dump_searches(dht4.searches);
dump_searches(dht6.searches);
out << std::endl;
out << getStorageLog() << std::endl;
if (logger_)
logger_->debug("{}", out.str());
}
std::string
Dht::getStorageLog() const
{
std::ostringstream out;
for (const auto& s : store) {
const auto& st = s.second;
out << "Storage " << s.first << " " << st.listeners.size() << " list., " << st.valueCount() << " values, "
<< printByteCount(st.totalSize()) << "\n";
}
out << std::endl << std::endl;
std::multimap<size_t, const SockAddr*> q_map;
for (const auto& ip : store_quota)
if (ip.second.size())
q_map.emplace(ip.second.size(), &ip.first);
for (auto ip = q_map.rbegin(); ip != q_map.rend(); ++ip)
out << "IP " << ip->second->toString() << " uses " << printByteCount(ip->first) << std::endl;
out << std::endl;
out << "Total " << store.size() << " storages, " << total_values << " values (";
if (total_store_size < 1024)
out << total_store_size << " bytes)";
else {
auto remote_store_size = total_store_size - local_store_quota->size();
out << (remote_store_size / 1024) << " / " << (max_store_size / 1024) << " KB remote, "
<< (local_store_quota->size() / 1024) << " / ";
if (max_local_store_size == STORAGE_LIMIT_UNLIMITED)
out << "unlimited";
else
out << (max_local_store_size / 1024) << " KB";
out << " local)";
}
out << std::endl;
return out.str();
}
std::string
Dht::getStorageLog(const InfoHash& h) const
{
auto s = store.find(h);
if (s == store.end())
return fmt::format("Storage {} empty\n", h);
return printStorageLog(*s);
}
std::string
Dht::printStorageLog(const decltype(store)::value_type& s) const
{
std::ostringstream out;
using namespace std::chrono;
const auto& st = s.second;
const auto& now = scheduler.time();
out << "Storage " << s.first << " " << st.listeners.size() << " list., " << st.valueCount() << " values, "
<< printByteCount(st.totalSize()) << std::endl;
for (const auto& v : st.getValues()) {
out << " Value " << v.data->id << " size: " << printByteCount(v.data->size())
<< " created: " << print_time_relative(now, v.created)
<< " expires: " << print_time_relative(now, v.expiration) << std::endl;
}
if (not st.local_listeners.empty())
out << " " << st.local_listeners.size() << " local listeners" << std::endl;
for (const auto& node_listeners : st.listeners) {
const auto& node = node_listeners.first;
out << " " << "Listener " << node->toString() << " : " << node_listeners.second.size() << " entries"
<< std::endl;
}
return out.str();
}
std::string
Dht::getRoutingTablesLog(sa_family_t af) const
{
std::ostringstream out;
for (const auto& b : buckets(af))
dumpBucket(b, out);
return out.str();
}
std::string
Dht::getSearchesLog(sa_family_t af) const
{
std::ostringstream out;
auto num_searches = dht4.searches.size() + dht6.searches.size();
if (num_searches > 8) {
if (not af or af == AF_INET)
for (const auto& sr : dht4.searches)
out << "[search " << sr.first << " IPv4]" << std::endl;
if (not af or af == AF_INET6)
for (const auto& sr : dht6.searches)
out << "[search " << sr.first << " IPv6]" << std::endl;
} else {
out << "s:synched, u:updated, a:announced, c:candidate, f:cur req, x:expired, *:known" << std::endl;
if (not af or af == AF_INET)
for (const auto& sr : dht4.searches)
dumpSearch(*sr.second, out);
if (not af or af == AF_INET6)
for (const auto& sr : dht6.searches)
dumpSearch(*sr.second, out);
}
out << "Total: " << num_searches << " searches (" << dht4.searches.size() << " IPv4, " << dht6.searches.size()
<< " IPv6)." << std::endl;
return out.str();
}
std::string
Dht::getSearchLog(const InfoHash& id, sa_family_t af) const
{
std::ostringstream out;
if (af == AF_UNSPEC) {
out << getSearchLog(id, AF_INET) << getSearchLog(id, AF_INET6);
} else {
auto& srs = searches(af);
auto sr = srs.find(id);
if (sr != srs.end())
dumpSearch(*sr->second, out);
}
return out.str();
}
Dht::~Dht()
{
for (auto& s : dht4.searches)
s.second->clear();
for (auto& s : dht6.searches)
s.second->clear();
}
net::NetworkConfig
fromDhtConfig(const Config& config)
{
net::NetworkConfig netConf;
netConf.network = config.network;
netConf.max_req_per_sec = config.max_req_per_sec ? config.max_req_per_sec : MAX_REQUESTS_PER_SEC;
netConf.max_peer_req_per_sec = config.max_peer_req_per_sec ? config.max_peer_req_per_sec
: netConf.max_req_per_sec / 8;
netConf.is_client = config.client_mode;
return netConf;
}
Dht::Dht(std::unique_ptr<net::DatagramSocket>&& sock,
const Config& config,
const Sp<Logger>& l,
std::unique_ptr<std::mt19937_64>&& rda)
: DhtInterface(l)
, rd(rda ? std::move(*rda) : crypto::getSeededRandomEngine<std::mt19937_64>())
, myid(config.node_id ? config.node_id : InfoHash::getRandom(rd))
, store()
, store_quota()
, local_store_quota(std::make_unique<StorageBucket>())
, max_store_keys(config.max_store_keys ? (int) config.max_store_keys : MAX_HASHES)
, max_store_size(config.max_store_size ? (size_t) config.max_store_size : STORAGE_LIMIT_DEFAULT)
, max_local_store_size(config.max_local_store_size ? (size_t) config.max_local_store_size : STORAGE_LIMIT_UNLIMITED)
, max_searches(config.max_searches ? (int) config.max_searches : MAX_SEARCHES)
, network_engine(myid,
fromDhtConfig(config),
std::move(sock),
logger_,
rd,
scheduler,
std::bind(&Dht::onError, this, _1, _2),
std::bind(&Dht::onNewNode, this, _1, _2),
std::bind(&Dht::onReportedAddr, this, _1, _2),
std::bind(&Dht::onPing, this, _1),
std::bind(&Dht::onFindNode, this, _1, _2, _3),
std::bind(&Dht::onGetValues, this, _1, _2, _3, _4),
std::bind(&Dht::onListen, this, _1, _2, _3, _4, _5, _6),
std::bind(&Dht::onAnnounce, this, _1, _2, _3, _4, _5),
std::bind(&Dht::onRefresh, this, _1, _2, _3, _4))
, persistPath(config.persist_path)
, is_bootstrap(config.is_bootstrap)
, maintain_storage(config.maintain_storage)
, public_stable(config.public_stable)
{
scheduler.syncTime();
auto s = network_engine.getSocket();
if (not s or (not s->hasIPv4() and not s->hasIPv6()))
throw DhtException("Opened socket required");
if (s->hasIPv4()) {
dht4.buckets = {Bucket {AF_INET}};
dht4.buckets.is_client = config.is_bootstrap;
}
if (s->hasIPv6()) {
dht6.buckets = {Bucket {AF_INET6}};
dht6.buckets.is_client = config.is_bootstrap;
}
search_id = std::uniform_int_distribution<decltype(search_id)> {}(rd);
uniform_duration_distribution<> time_dis {std::chrono::seconds(3), std::chrono::seconds(5)};
nextNodesConfirmation = scheduler.add(scheduler.time() + time_dis(rd), std::bind(&Dht::confirmNodes, this));
// Fill old secret
secret = std::uniform_int_distribution<uint64_t> {}(rd);
rotateSecrets();
if (not persistPath.empty())
loadState(persistPath);
expire();
if (logger_)
logger_->debug("DHT node initialised with ID {:s}", myid);
}
bool
Dht::neighbourhoodMaintenance(RoutingTable& list)
{
// logger__DBG("neighbourhoodMaintenance");
auto b = list.findBucket(myid);
if (b == list.end())
return false;
InfoHash id = myid;
#ifdef _WIN32
std::uniform_int_distribution<int> rand_byte {0, std::numeric_limits<uint8_t>::max()};
#else
std::uniform_int_distribution<uint8_t> rand_byte;
#endif
id[HASH_LEN - 1] = rand_byte(rd);
std::bernoulli_distribution rand_trial(1. / 8.);
auto q = b;
if (std::next(q) != list.end() && (q->nodes.empty() || rand_trial(rd)))
q = std::next(q);
if (b != list.begin() && (q->nodes.empty() || rand_trial(rd))) {
auto r = std::prev(b);
if (!r->nodes.empty())
q = r;
}
auto n = q->randomNode(rd);
if (n) {
if (logger_)
logger_->debug("[node {}] Sending [find {}] for neighborhood maintenance", n->toString(), id.to_view());
/* Since our node-id is the same in both DHTs, it's probably
profitable to query both families. */
network_engine.sendFindNode(n, id, network_engine.want());
}
return true;
}
bool
Dht::bucketMaintenance(RoutingTable& list)
{
std::bernoulli_distribution rand_trial(1. / 8.);
std::bernoulli_distribution rand_trial_38(1. / 38.);
bool sent {false};
for (auto b = list.begin(); b != list.end(); ++b) {
if (b->time < scheduler.time() - std::chrono::minutes(10) || b->nodes.empty()) {
/* This bucket hasn't seen any positive confirmation for a long
time. Pick a random id in this bucket's range, and send a request
to a random node. */
InfoHash id = list.randomId(b, rd);
auto q = b;
/* If the bucket is empty, we try to fill it from a neighbour.
We also sometimes do it gratuitiously to recover from
buckets full of broken nodes. */
if (std::next(b) != list.end() && (q->nodes.empty() || rand_trial(rd)))
q = std::next(b);
if (b != list.begin() && (q->nodes.empty() || rand_trial(rd))) {
auto r = std::prev(b);
if (!r->nodes.empty())
q = r;
}
auto n = q->randomNode(rd);
if (n and not n->isPendingMessage()) {
want_t want = -1;
if (network_engine.want() != want) {
auto otherbucket = findBucket(id, q->af == AF_INET ? AF_INET6 : AF_INET);
if (otherbucket && otherbucket->nodes.size() < TARGET_NODES)
/* The corresponding bucket in the other family
is emptyish -- querying both is useful. */
want = WANT4 | WANT6;
else if (rand_trial_38(rd))
/* Most of the time, this just adds overhead.
However, it might help stitch back one of
the DHTs after a network collapse, so query
both, but only very occasionally. */
want = WANT4 | WANT6;
}
if (logger_)
logger_->debug("[node {}] Sending find {} for bucket maintenance", n->toString(), id.to_view());
// auto start = scheduler.time();
network_engine.sendFindNode(n, id, want, nullptr, [this, n](const net::Request&, bool over) {
if (over) {
const auto& end = scheduler.time();
// using namespace std::chrono;
// if (logger_)
// logger_->d(n->id, "[node %s] Bucket maintenance op expired after %s",
// n->toString().c_str(), print_duration(end-start).c_str());
scheduler.edit(nextNodesConfirmation, end + Node::MAX_RESPONSE_TIME);
}
});
sent = true;
}
}
}
return sent;
}
void
Dht::dataPersistence(InfoHash id)
{
const auto& now = scheduler.time();
auto str = store.find(id);
if (str != store.end() and now > str->second.maintenance_time) {
if (logger_)
logger_->debug("[storage {}] Maintenance ({} values, {} bytes)",
id.to_view(),
str->second.valueCount(),
str->second.totalSize());
maintainStorage(*str);
str->second.maintenance_time = now + MAX_STORAGE_MAINTENANCE_EXPIRE_TIME;
scheduler.add(str->second.maintenance_time, std::bind(&Dht::dataPersistence, this, id));
}
}
size_t
Dht::maintainStorage(decltype(store)::value_type& storage, bool force, const DoneCallback& donecb)
{
const auto& now = scheduler.time();
size_t announce_per_af = 0;
auto maintain = [&](sa_family_t af) {
bool want = true;
auto nodes = buckets(af).findClosestNodes(storage.first, now);
if (!nodes.empty()) {
if (force || storage.first.xorCmp(nodes.back()->id, myid) < 0) {
for (auto& value : storage.second.getValues()) {
const auto& vt = getType(value.data->type);
if (force || value.created + vt.expiration > now + MAX_STORAGE_MAINTENANCE_EXPIRE_TIME) {
// gotta put that value there
announce(storage.first, af, value.data, donecb, value.created);
++announce_per_af;
}
}
want = false;
}
}
return want;
};
bool want4 = maintain(AF_INET), want6 = maintain(AF_INET6);
if (not want4 and not want6) {
if (logger_)
logger_->debug("Discarding storage values {}", storage.first.to_view());
auto diff = storage.second.clear(storage.first);
total_store_size += diff.size_diff;
total_values += diff.values_diff;
}
return announce_per_af;
}
time_point
Dht::periodic(const uint8_t* buf, size_t buflen, SockAddr from, const time_point& now)
{
scheduler.syncTime(now);
if (buflen) {
try {
network_engine.processMessage(buf, buflen, std::move(from));
} catch (const std::exception& e) {
if (logger_)
logger_->warn("Unable to process message: {}", e.what());
}
}
return scheduler.run();
}
void
Dht::expire()
{
uniform_duration_distribution<> time_dis(std::chrono::minutes(2), std::chrono::minutes(6));
auto expire_stuff_time = scheduler.time() + duration(time_dis(rd));
expireBuckets(dht4.buckets);
expireBuckets(dht6.buckets);
expireStore();
expireSearches();
scheduler.add(expire_stuff_time, std::bind(&Dht::expire, this));
}
void
Dht::onConnected()
{
stopBootstrap();
auto restartSearches = [this](std::map<InfoHash, Sp<Search>>& searches) {
const auto& now = scheduler.time();
for (auto& srp : searches) {
auto& sr = *srp.second;
if (sr.done)
continue;
scheduler.edit(sr.nextSearchStep, now);
}
};
restartSearches(dht4.searches);
restartSearches(dht6.searches);
auto callbacks = std::move(onConnectCallbacks_);
while (not callbacks.empty()) {
callbacks.front()();
callbacks.pop();
}
}
void
Dht::onDisconnected()
{
if (not bootstrapJob)
bootstrap();
}
void
Dht::bootstrap()
{
if (dht4.status != NodeStatus::Disconnected || dht6.status != NodeStatus::Disconnected) {
// The job that led here has already run, and the scheduler does not
// clear the handle to a job it has executed. onDisconnected() reads a
// set handle as "a bootstrap is already scheduled" and does nothing,
// so leaving it set here means the node never tries to rejoin the
// network once it has lost it. Clearing it is what says that no
// bootstrap is pending any more.
bootstrapJob.reset();
return;
}
if (logger_)
logger_->debug("[{}] Bootstraping", myid.to_view());
bootstrap_pending = false;
for (const auto& boootstrap : bootstrap_nodes) {
try {
auto ips = network_engine.getSocket()->resolve(boootstrap.first, boootstrap.second);
for (auto& ip : ips) {
if (ip.getPort() == 0)
ip.setPort(net::DHT_DEFAULT_PORT);
pingNode(ip);
}
} catch (const std::exception& e) {
if (logger_)
logger_->error("[{}] Unable to resolve {}:{}: {}",
myid.to_view(),
boootstrap.first,
boootstrap.second,
e.what());
}
}
scheduler.cancel(bootstrapJob);
bootstrapJob = scheduler.add(scheduler.time() + bootstrap_period, std::bind(&Dht::bootstrap, this));
bootstrap_period = std::min(bootstrap_period * 2, BOOTSTRAP_PERIOD_MAX);
}
void
Dht::startBootstrap()
{
stopBootstrap();
bootstrap_pending = !bootstrap_nodes.empty();
bootstrapJob = scheduler.add(scheduler.time(), std::bind(&Dht::bootstrap, this));
}
void
Dht::stopBootstrap()
{
scheduler.cancel(bootstrapJob);
bootstrap_pending = false;
bootstrap_period = BOOTSTRAP_PERIOD;
}
void
Dht::confirmNodes()
{
using namespace std::chrono;
bool soon = false;
const auto& now = scheduler.time();
if (dht4.searches.empty() and dht4.status == NodeStatus::Connected) {
if (logger_)
logger_->debug("[confirm nodes] Initial IPv4 'get' for my id ({})", myid.to_view());
search(myid, AF_INET);
}
if (dht6.searches.empty() and dht6.status == NodeStatus::Connected) {
if (logger_)
logger_->debug("[confirm nodes] Initial IPv6 'get' for my id ({})", myid.to_view());
search(myid, AF_INET6);
}
soon |= bucketMaintenance(dht4.buckets);
soon |= bucketMaintenance(dht6.buckets);
if (!soon) {
if (dht4.buckets.grow_time >= now - seconds(150))
soon |= neighbourhoodMaintenance(dht4.buckets);
if (dht6.buckets.grow_time >= now - seconds(150))
soon |= neighbourhoodMaintenance(dht6.buckets);
}
/* In order to maintain all buckets' age within 600 seconds, worst
case is roughly 27 seconds, assuming the table is 22 bits deep.
We want to keep a margin for neighborhood maintenance, so keep
this within 25 seconds. */
auto time_dis = soon ? uniform_duration_distribution<> {seconds(5), seconds(25)}
: uniform_duration_distribution<> {seconds(60), seconds(180)};
auto confirm_nodes_time = now + time_dis(rd);
scheduler.edit(nextNodesConfirmation, confirm_nodes_time);
}
std::vector<ValuesExport>
Dht::exportValues() const
{
const auto& now = scheduler.time();
const auto system_now = system_clock::now();
std::vector<ValuesExport> e {};
e.reserve(store.size());
for (const auto& h : store) {
ValuesExport ve;
ve.first = h.first;
msgpack::sbuffer buffer;
msgpack::packer<msgpack::sbuffer> pk(&buffer);
const auto& vals = h.second.getValues();
pk.pack_array(vals.size());
for (const auto& v : vals) {
pk.pack_array(4);
pk.pack(to_system_time(v.created, now, system_now));
v.data->msgpack_pack(pk);
if (const auto& store_addr = v.store_bucket ? v.store_bucket->getAddr() : SockAddr {}) {
pk.pack_bin(store_addr.getLength());
pk.pack_bin_body(reinterpret_cast<const char*>(store_addr.get()), store_addr.getLength());
} else {
pk.pack_bin(0);
}
pk.pack(to_system_time(v.expiration, now, system_now));
}
ve.second = {buffer.data(), buffer.data() + buffer.size()};
e.push_back(std::move(ve));
}
return e;
}
void
Dht::importValues(const std::vector<ValuesExport>& import)
{
const auto& now = scheduler.time();
const auto system_now = system_clock::now();
unsigned imported = 0, ignored = 0;
size_t imported_size = 0;
for (const auto& value : import) {
if (value.second.empty()) {
ignored++;
continue;
}
try {
msgpack::unpacked msg;
msgpack::unpack(msg, (const char*) value.second.data(), value.second.size());
auto valarr = msg.get();
if (valarr.type != msgpack::type::ARRAY)
throw msgpack::type_error();
for (unsigned i = 0; i < valarr.via.array.size; i++) {
auto& valel = valarr.via.array.ptr[i];
if (valel.type != msgpack::type::ARRAY or valel.via.array.size < 2)
throw msgpack::type_error();
time_point val_time;
time_point expiration = time_point::min();
Value tmp_val;
SockAddr store_addr;
try {
const bool uses_system_time = valel.via.array.ptr[0].type == msgpack::type::EXT;
if (uses_system_time) {
val_time = from_system_time(valel.via.array.ptr[0].as<system_clock::time_point>(),
now,
system_now);
} else {
val_time = time_point {
time_point::duration {valel.via.array.ptr[0].as<time_point::duration::rep>()}};
}
tmp_val.msgpack_unpack(valel.via.array.ptr[1]);
if (valel.via.array.size >= 3) {
auto addr_blob = valel.via.array.ptr[2].as<Blob>();
if (not addr_blob.empty()) {
store_addr = SockAddr(reinterpret_cast<const sockaddr*>(addr_blob.data()),
static_cast<socklen_t>(addr_blob.size()));
}
}
if (valel.via.array.size >= 4) {
if (uses_system_time) {
expiration = from_system_time(valel.via.array.ptr[3].as<system_clock::time_point>(),
now,
system_now);
} else {
expiration = time_point {
time_point::duration {valel.via.array.ptr[3].as<time_point::duration::rep>()}};
}
}
} catch (const std::exception&) {
if (logger_)
logger_->error("Error reading value at {}", value.first.to_view());
ignored++;
continue;
}
val_time = std::min(val_time, now);
auto val_size = tmp_val.size();
if (storageStore(value.first,
std::make_shared<Value>(std::move(tmp_val)),
val_time,
store_addr,
false,
expiration)) {
imported++;
imported_size += val_size;
} else {
ignored++;
}
}
} catch (const std::exception&) {
ignored++;
if (logger_)
logger_->error("Error reading values at {}", value.first.to_view());
continue;
}
}
if (logger_)
logger_->debug("Imported {} values, {}, ignored {}", imported, dht::printByteCount(imported_size), ignored);
}
std::vector<NodeExport>
Dht::exportNodes() const
{
const auto& now = scheduler.time();
std::vector<NodeExport> nodes;
const auto b4 = dht4.buckets.findBucket(myid);
if (b4 != dht4.buckets.end()) {
for (auto& n : b4->nodes)
if (n->isGood(now))
nodes.emplace_back(n->exportNode());
}
const auto b6 = dht6.buckets.findBucket(myid);
if (b6 != dht6.buckets.end()) {
for (auto& n : b6->nodes)
if (n->isGood(now))
nodes.emplace_back(n->exportNode());
}
for (auto b = dht4.buckets.begin(); b != dht4.buckets.end(); ++b) {
if (b == b4)
continue;
for (auto& n : b->nodes)
if (n->isGood(now))
nodes.emplace_back(n->exportNode());
}
for (auto b = dht6.buckets.begin(); b != dht6.buckets.end(); ++b) {
if (b == b6)
continue;
for (auto& n : b->nodes)
if (n->isGood(now))
nodes.emplace_back(n->exportNode());
}
return nodes;
}
void
Dht::insertNode(const InfoHash& id, const SockAddr& addr)
{
if (addr.getFamily() != AF_INET && addr.getFamily() != AF_INET6)
return;
scheduler.syncTime();
network_engine.insertNode(id, addr);
}
void
Dht::pingNode(SockAddr sa, DoneCallbackSimple&& cb)
{
scheduler.syncTime();
if (logger_)
logger_->debug("Sending ping to {}", sa);
auto& count = dht(sa.getFamily()).pending_pings;
count++;
network_engine.sendPing(
std::move(sa),
[&count, cb](const net::Request&, net::RequestAnswer&&) {
count--;
if (cb)
cb(true);
},
[&count, cb](const net::Request&, bool last) {
if (last) {
count--;
if (cb)
cb(false);
}
});
}
void
Dht::onError(Sp<net::Request> req, net::DhtProtocolException e)
{
const auto& node = req->node;
if (e.getCode() == net::DhtProtocolException::UNAUTHORIZED) {
if (logger_)
logger_->error("[node {}] Token flush", node->toString());
node->authError();
for (auto& srp : searches(node->getFamily())) {
auto& sr = srp.second;
for (auto& n : sr->nodes) {
if (n->node != node)
continue;
n->token.clear();
n->last_get_reply = time_point::min();
for (auto& listen : n->listenStatus) {
if (listen.second.req and listen.second.req->pending())
n->node->cancelRequest(listen.second.req);
listen.second.req = {};
n->node->closeSocket(listen.second.socketId);
if (listen.second.refresh) {
listen.second.refresh->cancel();
listen.second.refresh = {};
}
listen.second.cache.onSynced(false);
}
searchSendGetValues(sr);
scheduler.edit(sr->nextSearchStep, scheduler.time());
break;
}
}
} else if (e.getCode() == net::DhtProtocolException::NOT_FOUND) {
if (logger_)
logger_->error("[node {}] Returned error 404: storage not found", node->toString());
node->cancelRequest(req);
}
}
void
Dht::onReportedAddr(const InfoHash& /*id*/, const SockAddr& addr)
{
if (addr)
reportedAddr(addr);
}
net::RequestAnswer
Dht::onPing(Sp<Node>)
{
return {};
}
net::RequestAnswer
Dht::onFindNode(Sp<Node> node, const InfoHash& target, want_t want)
{
const auto& now = scheduler.time();
net::RequestAnswer answer;
answer.ntoken = makeToken(node->getAddr(), false);
if (want & WANT4)
answer.nodes4 = dht4.buckets.findClosestNodes(target, now, TARGET_NODES);
if (want & WANT6)
answer.nodes6 = dht6.buckets.findClosestNodes(target, now, TARGET_NODES);
return answer;
}
net::RequestAnswer
Dht::onGetValues(Sp<Node> node, const InfoHash& hash, want_t, const Query& query)
{
if (not hash) {
if (logger_)
logger_->warn("[node {}] Eek! Got get_values with no info_hash", node->toString());
throw net::DhtProtocolException {net::DhtProtocolException::NON_AUTHORITATIVE_INFORMATION,
net::DhtProtocolException::GET_NO_INFOHASH};
}
const auto& now = scheduler.time();
net::RequestAnswer answer {};
auto st = store.find(hash);
answer.ntoken = makeToken(node->getAddr(), false);
answer.nodes4 = dht4.buckets.findClosestNodes(hash, now, TARGET_NODES);
answer.nodes6 = dht6.buckets.findClosestNodes(hash, now, TARGET_NODES);
if (st != store.end() && not st->second.empty()) {
answer.values = st->second.get(query.where.getFilter());
if (logger_)
logger_->debug("[node {}] Sending {} values", node->toString(), answer.values.size());
}
return answer;
}
void
Dht::onGetValuesDone(const Sp<Node>& node, net::RequestAnswer& a, Sp<Search>& sr, const Sp<Query>& orig_query)
{
if (not sr) {
if (logger_)
logger_->warn("[search unknown] Got reply to 'get'. Ignoring.");
return;
}
/* if (logger_)
logger_->d(sr->id, "[search %s] [node %s] Got reply to 'get' with %u nodes",
sr->id.toString().c_str(), node->toString().c_str(), a.nodes4.size()+a.nodes6.size());*/
if (not a.ntoken.empty()) {
if (not a.values.empty() or not a.fields.empty()) {
if (logger_)
logger_->debug("[search {}] [node {}] Found {} values",
sr->id.to_view(),
node->toString(),
a.values.size());
for (auto& getp : sr->callbacks) { /* call all callbacks for this search */
auto& get = getp.second;
if (not(get.get_cb or get.query_cb)
or (orig_query and get.query and not get.isSatisfiedBy(*orig_query)))
continue;
if (get.query_cb) { /* in case of a request with query */
if (not a.fields.empty()) {
get.query_cb(a.fields);
} else if (not a.values.empty()) {
std::vector<Sp<FieldValueIndex>> fields;
fields.reserve(a.values.size());
for (const auto& v : a.values)
fields.emplace_back(
std::make_shared<FieldValueIndex>(*v, orig_query ? orig_query->select : Select {}));
get.query_cb(fields);
}
} else if (get.get_cb) { /* in case of a vanilla get request */
std::vector<Sp<Value>> tmp;
for (const auto& v : a.values)
if (not get.filter or get.filter(*v))
tmp.emplace_back(v);
if (not tmp.empty())
get.get_cb(tmp);
}
}
/* callbacks for local search listeners */
/*std::vector<std::pair<ValueCallback, std::vector<Sp<Value>>>> tmp_lists;
for (auto& l : sr->listeners) {
if (!l.second.get_cb or (orig_query and l.second.query and not
l.second.query->isSatisfiedBy(*orig_query))) continue; std::vector<Sp<Value>> tmp; for (const auto& v :
a.values) if (not l.second.filter or l.second.filter(*v)) tmp.emplace_back(v); if (not tmp.empty())
tmp_lists.emplace_back(l.second.get_cb, std::move(tmp));
}
for (auto& l : tmp_lists)
l.first(l.second, false);*/
} else if (not a.expired_values.empty()) {
if (logger_)
logger_->warn("[search {}] [node {}] {} expired values",
sr->id.toString(),
node->toString(),
a.expired_values.size());
}
} else {
if (logger_)
logger_->warn("[node {}] No token provided. Ignoring response content.", node->toString());
network_engine.blacklistNode(node);
}
if (not sr->done) {
searchSendGetValues(sr);
// Force to recompute the next step time
scheduler.edit(sr->nextSearchStep, scheduler.time());
}
}
net::RequestAnswer
Dht::onListen(Sp<Node> node, const InfoHash& hash, const Blob& token, size_t socket_id, const Query& query, int version)
{
if (not hash) {
if (logger_)
logger_->warn("[node {}] Listen with no info_hash", node->toString());
throw net::DhtProtocolException {net::DhtProtocolException::NON_AUTHORITATIVE_INFORMATION,
net::DhtProtocolException::LISTEN_NO_INFOHASH};
}
if (not tokenMatch(token, node->getAddr())) {
if (logger_)
logger_->warn("[node {}] Incorrect token {} for 'listen'", node->toString(), hash.toString());
throw net::DhtProtocolException {net::DhtProtocolException::UNAUTHORIZED,
net::DhtProtocolException::LISTEN_WRONG_TOKEN};
}
Query q = query;
storageAddListener(hash, node, socket_id, std::move(q), version);
net::RequestAnswer answer {};
answer.ntoken = makeToken(node->getAddr(), false);
return answer;
}
void
Dht::onListenDone(const Sp<Node>& node, net::RequestAnswer& answer, Sp<Search>& sr)
{
// if (logger_)
// logger_->d(sr->id, node->id, "[search %s] [node %s] Got listen confirmation",
// sr->id.toString().c_str(), node->toString().c_str(), answer.values.size());
if (not sr->done) {
const auto& now = scheduler.time();
if (not answer.ntoken.empty())
if (auto sn = sr->getNode(node))
sn->refreshToken(answer.ntoken);
searchSendGetValues(sr);
scheduler.edit(sr->nextSearchStep, now);
}
}
net::RequestAnswer
Dht::onAnnounce(Sp<Node> n,
const InfoHash& hash,
const Blob& token,
const std::vector<Sp<Value>>& values,
const time_point& creation_date)
{
auto& node = *n;
if (not hash) {
if (logger_)
logger_->warn("Put with no info_hash");
throw net::DhtProtocolException {net::DhtProtocolException::NON_AUTHORITATIVE_INFORMATION,
net::DhtProtocolException::PUT_NO_INFOHASH};
}
if (!tokenMatch(token, node.getAddr())) {
if (logger_)
logger_->warn("[node {}] Incorrect token {} for 'put'", node.toString(), hash.toString());
throw net::DhtProtocolException {net::DhtProtocolException::UNAUTHORIZED,
net::DhtProtocolException::PUT_WRONG_TOKEN};
}
{
// We store a value only if we think we're part of the
// SEARCH_NODES nodes around the target id.
auto closest_nodes = buckets(node.getFamily()).findClosestNodes(hash, scheduler.time(), SEARCH_NODES);
if (closest_nodes.size() >= TARGET_NODES and hash.xorCmp(closest_nodes.back()->id, myid) < 0) {
if (logger_)
logger_->warn("[node {}] Announce too far from the target. Dropping value.", node.toString());
return {};
}
}
auto created = std::min(creation_date, scheduler.time());
for (const auto& v : values) {
if (v->id == Value::INVALID_ID) {
if (logger_)
logger_->warn("[value {}] Incorrect value id", hash.to_view());
throw net::DhtProtocolException {net::DhtProtocolException::NON_AUTHORITATIVE_INFORMATION,
net::DhtProtocolException::PUT_INVALID_ID};
}
auto lv = getLocalById(hash, v->id);
Sp<Value> vc = v;
if (lv) {
if (*lv == *vc) {
storageRefresh(hash, v->id);
if (logger_)
logger_->debug("[store {}] [node {}] Refreshed value {:016x}",
hash.toString(),
node.toString(),
v->id);
} else {
const auto& type = getType(lv->type);
if (type.editPolicy(hash, lv, vc, node.id, node.getAddr())) {
if (logger_)
logger_->debug("[store {}] Editing {}", hash.toString(), vc->toString());
storageStore(hash, vc, created, node.getAddr());
} else {
if (logger_)
logger_->debug("[store {}] Rejecting edition of {} because of storage policy",
hash.toString(),
vc->toString());
}
}
} else {
// Allow the value to be edited by the storage policy
const auto& type = getType(vc->type);
if (type.storePolicy(hash, vc, node.id, node.getAddr())) {
// if (logger_)
// logger_->d(hash, node.id, "[store %s] Storing %s", hash.toString().c_str(),
// std::to_string(vc->id).c_str());
storageStore(hash, vc, created, node.getAddr());
} else {
if (logger_)
logger_->debug("[store {}] Rejecting storage of {}", hash.toString(), vc->toString());
}
}
}
return {};
}
net::RequestAnswer
Dht::onRefresh(Sp<Node> node, const InfoHash& hash, const Blob& token, const Value::Id& vid)
{
using namespace net;
if (not tokenMatch(token, node->getAddr())) {
if (logger_)
logger_->warn("[node {}] Incorrect token {} for 'put'", node->toString(), hash.toString());
throw DhtProtocolException {DhtProtocolException::UNAUTHORIZED, DhtProtocolException::PUT_WRONG_TOKEN};
}
if (storageRefresh(hash, vid)) {
if (logger_)
logger_->debug("[store {}] [node {}] Refreshed value {:016x}", hash.toString(), node->toString(), vid);
} else {
if (logger_)
logger_->debug("[store {}] [node {}] Got refresh for unknown value", hash.toString(), node->toString());
throw DhtProtocolException {DhtProtocolException::NOT_FOUND, DhtProtocolException::STORAGE_NOT_FOUND};
}
return {};
}
bool
Dht::storageRefresh(const InfoHash& id, Value::Id vid)
{
const auto& now = scheduler.time();
auto s = store.find(id);
if (s != store.end()) {
// Values like for a permanent put can be refreshed. So, inform remote listeners that the value
// need to be refreshed
auto& st = s->second;
if (not st.listeners.empty()) {
if (logger_)
logger_->debug("[store {}] {} remote listeners", id.to_view(), st.listeners.size());
std::vector<Value::Id> ids = {vid};
for (const auto& node_listeners : st.listeners) {
for (const auto& l : node_listeners.second) {
if (logger_)
logger_->warn("[store {}] [node {}] Sending refresh",
id.toString(),
node_listeners.first->toString());
Blob ntoken = makeToken(node_listeners.first->getAddr(), false);
network_engine
.tellListenerRefreshed(node_listeners.first, l.first, id, ntoken, ids, l.second.version);
}
}
}
auto expiration = s->second.refresh(id, now, vid, types);
if (expiration.first) {
scheduler.cancel(expiration.first->expiration_job);
if (expiration.second != time_point::max()) {
expiration.first->expiration_job = scheduler.add(expiration.second,
std::bind(&Dht::expireStorage, this, id));
}
}
return true;
}
return false;
}
void
Dht::onAnnounceDone(const Sp<Node>& node, net::RequestAnswer& answer, Sp<Search>& sr)
{
if (logger_)
logger_->debug("[search {}] [node {}] Got reply to put!", sr->id.toString(), node->toString());
searchSendGetValues(sr);
sr->checkAnnounced(answer.vid);
}
struct DhtState
{
unsigned v {1};
InfoHash id;
std::vector<NodeExport> nodes;
std::vector<ValuesExport> values;
MSGPACK_DEFINE_MAP(v, id, nodes, values)
};
void
Dht::saveState(const std::string& path) const
{
DhtState state;
state.id = myid;
state.nodes = exportNodes();
state.values = exportValues();
std::ofstream file(path);
msgpack::pack(file, state);
}
void
Dht::loadState(const std::string& path)
{
if (logger_)
logger_->debug("Importing state from {}", path);
try {
// Import nodes from binary file
msgpack::unpacker pac;
{
// Read whole file
std::ifstream file(path, std::ios::binary | std::ios::ate);
if (!file.is_open()) {
return;
}
auto size = file.tellg();
file.seekg(0, std::ios::beg);
pac.reserve_buffer(size);
file.read(pac.buffer(), size);
pac.buffer_consumed(size);
}
// Import nodes
msgpack::object_handle oh;
if (pac.next(oh)) {
auto state = oh.get().as<DhtState>();
if (logger_)
logger_->debug("Importing {} nodes", state.nodes.size());
if (state.id)
myid = state.id;
std::vector<Sp<Node>> tmpNodes;
tmpNodes.reserve(state.nodes.size());
for (const auto& node : state.nodes)
tmpNodes.emplace_back(network_engine.insertNode(node.id, node.addr));
importValues(state.values);
}
} catch (const std::exception& e) {
if (logger_)
logger_->warn("Error importing state from {}: {}", path, e.what());
}
}
} // namespace dht
Reference in new issue View Git Blame Copy Permalink