diff --git a/CMakeLists.txt b/CMakeLists.txt index 0cbb6eb6..03d21b71 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -107,6 +107,7 @@ list (APPEND opendht_SOURCES src/storage.h src/listener.h src/search.h + src/net.h src/parsed_message.h src/request.h src/callbacks.cpp @@ -134,7 +135,6 @@ list (APPEND opendht_HEADERS include/opendht/routing_table.h include/opendht/node_cache.h include/opendht/network_engine.h - include/opendht/net.h include/opendht/scheduler.h include/opendht/rate_limiter.h include/opendht/securedht.h diff --git a/MSVC/opendht.vcxproj b/MSVC/opendht.vcxproj index cfa9fcc9..4c494bf5 100644 --- a/MSVC/opendht.vcxproj +++ b/MSVC/opendht.vcxproj @@ -47,7 +47,6 @@ - @@ -60,6 +59,7 @@ + diff --git a/include/opendht/net.h b/include/opendht/net.h deleted file mode 100644 index fd680e51..00000000 --- a/include/opendht/net.h +++ /dev/null @@ -1,67 +0,0 @@ -/* - * Copyright (C) 2014-2017 Savoir-faire Linux Inc. - * Author(s) : Adrien Béraud - * Simon Désaulniers - * - * This program is free software; you can redistribute it and/or modify - * it under the terms of the GNU General Public License as published by - * the Free Software Foundation; either version 3 of the License, or - * (at your option) any later version. - * - * This program is distributed in the hope that it will be useful, - * but WITHOUT ANY WARRANTY; without even the implied warranty of - * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the - * GNU General Public License for more details. - * - * You should have received a copy of the GNU General Public License - * along with this program. If not, see . - */ - -#pragma once - -#include -#include - -namespace dht { -namespace net { - -struct TransPrefix : public std::array { - TransPrefix(const std::string& str) : std::array({{(uint8_t)str[0], (uint8_t)str[1]}}) {} - static const TransPrefix PING; - static const TransPrefix FIND_NODE; - static const TransPrefix GET_VALUES; - static const TransPrefix ANNOUNCE_VALUES; - static const TransPrefix REFRESH; - static const TransPrefix LISTEN; -}; - -/* Transaction-ids are 4-bytes long, with the first two bytes identifying - * the kind of request, and the remaining two a sequence number in - * host order. - */ -struct TransId final : public std::array { - static const constexpr uint16_t INVALID {0}; - - TransId() { std::fill(begin(), end(), 0); } - TransId(const std::array& o) { std::copy(o.begin(), o.end(), begin()); } - TransId(const TransPrefix prefix, uint16_t seqno = 0) { - std::copy_n(prefix.begin(), prefix.size(), begin()); - *reinterpret_cast(data()+prefix.size()) = seqno; - } - TransId(uint32_t id) { - *reinterpret_cast(data()) = htonl(id); - } - - uint32_t toInt() const { - return ntohl(*reinterpret_cast(&(*this)[0])); - } - - bool matches(const TransPrefix prefix) const { - return std::equal(begin(), begin()+2, prefix.begin()); - } -}; - -TransId unpackTid(msgpack::object& o); - -} /* namespace net */ -} /* dht */ diff --git a/include/opendht/network_engine.h b/include/opendht/network_engine.h index 1fc5f020..ba9a06ef 100644 --- a/include/opendht/network_engine.h +++ b/include/opendht/network_engine.h @@ -27,7 +27,6 @@ #include "utils.h" #include "rng.h" #include "rate_limiter.h" -#include "net.h" #include #include @@ -41,6 +40,7 @@ namespace net { struct Request; struct Socket; +struct TransId; #ifndef MSG_CONFIRM #define MSG_CONFIRM 0 @@ -169,7 +169,7 @@ private: std::function, const InfoHash&, const Blob&, - uint32_t, + Tid, const Query&)> onListen {}; /** * Called on announce request. @@ -231,7 +231,7 @@ public: * @param nodes6 The ipv6 closest nodes. * @param values The values to send. */ - void tellListener(Sp n, uint32_t socket_id, const InfoHash& hash, want_t want, const Blob& ntoken, + void tellListener(Sp n, Tid socket_id, const InfoHash& hash, want_t want, const Blob& ntoken, std::vector>&& nodes, std::vector>&& nodes6, std::vector>&& values, const Query& q); @@ -465,18 +465,18 @@ private: // basic wrapper for socket sendto function int send(const char *buf, size_t len, int flags, const SockAddr& addr); - void sendValueParts(TransId tid, const std::vector& svals, const SockAddr& addr); + void sendValueParts(const TransId& tid, const std::vector& svals, const SockAddr& addr); std::vector packValueHeader(msgpack::sbuffer&, const std::vector>&); - void maintainRxBuffer(const TransId& tid); + void maintainRxBuffer(Tid tid); /************* * Answers * *************/ /* answer to a ping request */ - void sendPong(const SockAddr& addr, TransId tid); + void sendPong(const SockAddr& addr, Tid tid); /* answer to findnodes/getvalues request */ void sendNodesValues(const SockAddr& addr, - TransId tid, + Tid tid, const Blob& nodes, const Blob& nodes6, const std::vector>& st, @@ -490,12 +490,12 @@ private: std::vector>& nodes, std::vector>& nodes6); /* answer to a listen request */ - void sendListenConfirmation(const SockAddr& addr, TransId tid); + void sendListenConfirmation(const SockAddr& addr, Tid tid); /* answer to put request */ - void sendValueAnnounced(const SockAddr& addr, TransId, Value::Id); + void sendValueAnnounced(const SockAddr& addr, Tid, Value::Id); /* answer in case of error */ void sendError(const SockAddr& addr, - TransId tid, + Tid tid, uint16_t code, const std::string& message, bool include_id=false); @@ -519,8 +519,8 @@ private: size_t limiter_maintenance {0}; // requests handling - std::map> requests {}; - std::map partial_messages; + std::map> requests {}; + std::map partial_messages; MessageStats in_stats {}, out_stats {}; std::set blacklist {}; diff --git a/include/opendht/node.h b/include/opendht/node.h index 1f6efda9..394b1156 100644 --- a/include/opendht/node.h +++ b/include/opendht/node.h @@ -22,7 +22,6 @@ #include "infohash.h" // includes socket structures #include "utils.h" #include "sockaddr.h" -#include "net.h" #include #include @@ -36,8 +35,8 @@ struct Socket; struct RequestAnswer; } /* namespace net */ +using Tid = uint32_t; using SocketCb = std::function&, net::RequestAnswer&&)>; -using SocketId = uint32_t; struct Socket { Socket() {} Socket(SocketCb&& on_receive) : @@ -102,7 +101,7 @@ struct Node { void requested(const Sp& req); void received(time_point now, const Sp& req); - Sp getRequest(const net::TransId& tid); + Sp getRequest(Tid tid); void cancelRequest(const Sp& req); void setExpired(); @@ -116,16 +115,16 @@ struct Node { * * @return the socket. */ - SocketId openSocket(SocketCb&& cb); + Tid openSocket(SocketCb&& cb); - Socket* getSocket(SocketId id); + Socket* getSocket(Tid id); /** * Closes a socket so that no further data will be red on that socket. * * @param socket The socket to close. */ - void closeSocket(SocketId id); + void closeSocket(Tid id); /** * Resets the state of the node so it's not expired anymore. @@ -137,9 +136,9 @@ struct Node { * * @return the new id. */ - uint16_t getNewTid() { + Tid getNewTid() { ++transaction_id; - return transaction_id == net::TransId::INVALID ? ++transaction_id : transaction_id; + return transaction_id ? ++transaction_id : transaction_id; } std::string toString() const; @@ -164,11 +163,11 @@ private: time_point reply_time {time_point::min()}; /* time of last correct reply received */ unsigned auth_errors {0}; bool expired_ {false}; - SocketId transaction_id; + Tid transaction_id; using TransactionDist = std::uniform_int_distribution; - std::map> requests_ {}; - std::map sockets_; + std::map> requests_ {}; + std::map sockets_; }; } diff --git a/src/Makefile.am b/src/Makefile.am index 634f2d88..f4614795 100644 --- a/src/Makefile.am +++ b/src/Makefile.am @@ -9,6 +9,7 @@ libopendht_la_SOURCES = \ listener.h \ request.h \ search.h \ + net.h \ parsed_message.h \ node_cache.cpp \ callbacks.cpp \ @@ -37,7 +38,6 @@ nobase_include_HEADERS = \ ../include/opendht/node_cache.h \ ../include/opendht/routing_table.h \ ../include/opendht/network_engine.h \ - ../include/opendht/net.h \ ../include/opendht/scheduler.h \ ../include/opendht/rate_limiter.h \ ../include/opendht/utils.h \ diff --git a/src/net.h b/src/net.h new file mode 100644 index 00000000..b47eeb25 --- /dev/null +++ b/src/net.h @@ -0,0 +1,37 @@ +/* + * Copyright (C) 2017 Savoir-faire Linux Inc. + * Author(s) : Adrien Béraud + * + * This program is free software; you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation; either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program. If not, see . + */ + +#pragma once + +namespace dht { +namespace net { + +enum class MessageType { + Error = 0, + Reply, + Ping, + FindNode, + GetValues, + AnnounceValue, + Refresh, + Listen, + ValueData +}; + +} /* namespace net */ +} /* dht */ diff --git a/src/network_engine.cpp b/src/network_engine.cpp index 26836f14..d48e4d64 100644 --- a/src/network_engine.cpp +++ b/src/network_engine.cpp @@ -53,14 +53,6 @@ constexpr std::chrono::seconds NetworkEngine::RX_MAX_PACKET_TIME; constexpr std::chrono::seconds NetworkEngine::RX_TIMEOUT; const std::string NetworkEngine::my_v {"RNG1"}; -const constexpr uint16_t TransId::INVALID; - -const TransPrefix TransPrefix::PING = {"pn"}; -const TransPrefix TransPrefix::FIND_NODE = {"fn"}; -const TransPrefix TransPrefix::GET_VALUES = {"gt"}; -const TransPrefix TransPrefix::ANNOUNCE_VALUES = {"pt"}; -const TransPrefix TransPrefix::REFRESH = {"rf"}; -const TransPrefix TransPrefix::LISTEN = {"lt"}; constexpr size_t NetworkEngine::MAX_REQUESTS_PER_SEC; static const uint8_t v4prefix[16] = { @@ -95,6 +87,27 @@ set_nonblocking(int fd, int nonblocking) #endif + +/* Transaction-ids are 4-bytes long, with the first two bytes identifying + * the kind of request, and the remaining two a sequence number in + * host order. + */ +struct TransId final : public std::array { + static const constexpr uint16_t INVALID {0}; + + TransId() { std::fill(begin(), end(), 0); } + TransId(const std::array& o) { std::copy(o.begin(), o.end(), begin()); } + TransId(uint32_t id) { + *reinterpret_cast(data()) = htonl(id); + } + + uint32_t toInt() const { + return ntohl(*reinterpret_cast(&(*this)[0])); + } +}; + + + struct NetworkEngine::PartialMessage { SockAddr from; time_point start; @@ -146,14 +159,14 @@ NetworkEngine::~NetworkEngine() { } void -NetworkEngine::tellListener(Sp node, uint32_t socket_id, const InfoHash& hash, want_t want, +NetworkEngine::tellListener(Sp node, Tid socket_id, const InfoHash& hash, want_t want, const Blob& ntoken, std::vector>&& nodes, std::vector>&& nodes6, std::vector>&& values, const Query& query) { auto nnodes = bufferNodes(node->getFamily(), hash, want, nodes, nodes6); try { - sendNodesValues(node->getAddr(), TransId(socket_id), nnodes.first, nnodes.second, values, query, ntoken); + sendNodesValues(node->getAddr(), socket_id, nnodes.first, nnodes.second, values, query, ntoken); } catch (const std::overflow_error& e) { DHT_LOG.e("Can't send value: buffer not large enough !"); } @@ -401,7 +414,7 @@ NetworkEngine::process(std::unique_ptr&& msg, const SockAddr& fro auto node = cache.getNode(msg->id, from, now, true, msg->is_client); if (msg->type == MessageType::Error or msg->type == MessageType::Reply) { - auto rsocket = node->getSocket(msg->tid.toInt()); + auto rsocket = node->getSocket(msg->tid); auto req = node->getRequest(msg->tid); /* either response for a request or data for an opened socket */ @@ -432,10 +445,10 @@ NetworkEngine::process(std::unique_ptr&& msg, const SockAddr& fro switch (msg->type) { case MessageType::Error: { - if (msg->id != zeroes and ( - (msg->error_code == DhtProtocolException::NOT_FOUND and msg->tid.matches(TransPrefix::REFRESH)) or - (msg->error_code == DhtProtocolException::UNAUTHORIZED and (msg->tid.matches(TransPrefix::ANNOUNCE_VALUES) - or msg->tid.matches(TransPrefix::LISTEN))))) + if (msg->id != zeroes and req and ( + (msg->error_code == DhtProtocolException::NOT_FOUND and req->getType() == MessageType::Refresh) or + (msg->error_code == DhtProtocolException::UNAUTHORIZED and (req->getType() == MessageType::AnnounceValue + or req->getType() == MessageType::Listen)))) { req->last_try = time_point::min(); req->reply_time = time_point::min(); @@ -513,7 +526,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; - RequestAnswer answer = onListen(node, msg->info_hash, msg->token, msg->socket_id.toInt(), std::move(msg->query)); + RequestAnswer answer = onListen(node, msg->info_hash, msg->token, msg->socket_id, std::move(msg->query)); sendListenConfirmation(from, msg->tid); break; } @@ -573,7 +586,7 @@ NetworkEngine::send(const char *buf, size_t len, int flags, const SockAddr& addr Sp NetworkEngine::sendPing(Sp node, RequestCb&& on_done, RequestExpiredCb&& on_expired) { - auto tid = TransId {TransPrefix::PING, node->getNewTid()}; + TransId tid (node->getNewTid()); msgpack::sbuffer buffer; msgpack::packer pk(&buffer); pk.pack_map(5+(network?1:0)); @@ -590,7 +603,7 @@ NetworkEngine::sendPing(Sp node, RequestCb&& on_done, RequestExpiredCb&& o pk.pack(std::string("n")); pk.pack(network); } - auto req = std::make_shared(tid, node, + auto req = std::make_shared(MessageType::Ping, tid.toInt(), node, Blob(buffer.data(), buffer.data() + buffer.size()), [=](const Request& req_status, ParsedMessage&&) { DHT_LOG.d(req_status.node->id, "[node %s] got pong !", req_status.node->toString().c_str()); @@ -610,7 +623,7 @@ NetworkEngine::sendPing(Sp node, RequestCb&& on_done, RequestExpiredCb&& o } void -NetworkEngine::sendPong(const SockAddr& addr, TransId tid) { +NetworkEngine::sendPong(const SockAddr& addr, Tid tid) { msgpack::sbuffer buffer; msgpack::packer pk(&buffer); pk.pack_map(4+(network?1:0)); @@ -619,8 +632,9 @@ NetworkEngine::sendPong(const SockAddr& addr, TransId tid) { pk.pack(std::string("id")); pk.pack(myid); insertAddr(pk, addr); - pk.pack(std::string("t")); pk.pack_bin(tid.size()); - pk.pack_bin_body((const char*)tid.data(), tid.size()); + TransId t (tid); + pk.pack(std::string("t")); pk.pack_bin(t.size()); + pk.pack_bin_body((const char*)t.data(), t.size()); pk.pack(std::string("y")); pk.pack(std::string("r")); pk.pack(std::string("v")); pk.pack(my_v); if (network) { @@ -633,7 +647,7 @@ NetworkEngine::sendPong(const SockAddr& addr, TransId tid) { Sp NetworkEngine::sendFindNode(Sp n, const InfoHash& target, want_t want, RequestCb&& on_done, RequestExpiredCb&& on_expired) { - auto tid = TransId {TransPrefix::FIND_NODE, n->getNewTid()}; + TransId tid (n->getNewTid()); msgpack::sbuffer buffer; msgpack::packer pk(&buffer); pk.pack_map(5+(network?1:0)); @@ -657,7 +671,7 @@ NetworkEngine::sendFindNode(Sp n, const InfoHash& target, want_t want, pk.pack(std::string("n")); pk.pack(network); } - auto req = std::make_shared(tid, n, + auto req = std::make_shared(MessageType::FindNode, tid.toInt(), n, Blob(buffer.data(), buffer.data() + buffer.size()), [=](const Request& req_status, ParsedMessage&& msg) { /* on done */ if (on_done) { @@ -679,7 +693,7 @@ NetworkEngine::sendFindNode(Sp n, const InfoHash& target, want_t want, Sp NetworkEngine::sendGetValues(Sp n, const InfoHash& info_hash, const Query& query, want_t want, RequestCb&& on_done, RequestExpiredCb&& on_expired) { - auto tid = TransId {TransPrefix::GET_VALUES, n->getNewTid()}; + TransId tid (n->getNewTid()); msgpack::sbuffer buffer; msgpack::packer pk(&buffer); pk.pack_map(5+(network?1:0)); @@ -706,7 +720,7 @@ NetworkEngine::sendGetValues(Sp n, const InfoHash& info_hash, const Query& pk.pack(std::string("n")); pk.pack(network); } - auto req = std::make_shared(tid, n, + auto req = std::make_shared(MessageType::GetValues, tid.toInt(), n, Blob(buffer.data(), buffer.data() + buffer.size()), [=](const Request& req_status, ParsedMessage&& msg) { /* on done */ if (on_done) { @@ -807,7 +821,7 @@ NetworkEngine::packValueHeader(msgpack::sbuffer& buffer, const std::vector& svals, const SockAddr& addr) +NetworkEngine::sendValueParts(const TransId& tid, const std::vector& svals, const SockAddr& addr) { msgpack::sbuffer buffer; unsigned i=0; @@ -837,7 +851,7 @@ NetworkEngine::sendValueParts(TransId tid, const std::vector& svals, const } void -NetworkEngine::sendNodesValues(const SockAddr& addr, TransId tid, const Blob& nodes, const Blob& nodes6, +NetworkEngine::sendNodesValues(const SockAddr& addr, Tid tid, const Blob& nodes, const Blob& nodes6, const std::vector>& st, const Query& query, const Blob& token) { msgpack::sbuffer buffer; @@ -878,8 +892,9 @@ NetworkEngine::sendNodesValues(const SockAddr& addr, TransId tid, const Blob& no } } - pk.pack(std::string("t")); pk.pack_bin(tid.size()); - pk.pack_bin_body((const char*)tid.data(), tid.size()); + TransId t (tid); + pk.pack(std::string("t")); pk.pack_bin(t.size()); + pk.pack_bin_body((const char*)t.data(), t.size()); pk.pack(std::string("y")); pk.pack(std::string("r")); pk.pack(std::string("v")); pk.pack(my_v); if (network) { @@ -954,8 +969,8 @@ NetworkEngine::sendListen(Sp n, RequestExpiredCb&& on_expired, SocketCb&& socket_cb) { - uint32_t socket; - auto tid = TransId { TransPrefix::LISTEN, n->getNewTid() }; + Tid socket; + TransId tid (n->getNewTid()); if (previous and previous->node == n) { socket = previous->getSocket(); } else { @@ -994,7 +1009,7 @@ NetworkEngine::sendListen(Sp n, pk.pack(std::string("n")); pk.pack(network); } - auto req = std::make_shared(tid, n, + auto req = std::make_shared(MessageType::Listen, tid.toInt(), n, Blob(buffer.data(), buffer.data() + buffer.size()), [=](const Request& req_status, ParsedMessage&& msg) { /* on done */ if (on_done) @@ -1012,7 +1027,7 @@ NetworkEngine::sendListen(Sp n, } void -NetworkEngine::sendListenConfirmation(const SockAddr& addr, TransId tid) { +NetworkEngine::sendListenConfirmation(const SockAddr& addr, Tid tid) { msgpack::sbuffer buffer; msgpack::packer pk(&buffer); pk.pack_map(4+(network?1:0)); @@ -1021,8 +1036,9 @@ NetworkEngine::sendListenConfirmation(const SockAddr& addr, TransId tid) { pk.pack(std::string("id")); pk.pack(myid); insertAddr(pk, addr); - pk.pack(std::string("t")); pk.pack_bin(tid.size()); - pk.pack_bin_body((const char*)tid.data(), tid.size()); + TransId t (tid); + pk.pack(std::string("t")); pk.pack_bin(t.size()); + pk.pack_bin_body((const char*)t.data(), t.size()); pk.pack(std::string("y")); pk.pack(std::string("r")); pk.pack(std::string("v")); pk.pack(my_v); if (network) { @@ -1041,7 +1057,7 @@ NetworkEngine::sendAnnounceValue(Sp n, RequestCb&& on_done, RequestExpiredCb&& on_expired) { - auto tid = TransId {TransPrefix::ANNOUNCE_VALUES, n->getNewTid()}; + TransId tid (n->getNewTid()); msgpack::sbuffer buffer; msgpack::packer pk(&buffer); pk.pack_map(5+(network?1:0)); @@ -1065,7 +1081,7 @@ NetworkEngine::sendAnnounceValue(Sp n, pk.pack(std::string("n")); pk.pack(network); } - auto req = std::make_shared(tid, n, + auto req = std::make_shared(MessageType::AnnounceValue, tid.toInt(), n, Blob(buffer.data(), buffer.data() + buffer.size()), [=](const Request& req_status, ParsedMessage&& msg) { /* on done */ if (msg.value_id == Value::INVALID_ID) { @@ -1099,7 +1115,7 @@ NetworkEngine::sendRefreshValue(Sp n, RequestCb&& on_done, RequestExpiredCb&& on_expired) { - auto tid = TransId {TransPrefix::REFRESH, n->getNewTid()}; + TransId tid (n->getNewTid()); msgpack::sbuffer buffer; msgpack::packer pk(&buffer); pk.pack_map(5+(network?1:0)); @@ -1119,7 +1135,7 @@ NetworkEngine::sendRefreshValue(Sp n, pk.pack(std::string("n")); pk.pack(network); } - auto req = std::make_shared(tid, n, + auto req = std::make_shared(MessageType::Refresh, tid.toInt(), n, Blob(buffer.data(), buffer.data() + buffer.size()), [=](const Request& req_status, ParsedMessage&& msg) { /* on done */ if (msg.value_id == Value::INVALID_ID) { @@ -1144,7 +1160,7 @@ NetworkEngine::sendRefreshValue(Sp n, } void -NetworkEngine::sendValueAnnounced(const SockAddr& addr, TransId tid, Value::Id vid) { +NetworkEngine::sendValueAnnounced(const SockAddr& addr, Tid tid, Value::Id vid) { msgpack::sbuffer buffer; msgpack::packer pk(&buffer); pk.pack_map(4+(network?1:0)); @@ -1154,8 +1170,9 @@ NetworkEngine::sendValueAnnounced(const SockAddr& addr, TransId tid, Value::Id v pk.pack(std::string("vid")); pk.pack(vid); insertAddr(pk, addr); - pk.pack(std::string("t")); pk.pack_bin(tid.size()); - pk.pack_bin_body((const char*)tid.data(), tid.size()); + TransId t(tid); + pk.pack(std::string("t")); pk.pack_bin(t.size()); + pk.pack_bin_body((const char*)t.data(), t.size()); pk.pack(std::string("y")); pk.pack(std::string("r")); pk.pack(std::string("v")); pk.pack(my_v); if (network) { @@ -1167,7 +1184,7 @@ NetworkEngine::sendValueAnnounced(const SockAddr& addr, TransId tid, Value::Id v void NetworkEngine::sendError(const SockAddr& addr, - TransId tid, + Tid tid, uint16_t code, const std::string& message, bool include_id) @@ -1185,8 +1202,9 @@ NetworkEngine::sendError(const SockAddr& addr, pk.pack(std::string("id")); pk.pack(myid); } - pk.pack(std::string("t")); pk.pack_bin(tid.size()); - pk.pack_bin_body((const char*)tid.data(), tid.size()); + TransId t(tid); + pk.pack(std::string("t")); pk.pack_bin(t.size()); + pk.pack_bin_body((const char*)t.data(), t.size()); pk.pack(std::string("y")); pk.pack(std::string("e")); pk.pack(std::string("v")); pk.pack(my_v); if (network) { @@ -1197,7 +1215,7 @@ NetworkEngine::sendError(const SockAddr& addr, } void -NetworkEngine::maintainRxBuffer(const TransId& tid) +NetworkEngine::maintainRxBuffer(Tid tid) { auto msg = partial_messages.find(tid); if (msg != partial_messages.end()) { diff --git a/src/node.cpp b/src/node.cpp index ba8f4cb9..8bd17a24 100644 --- a/src/node.cpp +++ b/src/node.cpp @@ -35,7 +35,7 @@ Node::Node(const InfoHash& id, const SockAddr& addr, bool client) : id(id), addr(addr), is_client(client), sockets_() { crypto::random_device rd; - transaction_id = std::uniform_int_distribution{1}(rd); + transaction_id = std::uniform_int_distribution{1}(rd); } /* This is our definition of a known-good node. */ @@ -101,7 +101,7 @@ Node::received(time_point now, const Sp& req) } Sp -Node::getRequest(const net::TransId& tid) +Node::getRequest(Tid tid) { auto it = requests_.find(tid); return it != requests_.end() ? it->second : nullptr; @@ -128,7 +128,7 @@ Node::setExpired() sockets_.clear(); } -SocketId +Tid Node::openSocket(SocketCb&& cb) { if (++transaction_id == 0) @@ -142,14 +142,14 @@ Node::openSocket(SocketCb&& cb) } Socket* -Node::getSocket(SocketId id) +Node::getSocket(Tid id) { auto it = sockets_.find(id); return it == sockets_.end() ? nullptr : &it->second; } void -Node::closeSocket(SocketId id) +Node::closeSocket(Tid id) { if (id) { sockets_.erase(id); diff --git a/src/parsed_message.h b/src/parsed_message.h index 1164342e..983ffefa 100644 --- a/src/parsed_message.h +++ b/src/parsed_message.h @@ -26,17 +26,14 @@ namespace dht { namespace net { -enum class MessageType { - Error = 0, - Reply, - Ping, - FindNode, - GetValues, - AnnounceValue, - Refresh, - Listen, - ValueData -}; +Tid unpackTid(const msgpack::object& o) { + switch (o.type) { + case msgpack::type::POSITIVE_INTEGER: + return o.as(); + default: + return ntohl(*reinterpret_cast(o.as>().data())); + } +} struct ParsedMessage { MessageType type; @@ -51,9 +48,9 @@ struct ParsedMessage { /* target id around which to find nodes */ InfoHash target; /* transaction id */ - TransId tid; + Tid tid; /* tid for packets going through request socket */ - TransId socket_id; + Tid socket_id; /* security token */ Blob token; /* the value id (announce confirmation) */ diff --git a/src/request.h b/src/request.h index 0b222eba..3c04cd34 100644 --- a/src/request.h +++ b/src/request.h @@ -59,18 +59,19 @@ struct Request { State getState() const { return state_; } Request(State state = State::PENDING) : state_(state) {} - Request(const TransId& tid, + Request(MessageType type, Tid tid, Sp node, Blob&& msg, std::function on_done, std::function on_expired, - uint32_t socket = 0) : - node(node), on_done(on_done), on_expired(on_expired), tid(tid), msg(std::move(msg)), socket(socket) { } + Tid socket = 0) : + node(node), tid(tid), type(type), on_done(on_done), on_expired(on_expired), msg(std::move(msg)), socket(socket) { } - TransId getTid() const { return tid; } + Tid getTid() const { return tid; } + MessageType getType() const { return type; } - uint32_t getSocket() { return socket; } - uint32_t closeSocket() { auto ret = socket; socket = 0; return ret; } + Tid getSocket() { return socket; } + Tid closeSocket() { auto ret = socket; socket = 0; return ret; } void setExpired() { if (pending()) { @@ -107,6 +108,8 @@ private: msg.clear(); } + const Tid tid {0}; /* the request id. */ + const MessageType type {}; State state_ {State::PENDING}; unsigned attempt_count {0}; /* number of attempt to process the request. */ @@ -116,9 +119,8 @@ private: std::function on_done {}; std::function on_expired {}; - const TransId tid; /* the request id. */ Blob msg {}; /* the serialized message. */ - uint32_t socket; /* the socket used for further reponses. */ + Tid socket; /* the socket used for further reponses. */ }; } /* namespace net */ diff --git a/src/utils.cpp b/src/utils.cpp index cb2fd5b3..87826e1f 100644 --- a/src/utils.cpp +++ b/src/utils.cpp @@ -19,7 +19,6 @@ #include "utils.h" #include "sockaddr.h" #include "default_types.h" -#include "net.h" /* An IPv4 equivalent to IN6_IS_ADDR_UNSPECIFIED */ #ifndef IN_IS_ADDR_UNSPECIFIED @@ -161,18 +160,4 @@ findMapValue(msgpack::object& map, const std::string& key) { return nullptr; } -namespace net { - -TransId -unpackTid(msgpack::object& o) { - switch (o.type) { - case msgpack::type::POSITIVE_INTEGER: - return o.as(); - default: - return o.as>(); - } -} - -} - }