dht: move socket to SearchNode

include/opendht/network_engine.h

@@ -211,6 +211,7 @@ private:
 
 public:
     using RequestCb = std::function<void(const Request&, RequestAnswer&&)>;
+    using RequestErrorCb = std::function<bool(const Request&, DhtProtocolException&&)>;
     using RequestExpiredCb = std::function<void(const Request&, bool)>;
 
     NetworkEngine(const Sp<Logger>& log, std::mt19937_64& rd, Scheduler& scheduler, std::unique_ptr<DatagramSocket>&& sock);
@@ -361,10 +362,9 @@ public:
                            const InfoHash& hash,
                            const Query& query,
                            const Blob& token,
-                           Sp<Request> previous,
+                           Tid socketId,
                            RequestCb&& on_done,
-                           RequestExpiredCb&& on_expired,
-                           SocketCb&& socket_cb);
+                           RequestExpiredCb&& on_expired);
     /**
      * Send a "announce" request to a given node.
      *
@@ -403,6 +403,7 @@ public:
                                  const Value::Id& vid,
                                  const Blob& token,
                                  RequestCb&& on_done,
+                                 RequestErrorCb&& on_error,
                                  RequestExpiredCb&& on_expired);
     /**
      * Send a "update" request to a given node. Used for Listen operations

src/dht.cpp

@@ -490,7 +490,29 @@ void Dht::searchSendAnnounceValue(const Sp<Search>& sr) {
                     logger_->w(sr->id, sn->node->id, "[search %s] [node %s] sending 'refresh' (vid: %d)",
                         sr->id.toString().c_str(), sn->node->toString().c_str(), a.value->id);
                 sn->acked[a.value->id] = {
-                    network_engine.sendRefreshValue(sn->node, sr->id, a.value->id, sn->token, onDone, onExpired),
+                    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_->e(node->id, "[node %s] returned error 404: storage not found", node->toString().c_str());
+                            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 {
@@ -569,33 +591,30 @@ Dht::searchSynchedNodeListen(const Sp<Search>& sr, SearchNode& n)
     std::weak_ptr<Search> ws = sr;
     for (const auto& l : sr->listeners) {
         const auto& query = l.second.query;
-        auto list_token = l.first;
-        if (n.getListenTime(query, listenExp) > scheduler.time())
+        
+        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());
 
-        auto r = n.listenStatus.find(query);
         if (r == n.listenStatus.end()) {
             r = n.listenStatus.emplace(std::piecewise_construct,
                 std::forward_as_tuple(query),
                 std::forward_as_tuple(
-                [ws,list_token](const std::vector<Sp<Value>>& values, bool expired){
-                    if (auto sr = ws.lock()) {
-                        auto l = sr->listeners.find(list_token);
-                        if (l != sr->listeners.end()) {
-                            l->second.get_cb(values, expired);
+                    l.second.get_cb,
+                    l.second.sync_cb,
+                    n.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());
+                            sr->insertNode(node, scheduler.time(), answer.ntoken);
+                            if (auto sn = sr->getNode(node)) {
+                                sn->onValues(query, std::move(answer), types, scheduler);
+                            }
                         }
-                    }
-                }, [ws,list_token] (ListenSyncStatus status) {
-                    if (auto sr = ws.lock()) {
-                        auto l = sr->listeners.find(list_token);
-                        if (l != sr->listeners.end()) {
-                            l->second.sync_cb(status);
-                        }
-                    }
-                })).first;
+                    }))).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)) {
@@ -604,8 +623,7 @@ Dht::searchSynchedNodeListen(const Sp<Search>& sr, SearchNode& n)
                 }
             });
         }
-        auto prev_req = r != n.listenStatus.end() ? r->second.req : nullptr;
-        auto new_req = network_engine.sendListen(n.node, sr->id, *query, n.token, prev_req,
+        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()) {
@@ -625,16 +643,6 @@ Dht::searchSynchedNodeListen(const Sp<Search>& sr, SearchNode& n)
                         if (auto sn = sr->getNode(req.node))
                             sn->listenStatus.erase(query);
                 }
-            },
-            [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());
-                    sr->insertNode(node, scheduler.time(), answer.ntoken);
-                    if (auto sn = sr->getNode(node)) {
-                        sn->onValues(query, std::move(answer), types, scheduler);
-                    }
-                }
             }
         );
         // Here the request may have failed and the CachedListenStatus removed
@@ -2175,7 +2183,6 @@ Dht::onError(Sp<net::Request> req, net::DhtProtocolException e) {
         if (logger_)
             logger_->e(node->id, "[node %s] token flush", node->toString().c_str());
         node->authError();
-        node->cancelRequest(req);
         for (auto& srp : searches(node->getFamily())) {
             auto& sr = srp.second;
             for (auto& n : sr->nodes) {

src/network_engine.cpp

@@ -560,7 +560,8 @@ NetworkEngine::process(std::unique_ptr<ParsedMessage>&& msg, const SockAddr& fro
             {
                 req->last_try = time_point::min();
                 req->reply_time = time_point::min();
-                onError(req, DhtProtocolException {msg->error_code});
+                if (not req->setError(DhtProtocolException {msg->error_code}))
+                    onError(req, DhtProtocolException {msg->error_code});
             } else {
                 if (logIncoming_)
                     if (logger_)
@@ -1072,27 +1073,11 @@ NetworkEngine::sendListen(Sp<Node> n,
         const InfoHash& hash,
         const Query& query,
         const Blob& token,
-        Sp<Request> previous,
+        Tid socketId,
         RequestCb&& on_done,
-        RequestExpiredCb&& on_expired,
-        SocketCb&& socket_cb)
+        RequestExpiredCb&& on_expired)
 {
-    Tid socket;
     Tid tid (n->getNewTid());
-    if (previous and previous->node == n) {
-        socket = previous->getSocket();
-    } else {
-        if (previous)
-            if (logger_)
-                logger_->e(hash, "[node %s] trying refresh listen contract with wrong node", previous->node->toString().c_str());
-        socket = n->openSocket(std::move(socket_cb));
-    }
-
-    if (not socket) {
-        if (logger_)
-            logger_->e(hash, "[node %s] unable to get a valid socket for listen. Aborting listen", n->toString().c_str());
-        return {};
-    }
     msgpack::sbuffer buffer;
     msgpack::packer<msgpack::sbuffer> pk(&buffer);
     pk.pack_map(5+(config.network?1:0));
@@ -1103,7 +1088,7 @@ NetworkEngine::sendListen(Sp<Node> n,
       pk.pack(KEY_VERSION);   pk.pack(1);
       pk.pack(KEY_REQ_H);     pk.pack(hash);
       pk.pack(KEY_REQ_TOKEN); packToken(pk, token);
-      pk.pack(KEY_REQ_SID);   pk.pack(socket);
+      pk.pack(KEY_REQ_SID);   pk.pack(socketId);
       if (has_query) {
           pk.pack(KEY_REQ_QUERY); pk.pack(query);
       }
@@ -1125,8 +1110,7 @@ NetworkEngine::sendListen(Sp<Node> n,
         [=](const Request& req_status, bool done) { /* on expired */
             if (on_expired)
                 on_expired(req_status, done);
