From 289ea964a231c7526874bbbea20380ca01028a2b Mon Sep 17 00:00:00 2001 From: Adrien Beraud Date: Sun, 2 Aug 2026 11:32:47 -0400 Subject: [PATCH] peer_discovery: cache DNS-SD responses --- src/dns_sd.cpp | 159 +++++++++++++++++++++++++++++++++++++++++ src/dns_sd.h | 24 +++++++ src/peer_discovery.cpp | 26 +++---- 3 files changed, 191 insertions(+), 18 deletions(-) diff --git a/src/dns_sd.cpp b/src/dns_sd.cpp index 5c4bab05..d65ceb78 100644 --- a/src/dns_sd.cpp +++ b/src/dns_sd.cpp @@ -170,6 +170,165 @@ isKnownAnswer(const Message& query, const ResourceRecord& record) } // namespace +class Cache::Impl +{ +public: + explicit Impl(Now now) + : now_(std::move(now)) + {} + + struct Entry + { + ResourceRecord record; + Clock::time_point expires; + }; + + std::vector update(const Message& message) + { + expire(); + std::vector records; + forEachRecord(message, [&](const ResourceRecord& record) { records.push_back(&record); }); + + std::vector flushed; + for (const auto* record : records) { + if (not record->cacheFlush) + continue; + const auto alreadyFlushed = std::any_of(flushed.begin(), flushed.end(), [&](const ResourceRecord* other) { + return other->name == record->name + and other->type == record->type and other->classCode == record->classCode; + }); + if (not alreadyFlushed) { + flushed.push_back(record); + entries_.erase( + std::remove_if(entries_.begin(), entries_.end(), [&](const Entry& entry) { + return entry.record.name == record->name and entry.record.type == record->type + and entry.record.classCode == record->classCode; + }), + entries_.end()); + } + } + + for (const auto* record : records) { + if (record->ttl == 0) { + entries_.erase( + std::remove_if(entries_.begin(), entries_.end(), [&](const Entry& entry) { + return sameRecord(entry.record, *record); + }), + entries_.end()); + continue; + } + const auto expires = now_() + std::chrono::seconds(record->ttl); + const auto existing = std::find_if(entries_.begin(), entries_.end(), [&](const Entry& entry) { + return sameRecord(entry.record, *record); + }); + if (existing == entries_.end()) + entries_.push_back({*record, expires}); + else { + existing->record = *record; + existing->expires = expires; + } + } + + auto current = collect(); + std::vector discovered; + for (const auto& service : current) { + if (std::find_if(emitted_.begin(), emitted_.end(), [&](const Service& previous) { + return equal(previous, service); + }) == emitted_.end()) + discovered.push_back(service); + } + emitted_ = std::move(current); + return discovered; + } + + std::vector services() + { + expire(); + auto current = collect(); + emitted_ = current; + return current; + } + + void clear() + { + entries_.clear(); + emitted_.clear(); + } + +private: + static bool equal(const Service& lhs, const Service& rhs) + { + return lhs.nodeId == rhs.nodeId and lhs.network == rhs.network and lhs.port == rhs.port + and lhs.addresses == rhs.addresses; + } + + void expire() + { + const auto now = now_(); + entries_.erase(std::remove_if(entries_.begin(), entries_.end(), [&](const Entry& entry) { + return entry.expires <= now; + }), + entries_.end()); + } + + std::vector collect() const + { + std::vector result; + for (const auto& entry : entries_) { + if (entry.record.type != Type::PTR or entry.record.classCode != CLASS_IN + or entry.record.name != serviceName()) + continue; + const auto* ptr = std::get_if(&entry.record.data); + if (not ptr) + continue; + + Message message; + message.response = true; + message.answers.push_back(entry.record); + for (const auto& additional : entries_) { + if (&additional != &entry and additional.record.type != Type::PTR) + message.additionals.push_back(additional.record); + } + const auto service = resolve(message); + if (service and std::none_of(result.begin(), result.end(), [&](const Service& existing) { + return equal(existing, *service); + })) + result.push_back(*service); + } + return result; + } + + Now now_; + std::vector entries_; + std::vector emitted_; +}; + +Cache::Cache(Now now) + : impl_(std::make_unique(std::move(now))) +{} + +Cache::~Cache() = default; +Cache::Cache(Cache&&) noexcept = default; +Cache& Cache::operator=(Cache&&) noexcept = default; + +std::vector +Cache::update(const Message& message) +{ + return impl_->update(message); +} + +std::vector +Cache::services() +{ + return impl_->services(); +} + +void +Cache::clear() +{ + impl_->clear(); +} + Name serviceName() { diff --git a/src/dns_sd.h b/src/dns_sd.h index f209fa39..a6e54468 100644 --- a/src/dns_sd.h +++ b/src/dns_sd.h @@ -7,6 +7,9 @@ #include "opendht/infohash.h" #include "opendht/utils.h" +#include +#include +#include #include namespace dht::mdns::dns_sd { @@ -24,6 +27,27 @@ struct Service std::vector addresses; }; +class OPENDHT_PUBLIC Cache +{ +public: + using Clock = std::chrono::steady_clock; + using Now = std::function; + + explicit Cache(Now now = [] { return Clock::now(); }); + ~Cache(); + + Cache(Cache&&) noexcept; + Cache& operator=(Cache&&) noexcept; + + std::vector update(const Message& message); + std::vector services(); + void clear(); + +private: + class Impl; + std::unique_ptr impl_; +}; + OPENDHT_PUBLIC Name serviceName(); OPENDHT_PUBLIC Name instanceName(const InfoHash& nodeId); OPENDHT_PUBLIC Name hostName(const InfoHash& nodeId); diff --git a/src/peer_discovery.cpp b/src/peer_discovery.cpp index 705d7082..2026bbf7 100644 --- a/src/peer_discovery.cpp +++ b/src/peer_discovery.cpp @@ -45,14 +45,6 @@ socketAddress(const mdns::dns_sd::AddressData& data, return address; } -bool -isGoodbye(const mdns::Message& message) -{ - return std::any_of(message.answers.begin(), message.answers.end(), [](const mdns::ResourceRecord& record) { - return record.type == mdns::Type::PTR and record.name == mdns::dns_sd::serviceName() and record.ttl == 0; - }); -} - } // namespace class PeerDiscovery::Impl : public std::enable_shared_from_this @@ -303,12 +295,6 @@ private: void discover(const mdns::Message& message, const asio::ip::udp::endpoint& source) { - if (isGoodbye(message)) - return; - const auto service = mdns::dns_sd::resolve(message); - if (not service) - return; - PeerDiscoveredCallback callback; std::optional network; std::optional localNode; @@ -319,12 +305,15 @@ private: if (published_) localNode = published_->nodeId; } - if (not callback or not network or service->network != *network - or (localNode and service->nodeId == *localNode)) + if (not callback or not network) return; - for (const auto& address : service->addresses) - callback(service->nodeId, socketAddress(address, service->port, source)); + for (const auto& service : cache_.update(message)) { + if (service.network != *network or (localNode and service.nodeId == *localNode)) + continue; + for (const auto& address : service.addresses) + callback(service.nodeId, socketAddress(address, service.port, source)); + } } std::shared_ptr ioContext_; @@ -338,6 +327,7 @@ private: std::optional discoveryNetwork_; PeerDiscoveredCallback discoveryCallback_; std::optional published_; + mdns::dns_sd::Cache cache_; }; PeerDiscovery::PeerDiscovery(std::shared_ptr ioContext, std::shared_ptr logger)