Compare commits

...
Author SHA1 Message Date
Adrien Béraud 37ddd8c7f9 update to version 1.7.3 2018-07-25 16:40:47 -04:00
Adrien Béraud 23057a3082 unit tests: fix race condition 2018-07-25 11:54:28 -04:00
Adrien Béraud dbdd3f8f63 Merge pull request #319 from AmarOk1412/fixLD
dhtrunner: don't use listeners structure without proxy client enabled
2018-07-25 11:32:55 -04:00
Sébastien Blin 751e79a250 dhtrunner: don't use listeners structure without proxy client enabled
The token used to cancel a listener was wrong. Indeed, tokenBGlobal
was used to get the linked listener, not to identify the listener
inside dhtrunner.
2018-07-25 11:28:46 -04:00
Adrien Béraud 7a961ffc51 Merge pull request #318 from atraczyk/master
build: replace WIN32_NATIVE macro with _MSC_VER
2018-07-23 14:01:17 -04:00
Andreas Traczyk bfada958b9 build: replace WIN32_NATIVE macro with _MSC_VER 2018-07-23 13:45:52 -04:00
Sébastien Blin 78d91a8e0d Merge pull request #316 from savoirfairelinux/proxy_server
dhtrunner: decrease max packet handling delay to 500ms
2018-07-13 14:37:09 -04:00
Adrien Béraud cd5dc38a57 dhtrunner: decrease max packet handling delay to 500ms
Allows for faster recovery in case of CPU contention
2018-07-13 14:29:59 -04:00
Sébastien Blin 1ff2eeb220 Merge pull request #313 from savoirfairelinux/proxy_server
Proxy server
2018-06-28 09:16:13 -04:00
Adrien Béraud ffbb9e23ef dht: show usage at the end, print max store size 2018-06-27 11:54:15 -04:00
Adrien Béraud af0a30c9c5 dhtrunner: drop oldest packet if queue is full 2018-06-26 20:41:04 -04:00
Sébastien Blin 97d5e3875d Merge pull request #312 from savoirfairelinux/dht_performance
Dht performance
2018-06-21 11:49:09 -04:00
Adrien Béraud 2509e8bc5c Revert "search: use pointer to SearchNode"
This reverts commit 3a41f2ce56.
2018-06-20 14:50:40 -04:00
Adrien Béraud e349912c07 search: improve canGet performance 2018-06-20 14:50:40 -04:00
8 changed files with 127 additions and 117 deletions

No files matched your search