-        },
-        socket
+        }
     );
     sendRequest(req);
     ++out_stats.listen;
@@ -1263,6 +1247,7 @@ NetworkEngine::sendRefreshValue(Sp<Node> n,
                 const Value::Id& vid,
                 const Blob& token,
                 RequestCb&& on_done,
+                RequestErrorCb&& on_error,
                 RequestExpiredCb&& on_expired)
 {
     Tid tid (n->getNewTid());
@@ -1298,6 +1283,7 @@ NetworkEngine::sendRefreshValue(Sp<Node> n,
                 }
             }
         },
+        on_error,
         [=](const Request& req_status, bool done) { /* on expired */
             if (on_expired) {
                 on_expired(req_status, done);

src/node.cpp

@@ -116,7 +116,6 @@ Node::cancelRequest(const Sp<net::Request>& req)
 {
     if (req) {
         req->cancel();
-        closeSocket(req->closeSocket());
         requests_.erase(req->getTid());
     }
 }

src/request.h

@@ -27,6 +27,7 @@ struct Node;
 namespace net {
 
 class NetworkEngine;
+class DhtProtocolException;
 struct ParsedMessage;
 
 /*!
@@ -72,16 +73,19 @@ struct Request {
             Sp<Node> node,
             Blob&& msg,
             std::function<void(const Request&, ParsedMessage&&)> on_done,
-            std::function<void(const Request&, bool)> on_expired,
-            Tid socket = 0) :
-        node(node), tid(tid), type(type), on_done(on_done), on_expired(on_expired), msg(std::move(msg)), socket(socket) { }
+            std::function<void(const Request&, bool)> on_expired) :
+        node(node), tid(tid), type(type), on_done(on_done), on_expired(on_expired), msg(std::move(msg)) { }
+    Request(MessageType type, Tid tid,
+            Sp<Node> node,
+            Blob&& msg,
+            std::function<void(const Request&, ParsedMessage&&)> on_done,
+            std::function<bool(const Request&, DhtProtocolException&&)> on_error,
+            std::function<void(const Request&, bool)> on_expired) :
+        node(node), tid(tid), type(type), on_done(on_done), on_error(on_error), on_expired(on_expired), msg(std::move(msg)) { }
 
     Tid getTid() const { return tid; }
     MessageType getType() const { return type; }
 
-    Tid getSocket() { return socket; }
-    Tid closeSocket() { auto ret = socket; socket = 0; return ret; }
-
     void setExpired() {
         if (pending()) {
             state_ = Request::State::EXPIRED;
@@ -96,6 +100,15 @@ struct Request {
             clear();
         }
     }
+    bool setError(DhtProtocolException&& e) {
+        if (pending()) {
+            state_ = Request::State::EXPIRED;
+            bool handled = on_error and on_error(*this, std::forward<DhtProtocolException>(e));
+            clear();
+            return handled;
+        }
+        return true;
+    }
 
     void cancel() {
         if (pending()) {
@@ -113,6 +126,7 @@ private:
 
     void clear() {
         on_done = {};
+        on_error = {};
         on_expired = {};
         msg = {};
         parts = {};
@@ -128,11 +142,11 @@ private:
     time_point last_try {time_point::min()};   /* time of the last attempt to process the request. */
 
     std::function<void(const Request&, ParsedMessage&&)> on_done {};
+    std::function<bool(const Request&, DhtProtocolException&&)> on_error {};
     std::function<void(const Request&, bool)> on_expired {};
 
     Blob msg {};                      /* the serialized message. */
     std::vector<Blob> parts;
-    Tid socket;   /* the socket used for further reponses. */
 };
 
 } /* namespace net  */

src/search.h

@@ -65,10 +65,16 @@ struct Dht::SearchNode {
         ValueCache cache;
         Sp<Scheduler::Job> cacheExpirationJob {};
         Sp<net::Request> req {};
-        CachedListenStatus(ValueStateCallback&& cb, SyncCallback&& scb)
-         : cache(std::forward<ValueStateCallback>(cb), std::forward<SyncCallback>(scb)) {}
+        Tid socketId {0};
+        CachedListenStatus(ValueStateCallback&& cb, SyncCallback scb, Tid sid)
+         : cache(std::forward<ValueStateCallback>(cb), std::forward<SyncCallback>(scb)), socketId(sid) {}
         CachedListenStatus(CachedListenStatus&&) = delete;
         CachedListenStatus(const CachedListenStatus&) = delete;
+        ~CachedListenStatus() {
+            if (socketId and req and req->node) {
+                req->node->closeSocket(socketId);
+            }
+        }
     };
     using NodeListenerStatus = std::map<Sp<Query>, CachedListenStatus>;
 
@@ -344,13 +350,15 @@ struct Dht::SearchNode {
      * Assuming the node is synced, should the "listen" request with Query q be
      * sent to this node now ?
      */
-    time_point getListenTime(const Sp<Query>& q, duration listen_expire) const {
-        auto listen_status = listenStatus.find(q);
-        if (listen_status == listenStatus.end() or not listen_status->second.req)
+    time_point getListenTime(const decltype(listenStatus)::const_iterator listen_status, duration listen_expire) const {
+        if (listen_status == listenStatus.cend() or not listen_status->second.req)
             return time_point::min();
         return listen_status->second.req->pending() ? time_point::max() :
             listen_status->second.req->reply_time + listen_expire - REANNOUNCE_MARGIN;
     }
+    time_point getListenTime(const Sp<Query>& q, duration listen_expire) const {
+        return getListenTime(listenStatus.find(q), listen_expire);
+    }
 
     /**
      * Is this node expired or candidate