// Copyright (c) 2014-2026 Savoir-faire Linux Inc. // SPDX-License-Identifier: MIT #include "node.h" #include "request.h" #include "rng.h" #include namespace dht { constexpr std::chrono::minutes Node::NODE_EXPIRE_TIME; constexpr std::chrono::minutes Node::NODE_GOOD_TIME; constexpr std::chrono::seconds Node::MAX_RESPONSE_TIME; Node::Node(const InfoHash& id, const SockAddr& addr, std::mt19937_64& rd, bool client) : id(id) , addr(addr) , is_client(client) , sockets_() { transaction_id = std::uniform_int_distribution {1}(rd); } Node::Node(const InfoHash& id, SockAddr&& addr, std::mt19937_64& rd, bool client) : id(id) , addr(std::move(addr)) , is_client(client) , sockets_() { transaction_id = std::uniform_int_distribution {1}(rd); } /* This is our definition of a known-good node. */ bool Node::isGood(time_point now) const { return not expired_ && reply_time >= now - NODE_GOOD_TIME && time >= now - NODE_EXPIRE_TIME; } bool Node::isPendingMessage() const { for (const auto& r : requests_) { if (r.second->pending()) return true; } return false; } size_t Node::getPendingMessageCount() const { size_t count {0}; for (const auto& r : requests_) { if (r.second->pending()) count++; } return count; } void Node::update(const SockAddr& new_addr) { addr = new_addr; } /** To be called when a message was sent to the node */ void Node::requested(const Sp& req) { auto e = requests_.emplace(req->getTid(), req); if (not e.second and req != e.first->second) { // Should not happen ! // Try to handle this scenario as well as we can e.first->second->setExpired(); e.first->second = req; } } /** To be called when a message was received from the node. Req should be true if the message was an answer to a request we made*/ void Node::received(time_point now, const Sp& req) { time = now; expired_ = false; if (req) { reply_time = now; requests_.erase(req->getTid()); } } Sp Node::getRequest(Tid tid) { auto it = requests_.find(tid); return it != requests_.end() ? it->second : nullptr; } void Node::cancelRequest(const Sp& req) { if (req) { req->cancel(); requests_.erase(req->getTid()); } } void Node::setExpired() { expired_ = true; for (auto r : requests_) { r.second->setExpired(); } requests_.clear(); sockets_.clear(); } Tid Node::openSocket(SocketCb&& cb) { if (++transaction_id == 0) transaction_id = 1; sockets_[transaction_id] = std::make_shared(std::move(cb)); return transaction_id; } Sp Node::getSocket(Tid id) { auto it = sockets_.find(id); return it == sockets_.end() ? nullptr : it->second; } void Node::closeSocket(Tid id) { if (id) { sockets_.erase(id); // DHT_LOG.w("Closing socket (tid: %d), %lu remaining", socket->id, sockets_.size()); } } std::string Node::toString() const { return fmt::format("{} {:<21}", id, addr.toString()); } std::ostream& operator<<(std::ostream& s, const Node& h) { s << h.id << " " << h.addr.toString(); return s; } void NodeExport::msgpack_unpack(msgpack::object o) { if (o.type != msgpack::type::MAP) throw msgpack::type_error(); if (o.via.map.size < 2) throw msgpack::type_error(); if (o.via.map.ptr[0].key.as() != "id"sv) throw msgpack::type_error(); if (o.via.map.ptr[1].key.as() != "addr"sv) throw msgpack::type_error(); const auto& maddr = o.via.map.ptr[1].val; if (maddr.type != msgpack::type::BIN) throw msgpack::type_error(); if (maddr.via.bin.size > sizeof(sockaddr_storage)) throw msgpack::type_error(); id.msgpack_unpack(o.via.map.ptr[0].val); addr = {(const sockaddr*) maddr.via.bin.ptr, (socklen_t) maddr.via.bin.size}; } std::ostream& operator<<(std::ostream& s, const NodeExport& h) { msgpack::pack(s, h); return s; } } // namespace dht