diff --git a/include/opendht/network_engine.h b/include/opendht/network_engine.h index a46b5918..0dde4a6b 100644 --- a/include/opendht/network_engine.h +++ b/include/opendht/network_engine.h @@ -200,7 +200,6 @@ private: const Value::Id&)> onRefresh {}; public: - using SocketCb = std::function&, RequestAnswer&&)>; using RequestCb = std::function; using RequestExpiredCb = std::function; @@ -386,23 +385,6 @@ public: const Blob& token, RequestCb&& on_done, RequestExpiredCb&& on_expired); - /** - * Opens a socket on which a node will be able allowed to write for further - * additionnal updates following the response to a previous request. - * - * @param node The node which will be allowed to write on this socket. - * @param cb The callback to execute once updates arrive on the socket. - * - * @return the socket. - */ - Sp openSocket(const Sp& node, TransPrefix tp, SocketCb&& cb); - - /** - * Closes a socket so that no further data will be red on that socket. - * - * @param socket The socket to close. - */ - void closeSocket(Sp socket); /** * Parses a message and calls appropriate callbacks. @@ -557,7 +539,6 @@ private: uint16_t transaction_id {1}; std::map> requests {}; std::map partial_messages; - std::map> opened_sockets {}; MessageStats in_stats {}, out_stats {}; std::set blacklist {}; diff --git a/include/opendht/node.h b/include/opendht/node.h index 15791a41..a36b4c51 100644 --- a/include/opendht/node.h +++ b/include/opendht/node.h @@ -22,13 +22,19 @@ #include "infohash.h" // includes socket structures #include "utils.h" #include "sockaddr.h" +#include "net.h" #include +#include namespace dht { +struct Node; namespace net { struct Request; +struct Socket; +struct RequestAnswer; +using SocketCb = std::function&, RequestAnswer&&)>; } /* namespace net */ struct Node { @@ -84,6 +90,26 @@ struct Node { void setExpired(); + /** + * Opens a socket on which a node will be able allowed to write for further + * additionnal updates following the response to a previous request. + * + * @param node The node which will be allowed to write on this socket. + * @param cb The callback to execute once updates arrive on the socket. + * + * @return the socket. + */ + Sp openSocket(const net::TransId& id, net::SocketCb&& cb); + + Sp getSocket(const net::TransId& id) const; + + /** + * Closes a socket so that no further data will be red on that socket. + * + * @param socket The socket to close. + */ + void closeSocket(const Sp& socket); + /** * Resets the state of the node so it's not expired anymore. */ @@ -106,6 +132,8 @@ private: static const constexpr unsigned MAX_AUTH_ERRORS {3}; std::list> requests_ {}; + std::map> sockets_ {}; + unsigned auth_errors {0}; bool expired_ {false}; diff --git a/src/network_engine.cpp b/src/network_engine.cpp index ca50133c..73991902 100644 --- a/src/network_engine.cpp +++ b/src/network_engine.cpp @@ -232,29 +232,12 @@ NetworkEngine::isRunning(sa_family_t af) const } } -Sp -NetworkEngine::openSocket(const Sp& node, TransPrefix tp, SocketCb&& cb) -{ - auto tid = TransId {tp, getNewTid()}; - auto s = opened_sockets.emplace(tid, std::make_shared(node, tid, cb)); - if (not s.second) - DHT_LOG.e(node->id, "[node %s] socket (tid: %d) already opened!", node->id.toString().c_str(), tid.toInt()); - return s.first->second; -} - -void -NetworkEngine::closeSocket(Sp socket) -{ - if (socket) - opened_sockets.erase(socket->id); -} - void NetworkEngine::cancelRequest(Sp& req) { if (req) { req->cancel(); - closeSocket(req->socket); + req->node->closeSocket(req->socket); requests.erase(req->tid); } } @@ -503,30 +486,23 @@ NetworkEngine::process(std::unique_ptr&& msg, const SockAddr& fro const auto& now = scheduler.time(); if (msg->type == MessageType::Error or msg->type == MessageType::Reply) { + auto node = cache.getNode(msg->id, from, now, true); + auto rsocket = node->getSocket(msg->tid); + //auto request = node->getRequest(msg->tid); + /* either response for a request or data for an opened socket */ auto req_it = requests.find(msg->tid); - auto rsocket_it = opened_sockets.end(); - if (req_it == requests.end()) - rsocket_it = opened_sockets.find(msg->tid); - if (req_it == requests.end() and rsocket_it == opened_sockets.end()) + if (req_it == requests.end() and not rsocket) { throw DhtProtocolException {DhtProtocolException::UNKNOWN_TID, "Can't find transaction", msg->id}; + } auto req = req_it != requests.end() ? req_it->second : nullptr; - auto rsocket = rsocket_it != opened_sockets.end() ? rsocket_it->second : nullptr; - - auto& node = req ? req->node : rsocket->node; - if (node->id != msg->id) { - if (node->id == zeroes) // received reply to a message sent when we didn't know the node ID. - node = cache.getNode(msg->id, from, now, true); - else { - // received reply from unexpected node - node->received(now, req); - onNewNode(node, 2); - DHT_LOG.w(node->id, "[node %s] message received from unexpected node", node->toString().c_str()); - return; - } - } else - node->update(from); + if (req and req->node != node and req->node->id != zeroes) { + node->received(now, req); + onNewNode(node, 2); + DHT_LOG.w(node->id, "[node %s] message received from unexpected node", node->toString().c_str()); + return; + } node->received(now, req); @@ -569,7 +545,7 @@ NetworkEngine::process(std::unique_ptr&& msg, const SockAddr& fro break; } else { /* request socket data */ deserializeNodes(*msg, from); - rsocket->on_receive(rsocket->node, std::move(*msg)); + rsocket->on_receive(node, std::move(*msg)); } default: break; @@ -624,9 +600,7 @@ NetworkEngine::process(std::unique_ptr&& msg, const SockAddr& fro case MessageType::Listen: { DHT_LOG.d(msg->info_hash, node->id, "[node %s] got 'listen' request for %s", node->toString().c_str(), msg->info_hash.toString().c_str()); ++in_stats.listen; - /* TODO: backward compatibility check to remove in a few versions (see ::sendListen doc) */ - auto socket_id = msg->socket_id.getTid() ? msg->socket_id : TransId { TransPrefix::GET_VALUES, msg->tid.getTid() }; - RequestAnswer answer = onListen(node, msg->info_hash, msg->token, socket_id.toInt(), std::move(msg->query)); + RequestAnswer answer = onListen(node, msg->info_hash, msg->token, msg->socket_id.toInt(), std::move(msg->query)); sendListenConfirmation(from, msg->tid); break; } @@ -1074,12 +1048,7 @@ NetworkEngine::sendListen(Sp n, } else { if (previous) DHT_LOG.e(hash, "[node %s] trying refresh listen contract with wrong node", previous->node->toString().c_str()); - /* TODO: Manually creating a socket for "listen" backward compatibility. - * As soon as this is not an issue anymore, switch to using - * ::createSocket function! */ - auto sid = TransId {TransPrefix::GET_VALUES, tid.getTid()}; - socket = std::make_shared(n, sid, socket_cb); - opened_sockets.emplace(sid, socket); + socket = n->openSocket(TransId {TransPrefix::GET_VALUES, getNewTid()}, std::move(socket_cb)); } if (not socket) { diff --git a/src/node.cpp b/src/node.cpp index 1c482af2..f4c3d0c8 100644 --- a/src/node.cpp +++ b/src/node.cpp @@ -104,9 +104,38 @@ Node::setExpired() r->setExpired(); } requests_.clear(); + sockets_.clear(); } +Sp +Node::openSocket(const net::TransId& tid, net::SocketCb&& cb) +{ + auto s = sockets_.emplace(tid, std::make_shared(tid, cb)); + //if (not s.second) + // DHT_LOG.e(id, "[node %s] socket (tid: %d) already opened!", id.toString().c_str(), tid.toInt()); + //else + // DHT_LOG.w("Opened socket (tid: %d), %lu opened", s.first->second->id, sockets_.size()); + return s.first->second; +} + + +Sp +Node::getSocket(const net::TransId& tid) const +{ + auto it = sockets_.find(tid); + return it == sockets_.end() ? nullptr : it->second; +} + +void +Node::closeSocket(const Sp& socket) +{ + if (socket) { + sockets_.erase(socket->id); + //DHT_LOG.w("Closing socket (tid: %d), %lu remaining", socket->id, sockets_.size()); + } +} + std::string Node::toString() const { diff --git a/src/request.h b/src/request.h index 0174e3b0..b377e610 100644 --- a/src/request.h +++ b/src/request.h @@ -20,17 +20,19 @@ #pragma once #include "net.h" +#include "value.h" namespace dht { +struct Node; namespace net { class NetworkEngine; struct ParsedMessage; +struct RequestAnswer; +using SocketCb = std::function&, RequestAnswer&&)>; -/*! - * @class Socket - * @brief Open route to a node for continous incoming packets. - * @details +/** + * Open route to a node for continous incoming packets. * A socket lets a remote node send us continuous packets treated using a * given callback. This is intended to provide an easy management of * specific updates nodes can send. For e.g, this is used in the case of the @@ -38,14 +40,10 @@ struct ParsedMessage; */ struct Socket { Socket() {} - Socket(std::shared_ptr node, - TransId id, - std::function&, ParsedMessage&&)> on_receive) : - node(node), id(id), on_receive(on_receive) {} - - std::shared_ptr node; + Socket(TransId id, SocketCb on_receive) : + id(id), on_receive(on_receive) {} TransId id; - std::function&, ParsedMessage&&)> on_receive {}; + SocketCb on_receive {}; }; /*! @@ -59,7 +57,7 @@ struct Socket { struct Request { friend class dht::net::NetworkEngine; - std::shared_ptr node {}; /* the node to whom the request is destined. */ + Sp node {}; /* the node to whom the request is destined. */ time_point reply_time {time_point::min()}; /* time when we received the response to the request. */ enum class State