Revert "search: use pointer to SearchNode"

This reverts commit 3a41f2ce56.

include/opendht/callbacks.h

@@ -83,6 +83,16 @@ struct OPENDHT_PUBLIC NodeInfo {
     MSGPACK_DEFINE_MAP(id, node_id, ipv4, ipv6)
 };
 
+struct OPENDHT_PUBLIC MessageStats {
+    unsigned packets {0};
+    unsigned ping    {0};
+    unsigned find    {0};
+    unsigned get     {0};
+    unsigned put     {0};
+    unsigned listen  {0};
+    unsigned refresh {0};
+};
+
 /**
  * Dht configuration.
  */

include/opendht/network_engine.h

@@ -27,6 +27,7 @@
 #include "utils.h"
 #include "rng.h"
 #include "rate_limiter.h"
+#include "callbacks.h"
 
 #include <vector>
 #include <string>
@@ -408,6 +409,10 @@ public:
         return stats;
     }
 
+    MessageStats getMessageStats(bool in) {
+        return in ? in_stats : out_stats;
+    }
+
     void blacklistNode(const Sp<Node>& n);
 
     std::vector<Sp<Node>> getCachedNodes(const InfoHash& id, sa_family_t sa_f, size_t count) {
@@ -457,16 +462,6 @@ private:
      */
     void sendRequest(const Sp<Request>& request);
 
-    struct MessageStats {
-        unsigned ping    {0};
-        unsigned find    {0};
-        unsigned get     {0};
-        unsigned put     {0};
-        unsigned listen  {0};
-        unsigned refresh {0};
-    };
-
-
     // basic wrapper for socket sendto function
     int send(const char *buf, size_t len, int flags, const SockAddr& addr);
 

src/dht.cpp

@@ -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;
     }
 }
 
@@ -2094,9 +2096,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;

src/dhtrunner.cpp

@@ -410,15 +410,25 @@ DhtRunner::loop_()
         received = std::move(rcv);
     }
     if (not received.empty()) {
+        size_t packet_num = received.size();
+        auto start = clock::now();
         while (not received.empty()) {
             auto& pck = received.front();
-            auto delay = clock::now() - pck.received;
+            auto now = clock::now();
+            auto delay = now - pck.received;
             if (delay > std::chrono::seconds(1))
                 std::cerr << "Dropping packet with high delay: " << print_dt(delay) << std::endl;
-            else
+            else {
                 wakeup = dht->periodic(pck.data.data(), pck.data.size()-1, pck.from);
+                auto end = clock::now();
+                std::cerr << "periodic took " << print_dt(end - now)*1000 << std::endl;
+            }
             received.pop();
         }
+        if (packet_num > 1) {
+            auto time_all = print_dt(clock::now() - start);
+            std::cerr << "Treating " << packet_num << " packets took " << time_all << " - " << (packet_num/time_all) << " packets/sec" << std::endl;
+        }
     } else {
         wakeup = dht->periodic(nullptr, 0, nullptr, 0);
     }

src/network_engine.cpp

@@ -403,6 +403,7 @@ NetworkEngine::isNodeBlacklisted(const SockAddr& addr) const
 void
 NetworkEngine::processMessage(const uint8_t *buf, size_t buflen, const SockAddr& from)
 {
+    in_stats.packets++;
     if (isMartian(from)) {
         DHT_LOG.w("Received packet from martian node %s", from.toString().c_str());
         return;
@@ -663,6 +664,8 @@ NetworkEngine::send(const char *buf, size_t len, int flags, const SockAddr& addr
 #ifdef MSG_NOSIGNAL
     flags |= MSG_NOSIGNAL;
 #endif
+
+    out_stats.packets++;
     if (sendto(s, buf, len, flags, addr.get(), addr.getLength()) == -1) {
         int err = errno;
         DHT_LOG.e("Can't send message to %s: %s", addr.toString().c_str(), strerror(err));

src/search.h

@@ -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;