+1 -1
View File
@@ -1,7 +1,7 @@
cmake_minimum_required (VERSION 3.1)
project (opendht)
set (opendht_VERSION_MAJOR 1)
set (opendht_VERSION_MINOR 7.2)
set (opendht_VERSION_MINOR 7.3)
set (opendht_VERSION ${opendht_VERSION_MAJOR}.${opendht_VERSION_MINOR})
set (PACKAGE_VERSION ${opendht_VERSION})
set (VERSION "${opendht_VERSION}")
+4 -4
View File
@@ -146,7 +146,7 @@
<Optimization>Disabled</Optimization>
<SDLCheck>true</SDLCheck>
<AdditionalIncludeDirectories>..\..\include;..\include;..\include\opendht;..\..\msgpack-c\include;</AdditionalIncludeDirectories>
<PreprocessorDefinitions>_CRT_SECURE_NO_WARNINGS;WIN32_NATIVE;WIN32_LEAN_AND_MEAN;_MBCS;%(PreprocessorDefinitions)</PreprocessorDefinitions>
<PreprocessorDefinitions>_CRT_SECURE_NO_WARNINGS;WIN32_LEAN_AND_MEAN;_MBCS;%(PreprocessorDefinitions)</PreprocessorDefinitions>
<DisableSpecificWarnings>4804;4800;4101;4267;4244;4503;</DisableSpecificWarnings>
<AdditionalOptions>-D_SCL_SECURE_NO_WARNINGS %(AdditionalOptions)</AdditionalOptions>
<SuppressStartupBanner>false</SuppressStartupBanner>
@@ -169,7 +169,7 @@ copy ..\include\opendht.h $(OutDir)\include\</Command>
<Optimization>Disabled</Optimization>
<SDLCheck>true</SDLCheck>
<AdditionalIncludeDirectories>..\..\include;..\include;..\include\opendht;..\..\msgpack-c\include;</AdditionalIncludeDirectories>
<PreprocessorDefinitions>_CRT_SECURE_NO_WARNINGS;WIN32_NATIVE;WIN32_LEAN_AND_MEAN;_MBCS;%(PreprocessorDefinitions)</PreprocessorDefinitions>
<PreprocessorDefinitions>_CRT_SECURE_NO_WARNINGS;WIN32_LEAN_AND_MEAN;_MBCS;%(PreprocessorDefinitions)</PreprocessorDefinitions>
<DisableSpecificWarnings>4804;4800;4101;4267;4244;4503;</DisableSpecificWarnings>
<AdditionalOptions>-D_SCL_SECURE_NO_WARNINGS %(AdditionalOptions)</AdditionalOptions>
</ClCompile>
@@ -193,7 +193,7 @@ copy ..\include\opendht.h $(OutDir)\include\</Command>
<IntrinsicFunctions>true</IntrinsicFunctions>
<SDLCheck>true</SDLCheck>
<AdditionalIncludeDirectories>..\..\include;..\include;..\include\opendht;..\..\msgpack-c\include;</AdditionalIncludeDirectories>
<PreprocessorDefinitions>_CRT_SECURE_NO_WARNINGS;WIN32_NATIVE;WIN32_LEAN_AND_MEAN;_MBCS;%(PreprocessorDefinitions)</PreprocessorDefinitions>
<PreprocessorDefinitions>_CRT_SECURE_NO_WARNINGS;WIN32_LEAN_AND_MEAN;_MBCS;%(PreprocessorDefinitions)</PreprocessorDefinitions>
<DisableSpecificWarnings>4804;4800;4101;4267;4244;4503;</DisableSpecificWarnings>
<AdditionalOptions>-D_SCL_SECURE_NO_WARNINGS %(AdditionalOptions)</AdditionalOptions>
<ProgramDataBaseFileName>$(OutDir)\lib\x86\$(TargetName).pdb</ProgramDataBaseFileName>
@@ -223,7 +223,7 @@ copy ..\include\opendht.h $(OutDir)\include\</Command>
<IntrinsicFunctions>true</IntrinsicFunctions>
<SDLCheck>true</SDLCheck>
<AdditionalIncludeDirectories>$(ProjectDir)contrib\build\include;$(ProjectDir)..\include;$(ProjectDir)..\include\opendht;$(ProjectDir)contrib\build\msgpack-c\include;$(ProjectDir)contrib\build\argon2\include;$(ProjectDir)..\..\include;$(ProjectDir)..\..\argon2\include</AdditionalIncludeDirectories>
<PreprocessorDefinitions>_CRT_SECURE_NO_WARNINGS;WIN32_NATIVE;WIN32_LEAN_AND_MEAN;_MBCS;%(PreprocessorDefinitions)</PreprocessorDefinitions>
<PreprocessorDefinitions>_CRT_SECURE_NO_WARNINGS;WIN32_LEAN_AND_MEAN;_MBCS;%(PreprocessorDefinitions)</PreprocessorDefinitions>
<DisableSpecificWarnings>4804;4800;4101;4267;4244;4503;4273;</DisableSpecificWarnings>
<AdditionalOptions>-D_SCL_SECURE_NO_WARNINGS %(AdditionalOptions)</AdditionalOptions>
<ProgramDataBaseFileName>$(OutDir)\lib\x64\$(TargetName).pdb</ProgramDataBaseFileName>
+1 -1
View File
@@ -1,7 +1,7 @@
dnl define macros
m4_define([opendht_major_version], 1)
m4_define([opendht_minor_version], 7)
m4_define([opendht_patch_version], 2)
m4_define([opendht_patch_version], 3)
m4_define([opendht_version],
[opendht_major_version.opendht_minor_version.opendht_patch_version])
+1 -1
View File
@@ -25,7 +25,7 @@
namespace dht {
namespace crypto {
#ifndef WIN32_NATIVE
#ifndef _MSC_VER
#ifdef _WIN32
/**
+53 -50
View File
@@ -319,19 +319,20 @@ Dht::searchSendGetValues(Sp<Search> sr, SearchNode* pn, bool update)
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 = not sr->callbacks.empty() ? cb->second.query : std::make_shared<Query>(Select {}, Where {}, true);
const time_point up = not sr->callbacks.empty() and update
? sr->getLastGetTime(query)
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();
if (sn.canGet(now, up, query)) {
n = &sn;
break;
}
}
@@ -384,11 +385,11 @@ void Dht::searchSendAnnounceValue(const Sp<Search>& sr) {
std::weak_ptr<Search> ws = sr;
const auto& now = scheduler.time();
for (auto& n : sr->nodes) {
if (not n->isSynced(now))
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;
return n.getAnnounceTime(a.value->id) <= now;
}) == sr->announce.cend())
continue;
@@ -426,12 +427,13 @@ void Dht::searchSendAnnounceValue(const Sp<Search>& sr) {
bool hasValue {false};
uint16_t seq_no = 0;
try {
for (const auto& field : answer.fields) {
if (field->index.at(Value::Field::Id).getInt() == a.value->id) {
hasValue = true;
seq_no = static_cast<uint16_t>(field->index.at(Value::Field::SeqNum).getInt());
break;
}
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&) { }
@@ -468,16 +470,16 @@ void Dht::searchSendAnnounceValue(const Sp<Search>& sr) {
}
}
};
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,
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)
if (not n.candidate and ++i == TARGET_NODES)
break;
}
}
@@ -588,8 +590,8 @@ Dht::searchStep(Sp<Search> sr)
// 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);
sn.getStatus.erase(get.query);
sn.pagination_queries.erase(get.query);
}
/* clearing callbacks for announced values */
@@ -608,10 +610,10 @@ Dht::searchStep(Sp<Search> sr)
if (not sr->listeners.empty()) {
unsigned i = 0;
for (auto& n : sr->nodes) {
if (not n->isSynced(now))
if (not n.isSynced(now))
continue;
searchSynchedNodeListen(sr, *n);
if (not n->candidate and ++i == LISTEN_NODES)
searchSynchedNodeListen(sr, n);
if (not n.candidate and ++i == LISTEN_NODES)
break;
}
}
@@ -776,8 +778,8 @@ Dht::announce(const InfoHash& id,
if (a_sr == sr->announce.end()) {
sr->announce.emplace_back(Announce {permanent, value, created, callback});
for (auto& n : sr->nodes) {
n->probe_query.reset();
n->acked[value->id].first.reset();
n.probe_query.reset();
n.acked[value->id].first.reset();
}
} else {
a_sr->permanent = permanent;
@@ -785,8 +787,8 @@ Dht::announce(const InfoHash& id,
if (a_sr->value != value) {
a_sr->value = value;
for (auto& n : sr->nodes) {
n->acked[value->id].first.reset();
n->probe_query.reset();
n.acked[value->id].first.reset();
n.probe_query.reset();
}
}
if (sr->isAnnounced(value->id)) {
@@ -1354,9 +1356,9 @@ Dht::connectivityChanged(sa_family_t af)
network_engine.connectivityChanged(af);
for (auto& sp : searches(af))
for (auto& sn : sp.second->nodes) {
for (auto& ls : sn->listenStatus)
sn->node->cancelRequest(ls.second.req);
sn->listenStatus.clear();
for (auto& ls : sn.listenStatus)
sn.node->cancelRequest(ls.second.req);
sn.listenStatus.clear();
}
reported_addr.erase(std::remove_if(reported_addr.begin(), reported_addr.end(), [&](const ReportedAddr& addr){
return addr.second.getFamily() == af;
@@ -1503,42 +1505,42 @@ Dht::dumpSearch(const Search& sr, std::ostream& out) const
auto last_get = sr.getLastGetTime();
for (const auto& n : sr.nodes) {
i++;
out << std::setfill (' ') << std::setw(3) << InfoHash::commonBits(sr.id, n->node->id) << ' ' << n->node->id;
out << ' ' << (findNode(n->node->id, sr.af) ? '*' : ' ');
out << std::setfill (' ') << std::setw(3) << InfoHash::commonBits(sr.id, n.node->id) << ' ' << n.node->id;
out << ' ' << (findNode(n.node->id, sr.af) ? '*' : ' ');
out << " [";
if (auto pendingCount = n->node->getPendingMessageCount())
if (auto pendingCount = n.node->getPendingMessageCount())
out << pendingCount;
else
out << ' ';
out << (n->node->isExpired() ? 'x' : ' ') << "]";
out << (n.node->isExpired() ? 'x' : ' ') << "]";
// Get status
{
char g_i = n->pending(n->getStatus) ? (n->candidate ? 'c' : 'f') : ' ';
char s_i = n->isSynced(now) ? (n->last_get_reply > last_get ? 'u' : 's') : '-';
char g_i = n.pending(n.getStatus) ? (n.candidate ? 'c' : 'f') : ' ';
char s_i = n.isSynced(now) ? (n.last_get_reply > last_get ? 'u' : 's') : '-';
out << " [" << s_i << g_i << "] ";
}
// Listen status
if (not sr.listeners.empty()) {
if (n->listenStatus.empty())
if (n.listenStatus.empty())
out << " ";
else
out << "["
<< (n->isListening(now) ? 'l' : (n->pending(n->listenStatus) ? 'f' : ' ')) << "] ";
<< (n.isListening(now) ? 'l' : (n.pending(n.listenStatus) ? 'f' : ' ')) << "] ";
}
// Announce status
if (not sr.announce.empty()) {
if (n->acked.empty()) {
if (n.acked.empty()) {
out << " ";
for (size_t a=0; a < sr.announce.size(); a++)
out << ' ';
} else {
out << "[";
for (const auto& a : sr.announce) {
auto ack = n->acked.find(a.value->id);
if (ack == n->acked.end() or not ack->second.first) {
auto ack = n.acked.find(a.value->id);
if (ack == n.acked.end() or not ack->second.first) {
out << ' ';
} else {
out << ack->second.first->getStateChar();
@@ -1547,7 +1549,7 @@ Dht::dumpSearch(const Search& sr, std::ostream& out) const
out << "] ";
}
}
out << n->node->getAddrStr() << std::endl;
out << n.node->getAddrStr() << std::endl;
}
}
@@ -1583,11 +1585,6 @@ Dht::getStorageLog() const
std::stringstream out;
for (const auto& s : store)
out << printStorageLog(s);
out << "Total " << store.size() << " storages, " << total_values << " values (";
if (total_store_size < 1024)
out << total_store_size << " bytes)";
else
out << (total_store_size/1024) << " KB)";
out << std::endl << std::endl;
std::multimap<size_t, const SockAddr*> q_map;
for (const auto& ip : store_quota)
@@ -1596,6 +1593,12 @@ Dht::getStorageLog() const
for (auto ip = q_map.rbegin(); ip != q_map.rend(); ++ip)
out << "IP " << ip->second->toString() << " uses " << ip->first << " bytes" << std::endl;
out << std::endl;
out << "Total " << store.size() << " storages, " << total_values << " values (";
if (total_store_size < 1024)
out << total_store_size << " bytes)";
else
out << (total_store_size/1024) << " / " << (max_store_size/1024) << " KB)";
out << std::endl;
return out.str();
}
@@ -2094,9 +2097,9 @@ Dht::onError(Sp<net::Request> req, net::DhtProtocolException e) {
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();
if (n.node != node) continue;
n.token.clear();
n.last_get_reply = time_point::min();
searchSendGetValues(sr);
scheduler.edit(sr->nextSearchStep, scheduler.time());
break;
+7 -7
View File
@@ -413,7 +413,7 @@ DhtRunner::loop_()
while (not received.empty()) {
auto& pck = received.front();
auto delay = clock::now() - pck.received;
if (delay > std::chrono::seconds(1))
if (delay > std::chrono::milliseconds(500))
std::cerr << "Dropping packet with high delay: " << print_dt(delay) << std::endl;
else
wakeup = dht->periodic(pck.data.data(), pck.data.size()-1, pck.from);
@@ -526,7 +526,7 @@ DhtRunner::startNetwork(const SockAddr sin4, const SockAddr sin6)
std::lock_guard<std::mutex> lck(sock_mtx);
if (rcv.size() >= RX_QUEUE_MAX_SIZE) {
std::cerr << "Dropping packet: queue is full!" << std::endl;
continue;
rcv.pop();
}
rcv.emplace(ReceivedPacket {Blob {buf.begin(), buf.begin()+rc+1}, SockAddr(from, from_len), clock::now()});
}
@@ -583,6 +583,7 @@ DhtRunner::listen(InfoHash hash, ValueCallback vcb, Value::Filter f, Where w)
{
std::lock_guard<std::mutex> lck(storage_mtx);
pending_ops.emplace([=](SecureDht& dht) mutable {
#if OPENDHT_PROXY_CLIENT
auto tokenbGlobal = listener_token_++;
Listener listener {};
listener.hash = hash;
@@ -590,21 +591,20 @@ DhtRunner::listen(InfoHash hash, ValueCallback vcb, Value::Filter f, Where w)
listener.w = std::move(w);
listener.gcb = [hash,vcb,tokenbGlobal,this](const std::vector<Sp<Value>>& vals, bool expired){
if (not vcb(vals, expired)) {
#if OPENDHT_PROXY_CLIENT
cancelListen(hash, tokenbGlobal);
#endif
return false;
}
return true;
};
#if OPENDHT_PROXY_CLIENT
if (use_proxy)
listener.tokenProxyDht = dht.listen(hash, listener.gcb, listener.f, listener.w);
else
#endif
listener.tokenClassicDht = dht.listen(hash, listener.gcb, listener.f, listener.w);
listeners_.emplace(tokenbGlobal, std::move(listener));
ret_token->set_value(tokenbGlobal);
#else
ret_token->set_value(dht.listen(hash, vcb, f, w));
#endif
});
}
cv.notify_all();
@@ -898,8 +898,8 @@ DhtRunner::findCertificate(InfoHash hash, std::function<void(const std::shared_p
void
DhtRunner::resetDht()
{
listeners_.clear();
#if OPENDHT_PROXY_CLIENT
listeners_.clear();
dht_via_proxy_.reset();
#endif // OPENDHT_PROXY_CLIENT
dht_.reset();
+59 -49
View File
@@ -91,9 +91,9 @@ struct Dht::SearchNode {
SearchNode() : node() {}
SearchNode(const SearchNode&) = delete;
SearchNode(SearchNode&&) = delete;
SearchNode(SearchNode&&) = default;
SearchNode& operator=(const SearchNode&) = delete;
SearchNode& operator=(SearchNode&&) = delete;
SearchNode& operator=(SearchNode&&) = default;
SearchNode(const Sp<Node>& node) : node(node) {}
~SearchNode() {
@@ -136,7 +136,7 @@ struct Dht::SearchNode {
*
* @return true if we can send get, else false.
*/
bool canGet(time_point now, time_point update, Sp<Query> q = {}) const {
bool canGet(time_point now, time_point update, const Sp<Query>& q) const {
if (node->isExpired())
return false;
@@ -147,9 +147,9 @@ struct Dht::SearchNode {
if (s.second and s.second->pending())
pending = true;
if (s.first and q and q->isSatisfiedBy(*s.first) and s.second) {
if (s.second->pending() and not pending_sq_status)
if (s.second->pending())
pending_sq_status = true;
if (s.second->completed() and not (update > s.second->reply_time) and not completed_sq_status)
else if (s.second->completed() and not (update > s.second->reply_time))
completed_sq_status = true;
if (completed_sq_status and pending_sq_status)
break;
@@ -157,7 +157,7 @@ struct Dht::SearchNode {
}
return (not pending and now > last_get_reply + Node::NODE_EXPIRE_TIME) or
not (hasStartedPagination(q) or completed_sq_status or pending_sq_status);
not (completed_sq_status or pending_sq_status or hasStartedPagination(q));
}
/**
@@ -369,7 +369,7 @@ struct Dht::Search {
bool expired {false}; /* no node, or all nodes expired */
bool done {false}; /* search is over, cached for later */
std::vector<std::unique_ptr<SearchNode>> nodes {};
std::vector<SearchNode> nodes {};
/* pending puts */
std::vector<Announce> announce {};
@@ -404,17 +404,17 @@ struct Dht::Search {
bool insertNode(const Sp<Node>& n, time_point now, const Blob& token={});
SearchNode* getNode(const Sp<Node>& n) {
for (auto& sn : nodes)
if (sn->node == n)
return sn.get();
return nullptr;
auto srn = std::find_if(nodes.begin(), nodes.end(), [&](SearchNode& sn) {
return n == sn.node;
});
return (srn == nodes.end()) ? nullptr : &(*srn);
}
/* number of concurrent sync requests */
unsigned currentlySolicitedNodeCount() const {
unsigned count = 0;
for (const auto& n : nodes)
if (not n->isBad() and n->pendingGet())
if (not n.isBad() and n.pendingGet())
count++;
return count;
}
@@ -430,7 +430,8 @@ struct Dht::Search {
*
* @param query The query identifying a 'get' request.
*/
time_point getLastGetTime(Sp<Query> query = {}) const;
time_point getLastGetTime(const Query&) const;
time_point getLastGetTime() const;
/**
* Is this get operation done ?
@@ -447,12 +448,12 @@ struct Dht::Search {
*/
void setDone(const Get& get) {
for (auto& n : nodes) {
auto pqs = n->pagination_queries.find(get.query);
if (pqs != n->pagination_queries.cend()) {
auto pqs = n.pagination_queries.find(get.query);
if (pqs != n.pagination_queries.cend()) {
for (auto& pq : pqs->second)
n->getStatus.erase(pq);
n.getStatus.erase(pq);
}
n->getStatus.erase(get.query);
n.getStatus.erase(get.query);
}
if (get.done_cb)
get.done_cb(true, getNodes());
@@ -465,9 +466,9 @@ struct Dht::Search {
*/
void setDone() {
for (auto& n : nodes) {
n->getStatus.clear();
n->listenStatus.clear();
n->acked.clear();
n.getStatus.clear();
n.listenStatus.clear();
n.acked.clear();
}
done = true;
}
@@ -499,9 +500,9 @@ struct Dht::Search {
}
for (auto& sn : nodes) {
if (listeners.empty())
sn->cancelListen();
sn.cancelListen();
else if (query)
sn->cancelListen(query);
sn.cancelListen(query);
}
});
scheduler.edit(opExpirationJob, nextExpire);
@@ -513,14 +514,14 @@ struct Dht::Search {
* @return The number of non-good search nodes.
*/
unsigned getNumberOfBadNodes() const {
return std::count_if(nodes.begin(), nodes.end(), [](const std::unique_ptr<SearchNode>& sn) {
return sn->isBad();
return std::count_if(nodes.begin(), nodes.end(), [](const SearchNode& sn) {
return sn.isBad();
});
}
unsigned getNumberOfConsecutiveBadNodes() const {
unsigned count = 0;
std::find_if(nodes.begin(), nodes.end(), [&count](const std::unique_ptr<SearchNode>& sn) {
if (not sn->isBad())
std::find_if(nodes.begin(), nodes.end(), [&count](const SearchNode& sn) {
if (not sn.isBad())
return true;
++count;
return false;
@@ -539,7 +540,7 @@ struct Dht::Search {
*/
bool removeExpiredNode(const time_point& now) {
for (auto e = nodes.cend(); e != nodes.cbegin();) {
const Node& n = *(*(--e))->node;
const Node& n = *(--e)->node;
if (n.isRemovable(now)) {
//std::cout << "Removing expired node " << n.id << " from IPv" << (af==AF_INET?'4':'6') << " search " << id << std::endl;
nodes.erase(e);
@@ -612,7 +613,7 @@ struct Dht::Search {
// remove acked for cleared annouces
for (auto it = announced; it != announce.end(); ++it) {
for (auto& n : nodes)
n->acked.erase(it->value->id);
n.acked.erase(it->value->id);
}
announce.erase(announced, announce.end());
}
@@ -645,13 +646,13 @@ Dht::Search::insertNode(const Sp<Node>& snode, time_point now, const Blob& token
auto n = nodes.end();
while (n != nodes.begin()) {
--n;
if ((*n)->node == snode) {
if (n->node == snode) {
found = true;
break;
}
/* Node not found. We could insert it after this one. */
if (id.xorCmp(nid, (*n)->node->id) > 0) {
if (id.xorCmp(nid, n->node->id) > 0) {
++n;
break;
}
@@ -675,7 +676,7 @@ Dht::Search::insertNode(const Sp<Node>& snode, time_point now, const Blob& token
full = nodes.size() - bad >= SEARCH_NODES;
while (std::distance(nodes.cbegin(), t) - bad > SEARCH_NODES) {
--t;
if ((*t)->isBad())
if (t->isBad())
bad--;
}
}
@@ -691,7 +692,7 @@ Dht::Search::insertNode(const Sp<Node>& snode, time_point now, const Blob& token
if (nodes.empty()) {
step_time = time_point::min();
}
n = nodes.insert(n, std::unique_ptr<SearchNode>(new SearchNode(snode)));
n = nodes.insert(n, SearchNode(snode));
node.setTime(now);
new_search_node = true;
if (node.isExpired()) {
@@ -703,16 +704,16 @@ Dht::Search::insertNode(const Sp<Node>& snode, time_point now, const Blob& token
}
while (nodes.size() - bad > SEARCH_NODES) {
if (not expired and nodes.back()->isBad())
if (not expired and nodes.back().isBad())
bad--;
nodes.pop_back();
}
}
if (not token.empty()) {
(*n)->candidate = false;
(*n)->last_get_reply = now;
n->candidate = false;
n->last_get_reply = now;
if (token.size() <= 64)
(*n)->token = token;
n->token = token;
expired = false;
}
if (new_search_node)
@@ -726,7 +727,7 @@ Dht::Search::getNodes() const
std::vector<Sp<Node>> ret {};
ret.reserve(nodes.size());
for (const auto& sn : nodes)
ret.emplace_back(sn->node);
ret.emplace_back(sn.node);
return ret;
}
@@ -735,9 +736,9 @@ Dht::Search::isSynced(time_point now) const
{
unsigned i = 0;
for (const auto& n : nodes) {
if (n->isBad())
if (n.isBad())
continue;
if (not n->isSynced(now))
if (not n.isSynced(now))
return false;
if (++i == TARGET_NODES)
break;
@@ -746,11 +747,20 @@ Dht::Search::isSynced(time_point now) const
}
time_point
Dht::Search::getLastGetTime(Sp<Query> q) const
Dht::Search::getLastGetTime(const Query& q) const
{
time_point last = time_point::min();
for (const auto& g : callbacks)
last = std::max(last, (not q or q->isSatisfiedBy(*g.second.query) ? g.second.start : time_point::min()));
last = std::max(last, (q.isSatisfiedBy(*g.second.query) ? g.second.start : time_point::min()));
return last;
}
time_point
Dht::Search::getLastGetTime() const
{
time_point last = time_point::min();
for (const auto& g : callbacks)
last = std::max(last, g.second.start);
return last;
}
@@ -759,9 +769,9 @@ Dht::Search::isDone(const Get& get) const
{
unsigned i = 0;
for (const auto& sn : nodes) {
if (sn->isBad())
if (sn.isBad())
continue;
if (not sn->isDone(get))
if (not sn.isDone(get))
return false;
if (++i == TARGET_NODES)
break;
@@ -776,9 +786,9 @@ Dht::Search::isAnnounced(Value::Id id) const
return false;
unsigned i = 0;
for (const auto& n : nodes) {
if (n->isBad())
if (n.isBad())
continue;
if (not n->isAnnounced(id))
if (not n.isAnnounced(id))
return false;
if (++i == TARGET_NODES)
return true;
@@ -793,14 +803,14 @@ Dht::Search::isListening(time_point now) const
return false;
unsigned i = 0;
for (const auto& n : nodes) {
if (n->isBad())
if (n.isBad())
continue;
SearchNode::NodeListenerStatus::const_iterator ls {};
for (ls = n->listenStatus.begin(); ls != n->listenStatus.end() ; ++ls) {
if (n->isListening(now, ls))
for (ls = n.listenStatus.begin(); ls != n.listenStatus.end() ; ++ls) {
if (n.isListening(now, ls))
break;
}
if (ls == n->listenStatus.end())
if (ls == n.listenStatus.end())
return false;
if (++i == LISTEN_NODES)
break;
+1 -4
View File
@@ -116,10 +116,7 @@ DhtProxyTester::testListen() {
// All values
dht::Value secondVal {"You're a monster"};
auto secondVal_data = secondVal.data;
nodePeer.put(key, std::move(secondVal), [&](bool) {
done = true;
cv.notify_all();
});
nodePeer.put(key, std::move(secondVal));
cv.wait_for(lk, std::chrono::seconds(10), [&]{ return done; });
// Here values should contains 3 values
CPPUNIT_ASSERT_EQUAL(static_cast<int>(values.size()), 2);