From e19671a168e9c563f637abec2612bbef6673eb69 Mon Sep 17 00:00:00 2001 From: Saurab Dulal Date: Sun, 5 Apr 2026 08:05:16 -0700 Subject: [PATCH 1/2] face: add Adaptive Multicast Suppression (AMS) - Add multicast-suppression.hpp/.cpp implementing AMS with EMA-based duplicate detection, per-prefix NameTree suppression timers, and AIMD-style adaptive delay (multiplicative increase / additive decrease) - Integrate suppression into LinkService send/receive paths; gated behind m_suppressionEnabled so stock NFD behavior is preserved by default - Add enableMulticastSuppression knob to GenericLinkServiceOptions, propagated from nfd.conf via UDP and Ethernet factory config parsing (mcast_suppression yes/no, default no) - Fix critical bugs: explicit Name copy in cancelIfSchdeuled, switch all scheduler::EventId storage to ScopedEventId for safe auto-cancellation on face/object teardown - Add 26 unit tests covering NameTree, EMAMeasurements, and MulticastSuppression; add status and code-review docs Co-Authored-By: Claude Sonnet 4.6 --- daemon/face/ethernet-factory.cpp | 4 + daemon/face/ethernet-factory.hpp | 1 + daemon/face/generic-link-service.cpp | 2 + daemon/face/generic-link-service.hpp | 4 + daemon/face/link-service.cpp | 323 ++++++++---- daemon/face/link-service.hpp | 493 ++++++++++-------- daemon/face/multicast-suppression.cpp | 378 ++++++++++++++ daemon/face/multicast-suppression.hpp | 195 +++++++ daemon/face/udp-factory.cpp | 4 + daemon/face/udp-factory.hpp | 1 + docs/multicast-suppression-review.md | 302 +++++++++++ docs/multicast-suppression-status.md | 119 +++++ nfd.conf.sample.in | 2 + tests/daemon/face/multicast-suppression.t.cpp | 316 +++++++++++ 14 files changed, 1815 insertions(+), 329 deletions(-) create mode 100644 daemon/face/multicast-suppression.cpp create mode 100644 daemon/face/multicast-suppression.hpp create mode 100644 docs/multicast-suppression-review.md create mode 100644 docs/multicast-suppression-status.md create mode 100644 tests/daemon/face/multicast-suppression.t.cpp diff --git a/daemon/face/ethernet-factory.cpp b/daemon/face/ethernet-factory.cpp index 175a7d52..f362f89d 100644 --- a/daemon/face/ethernet-factory.cpp +++ b/daemon/face/ethernet-factory.cpp @@ -107,6 +107,9 @@ EthernetFactory::doProcessConfig(OptionalConfigSection configSection, bool wantAdHoc = ConfigFile::parseYesNo(pair, "face_system.ether"); mcastConfig.linkType = wantAdHoc ? ndn::nfd::LINK_TYPE_AD_HOC : ndn::nfd::LINK_TYPE_MULTI_ACCESS; } + else if (key == "mcast_suppression") { + mcastConfig.enableMulticastSuppression = ConfigFile::parseYesNo(pair, "face_system.ether"); + } else if (key == "whitelist") { mcastConfig.netifPredicate.parseWhitelist(value); } @@ -248,6 +251,7 @@ EthernetFactory::createMulticastFace(const ndn::net::NetworkInterface& netif, GenericLinkService::Options opts; opts.allowFragmentation = true; opts.allowReassembly = true; + opts.enableMulticastSuppression = m_mcastConfig.enableMulticastSuppression; auto linkService = make_unique(opts); auto transport = make_unique(netif, address, m_mcastConfig.linkType); diff --git a/daemon/face/ethernet-factory.hpp b/daemon/face/ethernet-factory.hpp index d8d49c94..b6cc33ee 100644 --- a/daemon/face/ethernet-factory.hpp +++ b/daemon/face/ethernet-factory.hpp @@ -121,6 +121,7 @@ class EthernetFactory final : public ProtocolFactory ethernet::Address group = ethernet::getDefaultMulticastAddress(); ndn::nfd::LinkType linkType = ndn::nfd::LINK_TYPE_MULTI_ACCESS; NetworkInterfacePredicate netifPredicate; + bool enableMulticastSuppression = false; }; MulticastConfig m_mcastConfig; diff --git a/daemon/face/generic-link-service.cpp b/daemon/face/generic-link-service.cpp index de073b47..8f1c2ba1 100644 --- a/daemon/face/generic-link-service.cpp +++ b/daemon/face/generic-link-service.cpp @@ -48,6 +48,7 @@ GenericLinkService::GenericLinkService(const GenericLinkService::Options& option m_reassembler.beforeTimeout.connect([this] (auto&&...) { ++nReassemblyTimeouts; }); m_reliability.onDroppedInterest.connect([this] (const auto& i) { notifyDroppedInterest(i); }); nReassembling.observe(&m_reassembler); + setMulticastSuppression(m_options.enableMulticastSuppression); } void @@ -57,6 +58,7 @@ GenericLinkService::setOptions(const GenericLinkService::Options& options) m_fragmenter.setOptions(m_options.fragmenterOptions); m_reassembler.setOptions(m_options.reassemblerOptions); m_reliability.setOptions(m_options.reliabilityOptions); + setMulticastSuppression(m_options.enableMulticastSuppression); } ssize_t diff --git a/daemon/face/generic-link-service.hpp b/daemon/face/generic-link-service.hpp index 1d321e2d..4307ce1a 100644 --- a/daemon/face/generic-link-service.hpp +++ b/daemon/face/generic-link-service.hpp @@ -142,6 +142,10 @@ struct GenericLinkServiceOptions */ bool allowSelfLearning = true; + /** \brief Enables adaptive multicast suppression (AMS) on multicast faces. + */ + bool enableMulticastSuppression = false; + /** \brief Overrides the MTU provided by Transport. * * This MTU value will be used instead of the MTU provided by the transport if it is less than diff --git a/daemon/face/link-service.cpp b/daemon/face/link-service.cpp index 7d1445e2..97a55533 100644 --- a/daemon/face/link-service.cpp +++ b/daemon/face/link-service.cpp @@ -1,6 +1,6 @@ /* -*- Mode:C++; c-file-style:"gnu"; indent-tabs-mode:nil; -*- */ /* - * Copyright (c) 2014-2022, Regents of the University of California, + * Copyright (c) 2014-2020, Regents of the University of California, * Arizona Board of Regents, * Colorado State University, * University Pierre & Marie Curie, Sorbonne University, @@ -23,107 +23,220 @@ * NFD, e.g., in COPYING.md file. If not, see . */ -#include "link-service.hpp" -#include "face.hpp" - -namespace nfd::face { - -NFD_LOG_INIT(LinkService); - -LinkService::~LinkService() = default; - -void -LinkService::setFaceAndTransport(Face& face, Transport& transport) noexcept -{ - BOOST_ASSERT(m_face == nullptr); - BOOST_ASSERT(m_transport == nullptr); - - m_face = &face; - m_transport = &transport; -} - -void -LinkService::sendInterest(const Interest& interest) -{ - BOOST_ASSERT(m_transport != nullptr); - NFD_LOG_FACE_TRACE(__func__); - - ++this->nOutInterests; - - doSendInterest(interest); -} - -void -LinkService::sendData(const Data& data) -{ - BOOST_ASSERT(m_transport != nullptr); - NFD_LOG_FACE_TRACE(__func__); - - ++this->nOutData; - - doSendData(data); -} - -void -LinkService::sendNack(const ndn::lp::Nack& nack) -{ - BOOST_ASSERT(m_transport != nullptr); - NFD_LOG_FACE_TRACE(__func__); - - ++this->nOutNacks; - - doSendNack(nack); -} - -void -LinkService::receiveInterest(const Interest& interest, const EndpointId& endpoint) -{ - NFD_LOG_FACE_TRACE(__func__); - - ++this->nInInterests; - - afterReceiveInterest(interest, endpoint); -} - -void -LinkService::receiveData(const Data& data, const EndpointId& endpoint) -{ - NFD_LOG_FACE_TRACE(__func__); - - ++this->nInData; - - afterReceiveData(data, endpoint); -} - -void -LinkService::receiveNack(const ndn::lp::Nack& nack, const EndpointId& endpoint) -{ - NFD_LOG_FACE_TRACE(__func__); - - ++this->nInNacks; - - afterReceiveNack(nack, endpoint); -} - -void -LinkService::notifyDroppedInterest(const Interest& interest) -{ - ++this->nInterestsExceededRetx; - onDroppedInterest(interest); -} - -std::ostream& -operator<<(std::ostream& os, const FaceLogHelper& flh) -{ - const Face* face = flh.obj.getFace(); - if (face == nullptr) { - os << "[id=0,local=unknown,remote=unknown] "; - } - else { - os << "[id=" << face->getId() << ",local=" << face->getLocalUri() - << ",remote=" << face->getRemoteUri() << "] "; - } - return os; -} - -} // namespace nfd::face + #include "link-service.hpp" + #include "face.hpp" + + namespace nfd { + namespace face { + + NFD_LOG_INIT(LinkService); + + LinkService::LinkService() + : m_face(nullptr) + , m_transport(nullptr) + { + } + + LinkService::~LinkService() + { + } + + void + LinkService::setFaceAndTransport(Face& face, Transport& transport) + { + BOOST_ASSERT(m_face == nullptr); + BOOST_ASSERT(m_transport == nullptr); + + m_face = &face; + m_transport = &transport; + } + + void + LinkService::sendInterest(const Interest& interest) + { + BOOST_ASSERT(m_transport != nullptr); + NFD_LOG_FACE_TRACE(__func__); + + if (this->getFace()->getLinkType() != ndn::nfd::LINK_TYPE_MULTI_ACCESS || !m_suppressionEnabled) + { + ++this->nOutInterests; + doSendInterest(interest); + return; + } + // apply suppression algorithm if sending through multicast face + // check if the interest is already in flight + if (m_multicastSuppression.interestInflight(interest)) { + NFD_LOG_INFO ("Interest drop by suppression, with name " << interest.getName() << " is in flight, drop the forwarding"); + return; // need to catch this, what should be the behaviour after dropping the interest?? + } + // wait for suppression time before forwarding + // check if another interest is overheard during the wait, if heard, cancle the forwarding + auto suppressionTime = m_multicastSuppression.getDelayTimer(interest.getName(), 'i'); + NFD_LOG_INFO ("Interest " << interest.getName() << " not in flight, waiting" << suppressionTime << "before forwarding"); + + auto entry_name = interest.getName(); + entry_name.appendNumber(0); + auto eventId = getScheduler().schedule(suppressionTime, [this, interest, entry_name] { + NFD_LOG_INFO ("Interest " << interest.getName() << " Analysis History Interest sent finally: "); + int result = m_multicastSuppression.recordInterest(interest, true); + if (result == 0) { + NFD_LOG_INFO("Interest drop by suppression, Interest " << interest.getName() << " is in measurement table, drop the forwarding"); + } + else{ + ++this->nOutInterests; + doSendInterest(interest); + } + + if (m_scheduledEntry.count(entry_name) > 0) + m_scheduledEntry.erase(entry_name); + }); + m_scheduledEntry.emplace(entry_name, std::move(eventId)); + } + + void + LinkService::sendData(const Data& data) + { + BOOST_ASSERT(m_transport != nullptr); + NFD_LOG_FACE_TRACE(__func__); + if (this->getFace()->getLinkType() != ndn::nfd::LINK_TYPE_MULTI_ACCESS || !m_suppressionEnabled) + { + ++this->nOutData; + doSendData(data); + return; + } + if (m_multicastSuppression.dataInflight(data)) { + NFD_LOG_INFO("Data drop by suppression, Data " << data.getName() << " is in measurement table, drop the forwarding"); + return; // need to catch this, what should be the behaviour after dropping the interest?? + } + /* + Same suppression logic as that of interest cannot be applied to multicast data. + Doing so we might end up suppressing data for the interest received at different interval + from different node, additionally it will also trigger multiple retransmission from the node + that didn't received the data due to suppression. This might not happen if unsolicated data are + cached, but the node that's supposed to send the data can't gurantee it. + + data sending should wait before forwarding, during this wait time if another data is overheard, need to drop the forwarding + wait time should be determined based on the number of duplicate overhearing + */ + // for now, lets wait for some time before forwarding, if overheard, drop the reply + auto suppressionTime = m_multicastSuppression.getDelayTimer(data.getName(), 'd'); + NFD_LOG_INFO("Waiting : " << suppressionTime<< "ms before sending data" << data.getName()); + + auto entry_name = data.getName(); + entry_name.appendNumber(1); + auto eventId = getScheduler().schedule(suppressionTime, [this, data, entry_name] { + + NFD_LOG_INFO("Sending data finally, via multicast face Analysis History Data sent finally:" << data.getName()); + int result = m_multicastSuppression.recordData(data, true); + if (result == 0) { + NFD_LOG_INFO("Data drop by suppression, Data " << data.getName() << " is in measurement table, drop the forwarding"); + } + else{ + ++this->nOutData; + doSendData(data); + } + if(m_scheduledEntry.count(entry_name) > 0) + m_scheduledEntry.erase(entry_name); + + }); + m_scheduledEntry.emplace(entry_name, std::move(eventId)); + } + + + void + LinkService::sendNack(const ndn::lp::Nack& nack) + { + BOOST_ASSERT(m_transport != nullptr); + NFD_LOG_FACE_TRACE(__func__); + + ++this->nOutNacks; + + doSendNack(nack); + } + + bool + LinkService::cancelIfSchdeuled(Name name, int type) + { + Name entry_name = name; + entry_name.appendNumber(type); + auto it = m_scheduledEntry.find(entry_name); + if (it != m_scheduledEntry.end()) { + it->second.cancel(); + m_scheduledEntry.erase(entry_name); + return true; + } + return false; + } + + void + LinkService::receiveInterest(const Interest& interest, const EndpointId& endpoint) + { + NFD_LOG_FACE_TRACE(__func__); + // record multicast interest + if (this->getFace()->getLinkType() == ndn::nfd::LINK_TYPE_MULTI_ACCESS && m_suppressionEnabled) + { + NFD_LOG_INFO("Multicast interest received: " << interest.getName()); + // check if a same interest is scheduled, if so drop it + if (cancelIfSchdeuled(interest.getName(), 0)) + NDN_LOG_INFO("Interest drop by suppression, with name" << interest.getName() << " overheard, duplicate forwarding dropped"); + m_multicastSuppression.recordInterest(interest, false); + } + ++this->nInInterests; + afterReceiveInterest(interest, endpoint); + } + + void + LinkService::receiveData(const Data& data, const EndpointId& endpoint) + { + NFD_LOG_FACE_TRACE(__func__); + // record multicast Data received + if (this->getFace()->getLinkType() == ndn::nfd::LINK_TYPE_MULTI_ACCESS && m_suppressionEnabled) + { + NFD_LOG_INFO("Multicast data received: " << data.getName()); + if (cancelIfSchdeuled(data.getName(), 1)) + NDN_LOG_INFO("Data drop by suppression, with name " << data.getName() << " overheard, duplicate forwarding dropped"); + + if (cancelIfSchdeuled(data.getName(), 0)) // also can drop interest if shceduled for this data + NDN_LOG_INFO("Interest drop by suppression, with name " << data.getName() << " overheard, drop the corresponding scheduled interest"); + + m_multicastSuppression.recordData(data, false); + } + + ++this->nInData; + // record multicast data + afterReceiveData(data, endpoint); + } + + void + LinkService::receiveNack(const ndn::lp::Nack& nack, const EndpointId& endpoint) + { + NFD_LOG_FACE_TRACE(__func__); + + ++this->nInNacks; + + afterReceiveNack(nack, endpoint); + } + + void + LinkService::notifyDroppedInterest(const Interest& interest) + { + ++this->nInterestsExceededRetx; + onDroppedInterest(interest); + } + + std::ostream& + operator<<(std::ostream& os, const FaceLogHelper& flh) + { + const Face* face = flh.obj.getFace(); + if (face == nullptr) { + os << "[id=0,local=unknown,remote=unknown] "; + } + else { + os << "[id=" << face->getId() << ",local=" << face->getLocalUri() + << ",remote=" << face->getRemoteUri() << "] "; + } + return os; + } + + } // namespace face + } // namespace nfd \ No newline at end of file diff --git a/daemon/face/link-service.hpp b/daemon/face/link-service.hpp index 66dab05f..8b996fb9 100644 --- a/daemon/face/link-service.hpp +++ b/daemon/face/link-service.hpp @@ -1,6 +1,6 @@ /* -*- Mode:C++; c-file-style:"gnu"; indent-tabs-mode:nil; -*- */ /* - * Copyright (c) 2014-2024, Regents of the University of California, + * Copyright (c) 2014-2020, Regents of the University of California, * Arizona Board of Regents, * Colorado State University, * University Pierre & Marie Curie, Sorbonne University, @@ -23,226 +23,271 @@ * NFD, e.g., in COPYING.md file. If not, see . */ -#ifndef NFD_DAEMON_FACE_LINK_SERVICE_HPP -#define NFD_DAEMON_FACE_LINK_SERVICE_HPP - -#include "face-common.hpp" -#include "transport.hpp" -#include "common/counter.hpp" - -namespace nfd::face { - -/** - * \brief Counters provided by LinkService. - * \note The type name LinkServiceCounters is an implementation detail. - * Use LinkService::Counters in public API. - */ -class LinkServiceCounters -{ -public: - /// Count of incoming Interest packets. - PacketCounter nInInterests; - /// Count of outgoing Interest packets. - PacketCounter nOutInterests; - /// Count of Interests dropped by reliability system for exceeding allowed number of retx. - PacketCounter nInterestsExceededRetx; - /// Count of incoming Data packets. - PacketCounter nInData; - /// Count of outgoing Data packets. - PacketCounter nOutData; - /// Count of incoming Nack packets. - PacketCounter nInNacks; - /// Count of outgoing Nack packets. - PacketCounter nOutNacks; -}; - -/** - * \brief The upper half of a Face. - * \sa Face, Transport - */ -class LinkService : protected virtual LinkServiceCounters, noncopyable -{ -public: - /** - * \brief %Counters provided by LinkService. - */ - using Counters = LinkServiceCounters; - -public: - virtual - ~LinkService(); - - /** - * \brief Set Face and Transport for this LinkService. - * \pre setFaceAndTransport() has not been called. - */ - void - setFaceAndTransport(Face& face, Transport& transport) noexcept; - - /** - * \brief Returns the Face to which this LinkService is attached. - */ - const Face* - getFace() const noexcept - { - return m_face; - } - - /** - * \brief Returns the Transport to which this LinkService is attached. - */ - const Transport* - getTransport() const noexcept - { - return m_transport; - } - - /** - * \brief Returns the Transport to which this LinkService is attached. - */ - Transport* - getTransport() noexcept - { - return m_transport; - } - - virtual const Counters& - getCounters() const - { - return *this; - } - - virtual ssize_t - getEffectiveMtu() const - { - return m_transport->getMtu(); - } - -public: // upper interface to be used by forwarding - /** - * \brief Send Interest. - * \pre setFaceAndTransport() has been called. - */ - void - sendInterest(const Interest& interest); - - /** - * \brief Send Data. - * \pre setFaceAndTransport() has been called. - */ - void - sendData(const Data& data); - - /** - * \brief Send Nack. - * \pre setFaceAndTransport() has been called. - */ - void - sendNack(const ndn::lp::Nack& nack); - - /** - * \brief Called when an Interest packet is received. - */ - signal::Signal afterReceiveInterest; - - /** - * \brief Called when a Data packet is received. - */ - signal::Signal afterReceiveData; - - /** - * \brief Called when a Nack packet is received. - */ - signal::Signal afterReceiveNack; - - /** - * \brief Called when an Interest is dropped by the reliability system - * for exceeding the allowed number of retransmissions. - */ - signal::Signal onDroppedInterest; - -public: // lower interface to be invoked by Transport - /** - * \brief Performs LinkService-specific operations to receive a lower-layer packet. - */ - void - receivePacket(const Block& packet, const EndpointId& endpoint) - { - doReceivePacket(packet, endpoint); - } - -protected: // upper interface to be invoked in subclass (receive path termination) - /** - * \brief Delivers received Interest to forwarding. - */ - void - receiveInterest(const Interest& interest, const EndpointId& endpoint); - - /** - * \brief Delivers received Data to forwarding. - */ - void - receiveData(const Data& data, const EndpointId& endpoint); - - /** - * \brief Delivers received Nack to forwarding. - */ - void - receiveNack(const lp::Nack& nack, const EndpointId& endpoint); - -protected: // lower interface to be invoked in subclass (send path termination) - /** - * \brief Send a lower-layer packet via Transport. - */ - void - sendPacket(const Block& packet) - { - m_transport->send(packet); - } - -protected: - void - notifyDroppedInterest(const Interest& packet); - -private: // upper interface to be overridden in subclass (send path entrypoint) - /** - * \brief Performs LinkService-specific operations to send an Interest. - */ - virtual void - doSendInterest(const Interest& interest) = 0; - - /** - * \brief Performs LinkService-specific operations to send a Data. - */ - virtual void - doSendData(const Data& data) = 0; - - /** - * \brief Performs LinkService-specific operations to send a Nack. - */ - virtual void - doSendNack(const lp::Nack& nack) = 0; - -private: // lower interface to be overridden in subclass - virtual void - doReceivePacket(const Block& packet, const EndpointId& endpoint) = 0; - -private: - Face* m_face = nullptr; - Transport* m_transport = nullptr; -}; - -std::ostream& -operator<<(std::ostream& os, const FaceLogHelper& flh); - -template -std::enable_if_t && !std::is_same_v, - std::ostream&> -operator<<(std::ostream& os, const FaceLogHelper& flh) -{ - return os << FaceLogHelper(flh.obj); -} - -} // namespace nfd::face - -#endif // NFD_DAEMON_FACE_LINK_SERVICE_HPP + #ifndef NFD_DAEMON_FACE_LINK_SERVICE_HPP + #define NFD_DAEMON_FACE_LINK_SERVICE_HPP + + #include "face-common.hpp" + #include "transport.hpp" + #include "common/counter.hpp" + #include + #include "multicast-suppression.hpp" + + #include "common/global.hpp" + // #include "common/logger.hpp" + + namespace nfd { + namespace face { + + /** \brief counters provided by LinkService + * \note The type name 'LinkServiceCounters' is implementation detail. + * Use 'LinkService::Counters' in public API. + */ + class LinkServiceCounters + { + public: + /** \brief count of incoming Interests + */ + PacketCounter nInInterests; + + /** \brief count of outgoing Interests + */ + PacketCounter nOutInterests; + + /** \brief count of Interests dropped by reliability system for exceeding allowed number of retx + */ + PacketCounter nInterestsExceededRetx; + + /** \brief count of incoming Data packets + */ + PacketCounter nInData; + + /** \brief count of outgoing Data packets + */ + PacketCounter nOutData; + + /** \brief count of incoming Nacks + */ + PacketCounter nInNacks; + + /** \brief count of outgoing Nacks + */ + PacketCounter nOutNacks; + }; + + /** \brief the upper part of a Face + * \sa Face + */ + class LinkService : protected virtual LinkServiceCounters, noncopyable + { + public: + /** \brief counters provided by LinkService + */ + typedef LinkServiceCounters Counters; + + public: + LinkService(); + + virtual + ~LinkService(); + + /** \brief set Face and Transport for LinkService + * \pre setFaceAndTransport has not been called + */ + void + setFaceAndTransport(Face& face, Transport& transport); + + /** \return Face to which this LinkService is attached + */ + const Face* + getFace() const; + + /** \return Transport to which this LinkService is attached + */ + const Transport* + getTransport() const; + + /** \return Transport to which this LinkService is attached + */ + Transport* + getTransport(); + + virtual const Counters& + getCounters() const; + + virtual ssize_t + getEffectiveMtu() const; + + public: // upper interface to be used by forwarding + /** \brief Send Interest + * \pre setTransport has been called + */ + void + sendInterest(const Interest& interest); + + /** \brief Send Data + * \pre setTransport has been called + */ + void + sendData(const Data& data); + + /** \brief Send Nack + * \pre setTransport has been called + */ + void + sendNack(const ndn::lp::Nack& nack); + + /** \brief signals on Interest received + */ + signal::Signal afterReceiveInterest; + + /** \brief signals on Data received + */ + signal::Signal afterReceiveData; + + /** \brief signals on Nack received + */ + signal::Signal afterReceiveNack; + + /** \brief signals on Interest dropped by reliability system for exceeding allowed number of retx + */ + signal::Signal onDroppedInterest; + + public: // lower interface to be invoked by Transport + /** \brief performs LinkService specific operations to receive a lower-layer packet + */ + void + receivePacket(const Block& packet, const EndpointId& endpoint); + + protected: // upper interface to be invoked in subclass (receive path termination) + /** \brief delivers received Interest to forwarding + */ + void + receiveInterest(const Interest& interest, const EndpointId& endpoint); + + /** \brief delivers received Data to forwarding + */ + void + receiveData(const Data& data, const EndpointId& endpoint); + + /** \brief delivers received Nack to forwarding + */ + void + receiveNack(const lp::Nack& nack, const EndpointId& endpoint); + + void + scheduleEntry(Name name, scheduler::EventId& eid) + { + m_scheduledEntry.emplace(name, eid); + } + + bool + cancelIfSchdeuled(Name name, int type); + + protected: // lower interface to be invoked in subclass (send path termination) + /** \brief send a lower-layer packet via Transport + */ + void + sendPacket(const Block& packet); + + protected: + void + notifyDroppedInterest(const Interest& packet); + + private: // upper interface to be overridden in subclass (send path entrypoint) + /** \brief performs LinkService specific operations to send an Interest + */ + virtual void + doSendInterest(const Interest& interest) = 0; + + /** \brief performs LinkService specific operations to send a Data + */ + virtual void + doSendData(const Data& data) = 0; + + /** \brief performs LinkService specific operations to send a Nack + */ + virtual void + doSendNack(const lp::Nack& nack) = 0; + + private: // lower interface to be overridden in subclass + virtual void + doReceivePacket(const Block& packet, const EndpointId& endpoint) = 0; + + public: + void + setMulticastSuppression(bool enabled) + { + m_suppressionEnabled = enabled; + } + + bool + isMulticastSuppressionEnabled() const + { + return m_suppressionEnabled; + } + + private: + Face* m_face; + Transport* m_transport; + bool m_suppressionEnabled = false; + nfd::face::ams::MulticastSuppression m_multicastSuppression; + std::map m_scheduledEntry; + }; + + inline const Face* + LinkService::getFace() const + { + return m_face; + } + + inline const Transport* + LinkService::getTransport() const + { + return m_transport; + } + + inline Transport* + LinkService::getTransport() + { + return m_transport; + } + + inline const LinkService::Counters& + LinkService::getCounters() const + { + return *this; + } + + inline ssize_t + LinkService::getEffectiveMtu() const + { + return m_transport->getMtu(); + } + + inline void + LinkService::receivePacket(const Block& packet, const EndpointId& endpoint) + { + doReceivePacket(packet, endpoint); + } + + inline void + LinkService::sendPacket(const Block& packet) + { + m_transport->send(packet); + } + + std::ostream& + operator<<(std::ostream& os, const FaceLogHelper& flh); + + template + typename std::enable_if::value && + !std::is_same::value, std::ostream&>::type + operator<<(std::ostream& os, const FaceLogHelper& flh) + { + return os << FaceLogHelper(flh.obj); + } + + } // namespace face + } // namespace nfd + + #endif // NFD_DAEMON_FACE_LINK_SERVICE_HPP \ No newline at end of file diff --git a/daemon/face/multicast-suppression.cpp b/daemon/face/multicast-suppression.cpp new file mode 100644 index 00000000..fe1bef59 --- /dev/null +++ b/daemon/face/multicast-suppression.cpp @@ -0,0 +1,378 @@ +#include "multicast-suppression.hpp" +#include +#include +#include +#include "common/global.hpp" +#include "common/logger.hpp" +#include +#include +#include + +namespace nfd { +namespace face { +namespace ams { + +NFD_LOG_INIT(MulticastSuppression); + +const double DISCOUNT_FACTOR = 0.125; //a in paper +const double MAX_PROPOGATION_DELAY = 15; +const time::milliseconds MAX_MEASURMENT_INACTIVE_PERIOD = 300_s; // 5 minutes + +/* This is 2*MAX_PROPOGATION_DELAY. Basically, when a nodes (C1) forwards a packet, it will take 15ms to reach +its neighbors (C2). The packet will be recevied by the neighbors and they will suppress their forwarding. In case, +if the neighbor didnt received the packet from C1 in 15 ms, it will forward its own packet. Now, the actual duplicate count +in the network is 2, both nodes C1 & C2 should record dc = 2. For this to happen, it takes about 15ms for the packet from C2 to +reach C1. Thus, DEFAULT_INSTANT_LIFETIME = 30ms*/ + +const time::milliseconds DEFAULT_INSTANT_LIFETIME = 30_ms; +const double DUPLICATE_THRESHOLD = 1.3; // parameter to tune +const double ADATIVE_DECREASE = 5 ; +const double MULTIPLICATIVE_INCREASE = 1.3; + +// in milliseconds ms +const double minSuppressionTime = 15.0f; // probably we need to provide sufficient time for other party to hear you?? 5ms wont be sufficient?? +const double maxSuppressionTime= 15000.0f; +unsigned int UNSET = -1234; +int CHARACTER_SIZE = 126; +int MAX_IGNORE = 3; + +int +getRandomNumber(int upperBound) +{ + return ndn::random::generateWord32() % upperBound; +} + +NameTree::NameTree() +: isLeaf(false) +, suppressionTime(UNSET) +{ +} + +std::vector +NameTree::parseNameComponents(const std::string& name) +{ + std::vector components; + if (name.empty() || name[0] != '/') + return components; + + size_t start = 1; // skip leading '/' + size_t pos = name.find('/', start); + + while (pos != std::string::npos) { + components.push_back(name.substr(start, pos - start)); + start = pos + 1; + pos = name.find('/', start); + } + + // last component + if (start < name.length()) { + components.push_back(name.substr(start)); + } + + return components; +} + +void +NameTree::insert(const std::string& prefix, double value) +{ + auto components = parseNameComponents(prefix); + auto node = this; + + for (size_t i = 0; i < components.size(); i++) { + const auto& component = components[i]; + if (node->children.find(component) == node->children.end()) { + node->children[component] = std::make_unique(); + } + + node = node->children[component].get(); + + if (i == components.size() - 1) { + node->suppressionTime = value; + node->isLeaf = true; + } + } +} + +double +NameTree::longestPrefixMatch(const std::string& prefix) +{ + auto components = parseNameComponents(prefix); + auto node = this; + double lastValueFound = UNSET; + + for (size_t i = 0; i < components.size(); i++) { + const auto& component = components[i]; + auto it = node->children.find(component); + if (it == node->children.end()) { + break; + } + + node = it->second.get(); + + if (node->suppressionTime != UNSET) { + lastValueFound = node->suppressionTime; + } + } + + return lastValueFound; +} + +time::milliseconds +NameTree::getSuppressionTimer(const std::string& name) +{ + double val, suppressionTime; + val = longestPrefixMatch(name); + suppressionTime = (val == UNSET) ? minSuppressionTime : val; + time::milliseconds suppressionTimer (getRandomNumber(static_cast (2*suppressionTime))); // timer is randomized value + NFD_LOG_INFO("Suppression time: " << suppressionTime << " Suppression timer: " << suppressionTimer); + return suppressionTimer; +} + +/* objectName granularity is (-1) name component + m_lastForwardStaus is set to true if this node has successfully forwarded an interest or data + else is set to false. + start with 15ms, MAX propagation time, for a node to hear other node +*/ +EMAMeasurements::EMAMeasurements(double expMovingAverage, int lastDuplicateCount, double suppressionTime) +: m_expMovingAveragePrev (expMovingAverage) +, m_expMovingAverageCurrent (expMovingAverage) +, m_currentSuppressionTime(suppressionTime) +, m_lastDuplicateCount(1) +, m_maxDuplicateCount(0) +, m_minSuppressionTime(minSuppressionTime) +, m_ignoreDuplicateRecoring(0) +{ +} + +/* + we compute exponential moving average to give higher preference to the most recent interest/data + EMA = duplicate count if t = 1 + EMA = alpha*Dt + (1 - alpha) * EMA t-1 +*/ + +void +EMAMeasurements::addUpdateEMA(int duplicateCount, bool wasForwarded) +{ + NFD_LOG_INFO("addUPdateEma for " << wasForwarded); + // If duplicate count is greater than last duplicate count, increase the ignore counter + // else reset it to 0. + m_ignoreDuplicateRecoring = (duplicateCount > m_lastDuplicateCount) ? (m_ignoreDuplicateRecoring+1) : 0; + + if (m_ignoreDuplicateRecoring > 0 && m_ignoreDuplicateRecoring < MAX_IGNORE) { + NDN_LOG_INFO("Duplicate count: " << duplicateCount << " m_lastdup: " + << m_lastDuplicateCount << " ignore counter: " << m_ignoreDuplicateRecoring); + return; + } + + // Update/Reset duplicate count and ignore counter + m_lastDuplicateCount = duplicateCount; + m_ignoreDuplicateRecoring = 0; + + m_expMovingAveragePrev = m_expMovingAverageCurrent; + if (m_expMovingAverageCurrent == 0) { + m_expMovingAverageCurrent = duplicateCount; + } + else { + // rounding to 2 decimal place + m_expMovingAverageCurrent = round ((DISCOUNT_FACTOR*duplicateCount + + (1 - DISCOUNT_FACTOR)*m_expMovingAverageCurrent)*10.0)/10.0; + } + // Update maximum duplicate count + if (m_maxDuplicateCount < duplicateCount) { + m_maxDuplicateCount = duplicateCount; + } + // Update min suppression time + if (m_maxDuplicateCount > 1) { + m_minSuppressionTime = (float) MAX_PROPOGATION_DELAY; + } else if (m_maxDuplicateCount == 1 && m_minSuppressionTime > 1) { + m_minSuppressionTime--; + } + + // Update the suppression time, only if this node has forwarded + updateDelayTime(wasForwarded); + + // Log the results + NFD_LOG_INFO("Moving average" << " before: " << m_expMovingAveragePrev + << " after: " << m_expMovingAverageCurrent + << " duplicate count: " << duplicateCount + << " suppression time: "<< m_currentSuppressionTime); +} +void +EMAMeasurements::updateDelayTime(bool wasForwarded) +{ + NFD_LOG_INFO("Update delay timer called " << wasForwarded); + double temp; + // Implicit action: if you haven’t reached the goal, but your moving average is decreasing then do nothing. + if (m_expMovingAverageCurrent > DUPLICATE_THRESHOLD && + m_expMovingAverageCurrent >= m_expMovingAveragePrev ) { + // only increase the suppression timer if this node as forwarded + temp = (wasForwarded) ? (m_currentSuppressionTime * MULTIPLICATIVE_INCREASE) : m_currentSuppressionTime; + + } + else if (m_expMovingAverageCurrent <= DUPLICATE_THRESHOLD && + m_expMovingAverageCurrent <= m_expMovingAveragePrev) { + temp = m_currentSuppressionTime - ADATIVE_DECREASE; + } + else { + temp = m_currentSuppressionTime; + } + m_currentSuppressionTime = std::min(std::max(m_minSuppressionTime, temp), maxSuppressionTime); + NFD_LOG_INFO("Suppression time updated with " << m_currentSuppressionTime << " forwarded status " << wasForwarded); +} + + +int +MulticastSuppression::recordInterest(const Interest& interest, bool isForwarded) +{ + auto name = interest.getName(); + NFD_LOG_INFO("Interest to check/record" << name); + auto it = m_interestHistory.find(name); + if (it == m_interestHistory.end()) // check if interest is already in the map + { + auto forwardStatus = isForwarded ? true : getForwardedStatus(name, 'i'); + m_interestHistory.emplace(name, ObjectHistory{1, forwardStatus}); + NFD_LOG_INFO ("Interest: " << name << " inserted into map"); + + // remove the entry after the lifetime expries + time::milliseconds entryLifetime = DEFAULT_INSTANT_LIFETIME; + NFD_LOG_INFO("Erasing the interest from the map in : " << entryLifetime); + setUpdateExpiration(entryLifetime, name, 'i'); + return 1; + } + else { + NFD_LOG_INFO("Counter for interest " << name << " incremented"); + ++it->second.counter; + } + return 0; +} + +int +MulticastSuppression::recordData(const Data& data, bool isForwarded) +{ + auto name = data.getName(); //.getPrefix(-1); //removing nounce + NFD_LOG_INFO("Data to check/record " << name); + auto it = m_dataHistory.find(name); + if (it == m_dataHistory.end()) + { + NFD_LOG_INFO("Inserting data " << name << " into the map"); + auto forwardStatus = isForwarded ? true : getForwardedStatus(name, 'd'); + m_dataHistory.emplace(name, ObjectHistory{1, forwardStatus}); + + time::milliseconds entryLifetime = DEFAULT_INSTANT_LIFETIME; + NFD_LOG_INFO("Erasing the data from the map in : " << entryLifetime); + setUpdateExpiration(entryLifetime, name, 'd'); + return 1; + } + else + { + NFD_LOG_INFO("Counter for data " << name << " incremented"); + ++it->second.counter; + } + // need to check if we have the interest in the map + // if present, need to remove it from the map + ndn::Name name_cop = name; + name_cop.appendNumber(0); + auto itr_timer = m_objectExpirationTimer.find(name_cop); + if (itr_timer != m_objectExpirationTimer.end()) + { + NFD_LOG_INFO("Data overheard, deleting interest " <second.cancel(); + // schedule deletion now + if (m_interestHistory.count(name) > 0) + { + updateMeasurement(name, 'i'); + m_interestHistory.erase(name); + NFD_LOG_INFO("Interest successfully deleted from the history " <count(name) > 0) + { + // record interest into moving average + updateMeasurement(name, type); + vec->erase(name); + NFD_LOG_INFO("Name: " << name << " type: " << type << " expired, and deleted from the instant history"); + } + }); + + name = (type == 'i') ? name.appendNumber(0) : name.appendNumber(1); + auto itr_timer = m_objectExpirationTimer.find(name); + if (itr_timer != m_objectExpirationTimer.end()) + { + NFD_LOG_INFO("Updating timer for name: " << name << "type: " << type); + itr_timer->second.cancel(); + itr_timer->second = std::move(eventId); + } + else + { + m_objectExpirationTimer.emplace(name, std::move(eventId)); + } +} + +void +MulticastSuppression::updateMeasurement(Name name, char type) +{ + // if the measurment expires, can't the name stay with EMA = 0? so that we dont have to recreate it again later + auto vec = getEMARecorder(type); + auto nameTree = getNameTree(type); + auto duplicateCount = getDuplicateCount(name, type); + bool wasForwarded = getForwardedStatus(name, type); + + NDN_LOG_INFO("Update Measurement for " << type <<" Name: " << name << " Duplicate Count: " << duplicateCount << " type: " << type); + // granularity = name - last component e.g. /a/b --> /a + name = name.getPrefix(-1); + auto it = vec->find(name); + + // no records + if (it == vec->end()) + { + NFD_LOG_INFO("Creating EMA record for name: " << name << " type: " << type); + auto expirationId = getScheduler().schedule(MAX_MEASURMENT_INACTIVE_PERIOD, [=] { + if (vec->count(name) > 0) + vec->erase(name); + // dont delete the entry in the nametree, just unset the value + nameTree->insert(name.toUri(), UNSET); + }); + auto& emaEntry = vec->emplace(name, std::make_shared()).first->second; + emaEntry->setEMAExpiration(std::move(expirationId)); + emaEntry->addUpdateEMA(duplicateCount, wasForwarded); + nameTree->insert(name.toUri(), emaEntry->getCurrentSuppressionTime()); + } + // update existing record + else + { + NFD_LOG_INFO("Updating EMA record for name: " << name << " type: " << type); + it->second->getEMAExpiration().cancel(); + auto expirationId = getScheduler().schedule(MAX_MEASURMENT_INACTIVE_PERIOD, [=] { + if (vec->count(name) > 0) + vec->erase(name); + // set the value in the nametree = -1 + nameTree->insert(name.toUri(), UNSET); + }); + + it->second->setEMAExpiration(std::move(expirationId)); + it->second->addUpdateEMA(duplicateCount, wasForwarded); + nameTree->insert(name.toUri(), it->second->getCurrentSuppressionTime()); + } +} + +time::milliseconds +MulticastSuppression::getDelayTimer(Name name, char type) +{ + NFD_LOG_INFO("Getting supperssion timer for name: " << name); + auto nameTree = getNameTree(type); + auto suppressionTimer = nameTree->getSuppressionTimer(name.getPrefix(-1).toUri()); + NFD_LOG_INFO("Suppression timer for name: " << name << " and type: "<< type << " = " << suppressionTimer); + return suppressionTimer; +} + +} //namespace ams +} //namespace face +} //namespace nfd \ No newline at end of file diff --git a/daemon/face/multicast-suppression.hpp b/daemon/face/multicast-suppression.hpp new file mode 100644 index 00000000..fcb1a094 --- /dev/null +++ b/daemon/face/multicast-suppression.hpp @@ -0,0 +1,195 @@ +#include "core/common.hpp" +#include +#include +#include + +#ifndef NFD_DAEMON_FACE_AMS_MULTICAST_SUPPRESSION_HPP +#define NFD_DAEMON_FACE_AMS_MULTICAST_SUPPRESSION_HPP + +namespace nfd { +namespace face { +namespace scheduler = ndn::scheduler; +namespace ams { + +/* Component-based trie for NDN name prefix match */ +class NameTree +{ +public: + std::map> children; + bool isLeaf; + double suppressionTime; + + NameTree(); + + void + insert(const std::string& prefix, double value); + + double + longestPrefixMatch(const std::string& prefix); + + time::milliseconds + getSuppressionTimer(const std::string& prefix); + +private: + std::vector + parseNameComponents(const std::string& name); +}; + +class EMAMeasurements +{ + +public: + EMAMeasurements(double expMovingAverage = 0, int lastDuplicateCount = 0, double suppressionTime = 1); + + void + addUpdateEMA(int duplicateCount, bool wasForwarded); + + scheduler::ScopedEventId& + getEMAExpiration() + { + return this->m_expirationId; + } + + void + setEMAExpiration(scheduler::EventId expirationId) + { + this->m_expirationId = std::move(expirationId); + } + + float + getEMACurrent() + { + return this->m_expMovingAverageCurrent; + } + + float + getEMAPrev() + { + return this->m_expMovingAveragePrev; + } + + void + updateDelayTime(bool wasForwarded); + + double + getCurrentSuppressionTime() + { + return m_currentSuppressionTime; + } + + void + setSSthress(double val, int factor = 2) + { + m_ssthress = val/factor; + } + +private: +double m_expMovingAveragePrev; +double m_expMovingAverageCurrent; +double m_currentSuppressionTime; +scheduler::ScopedEventId m_expirationId; +int m_lastDuplicateCount; +int m_maxDuplicateCount; +double m_minSuppressionTime; +double m_ssthress; +int m_ignoreDuplicateRecoring; +}; + + +class MulticastSuppression +{ +public: + + struct ObjectHistory + { + int counter; + bool isForwarded; + }; + + int + recordInterest(const Interest& interest, bool isForwarded = false); + + int + recordData(const Data& data, bool isForwarded = false); + + int + getDuplicateCount(const Name name, char type) + { + auto temp_map = getRecorder(type); + auto it = temp_map->find(name); + if (it != temp_map->end()) + return it->second.counter; + return 0; + } + + std::map* + getRecorder(char type) + { + return (type == 'i') ? &m_interestHistory : &m_dataHistory; + } + + std::map>* + getEMARecorder(char type) + { + return (type =='i') ? &m_EMA_interest : &m_EMA_data; + } + + NameTree* + getNameTree(char type) + { + return (type =='i') ? &m_interestNameTree : &m_dataNameTree; + } + + bool + interestInflight(const Interest& interest) const + { + auto name = interest.getName(); + return (m_interestHistory.find(name) != m_interestHistory.end() ); + } + + bool + dataInflight(const Data& data) const + { + auto name = data.getName(); + return (m_dataHistory.find(name) != m_dataHistory.end()); + } + +time::milliseconds +getRandomTime() + { + return time::milliseconds(1 + (ndn::random::generateWord32() % 10)); + } + +void +updateMeasurement(Name name, char type); + +// set interest or data expiration +void +setUpdateExpiration(time::milliseconds entryLifetime, Name name, char type); + +time::milliseconds +getDelayTimer(Name name, char type); + +bool +getForwardedStatus(ndn::Name prefix, char type) +{ + auto recorder = getRecorder(type); + auto it = recorder->find(prefix); + return it != recorder->end() ? it->second.isForwarded : false; // if record exist, send whatever is the status else send false +} + +private: + + std::map m_dataHistory; + std::map m_interestHistory; + std::map m_objectExpirationTimer; + std::map> m_EMA_data; + std::map> m_EMA_interest; + NameTree m_dataNameTree; + NameTree m_interestNameTree; +}; +} //namespace ams +} //namespace face +} //namespace nfd + +#endif // NFD_DAEMON_FACE_SUPPRESSION_STRATEGY_HPP \ No newline at end of file diff --git a/daemon/face/udp-factory.cpp b/daemon/face/udp-factory.cpp index 52813031..e0b6c900 100644 --- a/daemon/face/udp-factory.cpp +++ b/daemon/face/udp-factory.cpp @@ -163,6 +163,9 @@ UdpFactory::doProcessConfig(OptionalConfigSection configSection, bool wantAdHoc = ConfigFile::parseYesNo(pair, "face_system.udp"); mcastConfig.linkType = wantAdHoc ? ndn::nfd::LINK_TYPE_AD_HOC : ndn::nfd::LINK_TYPE_MULTI_ACCESS; } + else if (key == "mcast_suppression") { + mcastConfig.enableMulticastSuppression = ConfigFile::parseYesNo(pair, "face_system.udp"); + } else if (key == "whitelist") { mcastConfig.netifPredicate.parseWhitelist(value); } @@ -365,6 +368,7 @@ UdpFactory::createMulticastFace(const net::NetworkInterface& netif, GenericLinkService::Options options; options.allowCongestionMarking = m_wantCongestionMarking; + options.enableMulticastSuppression = m_mcastConfig.enableMulticastSuppression; auto linkService = make_unique(options); auto transport = make_unique(mcastEp, std::move(rxSock), std::move(txSock), m_mcastConfig.linkType); diff --git a/daemon/face/udp-factory.hpp b/daemon/face/udp-factory.hpp index ca697e79..61d1556c 100644 --- a/daemon/face/udp-factory.hpp +++ b/daemon/face/udp-factory.hpp @@ -134,6 +134,7 @@ class UdpFactory final : public ProtocolFactory udp::Endpoint groupV6 = udp::getDefaultMulticastGroupV6(); ndn::nfd::LinkType linkType = ndn::nfd::LINK_TYPE_MULTI_ACCESS; NetworkInterfacePredicate netifPredicate; + bool enableMulticastSuppression = false; }; MulticastConfig m_mcastConfig; std::map> m_mcastFaces; diff --git a/docs/multicast-suppression-review.md b/docs/multicast-suppression-review.md new file mode 100644 index 00000000..188b8ab6 --- /dev/null +++ b/docs/multicast-suppression-review.md @@ -0,0 +1,302 @@ +# Adaptive Multicast Suppression (AMS) — Code Review + +> Reviewed: 2026-04-05 +> Files: `multicast-suppression.hpp/.cpp`, `link-service.hpp/.cpp`, `multicast-suppression.t.cpp` + +--- + +## Critical Bugs + +### C1 — `cancelIfSchdeuled` mutates the caller's Name +**File:** `link-service.cpp:159` +**Status:** [x] Fixed + +```cpp +// WRONG — appendNumber returns *this, entry_name is just an alias to the mutated name +auto entry_name = name.appendNumber(type); + +// CORRECT +Name entry_name = name; +entry_name.appendNumber(type); +``` + +`appendNumber()` modifies the Name in-place and returns a reference to `*this`. Every call to +`cancelIfSchdeuled` permanently corrupts the name passed in. This breaks any subsequent lookup +using that name. + +> **Fix applied:** Changed to `Name entry_name = name; entry_name.appendNumber(type);` — explicit +> copy makes type `Name` (not `Name&`) and the intent unambiguous. + +--- + +### C2 — Lambda captures `this` with no lifetime guarantee +**File:** `link-service.cpp:79, 127` | `multicast-suppression.cpp:295, 337, 353` +**Status:** [x] Fixed + +The lambdas scheduled for delayed forwarding capture `this` raw. If the face is torn down before +the timer fires, accessing `this->nOutInterests` or calling `doSendInterest` is a use-after-free. + +Same problem in `multicast-suppression.cpp`: `vec` and `nameTree` are raw pointers to member +variables captured inside scheduler lambdas. If `MulticastSuppression` is destroyed before the +timer fires, they dangle. + +**Fix:** Use `shared_from_this()` or ensure all scheduled events are cancelled in the destructor +before the object is destroyed. + +> **Fix applied:** Changed `m_scheduledEntry` to `std::map`, +> `m_expirationId` in `EMAMeasurements` to `scheduler::ScopedEventId`, and +> `m_objectExpirationTimer` to `std::map`. All EventIds are now +> stored via `std::move`. `ScopedEventId` auto-cancels on destruction, so all pending timers are +> cancelled when the face or suppression object is torn down. + +--- + +## High Severity + +### H1 — `UNSET = -1234` assigned to `unsigned int` +**File:** `multicast-suppression.cpp:35` +**Status:** [ ] Open + +```cpp +unsigned int UNSET = -1234; // wraps to 4294966062 on 32-bit +``` + +This giant value propagates into `longestPrefixMatch` return, then into `2 * suppressionTime` +cast to `int` — causing overflow and undefined behavior. + +**Fix:** Use `const double UNSET = -1.0;` or `std::optional` to represent "no value". + +--- + +### H2 — Global variables are not `const` +**File:** `multicast-suppression.cpp:35–37` +**Status:** [ ] Open + +```cpp +unsigned int UNSET = -1234; // should be constexpr +int CHARACTER_SIZE = 126; // unused and mutable +int MAX_IGNORE = 3; // mutable global +``` + +Mutable globals cause linkage issues if the file is ever included from multiple translation units, +and allow accidental modification. + +**Fix:** `static constexpr` for all three. Remove `CHARACTER_SIZE` if it is unused. + +--- + +### H3 — EMA constructor ignores its `lastDuplicateCount` parameter +**File:** `multicast-suppression.cpp:140` +**Status:** [ ] Open + +```cpp +EMAMeasurements::EMAMeasurements(double expMovingAverage, int lastDuplicateCount, ...) + : m_lastDuplicateCount(1) // always hardcoded — parameter silently discarded +``` + +The `lastDuplicateCount` argument passed by callers (e.g. tests) has no effect. + +**Fix:** Change to `: m_lastDuplicateCount(lastDuplicateCount)` or remove the parameter. + +--- + +### H4 — `getRandomNumber(0)` is undefined behavior +**File:** `multicast-suppression.cpp:42` +**Status:** [ ] Open + +```cpp +return ndn::random::generateWord32() % upperBound; // UB if upperBound == 0 +``` + +When `suppressionTime` rounds to 0, `2 * suppressionTime = 0`, and `% 0` is undefined behavior +in C++. No bounds check exists anywhere in the call chain. + +**Fix:** +```cpp +int getRandomNumber(int upperBound) { + if (upperBound <= 0) return 0; + return ndn::random::generateWord32() % upperBound; +} +``` + +--- + +### H5 — Interest and Data expiration timer keys may collide +**File:** `multicast-suppression.cpp:273–274` and `link-service.cpp:78, 126` +**Status:** [ ] Open + +Interests are keyed with `name.appendNumber(0)` and Data with `name.appendNumber(1)` in +`m_objectExpirationTimer`. However, `recordData` does `name_cop.appendNumber(0)` when looking +up whether a pending interest timer should be cancelled. If the interest and data share the exact +same base name the lookup is correct, but if naming conventions drift between send and receive +paths the keys will not match and timers will be orphaned. + +**Fix:** Document and enforce a strict naming convention for timer keys, or use a dedicated +`struct TimerKey { Name name; char type; }` as the map key. + +--- + +### H6 — `setUpdateExpiration` overwrites EventId without checking if already fired +**File:** `multicast-suppression.cpp:308–311` +**Status:** [ ] Open + +```cpp +itr_timer->second.cancel(); // may be a no-op if already fired +itr_timer->second = eventId; // assignment of EventId may not be safe +``` + +Calling `cancel()` on an already-fired event is implementation-defined in the NDN scheduler. +Safer to erase the old entry and insert fresh. + +**Fix:** +```cpp +m_objectExpirationTimer[name] = eventId; // handles create + update safely +``` + +--- + +### H7 — Forwarded status lost after entry expiration +**File:** `multicast-suppression.cpp:232` +**Status:** [ ] Open + +`getForwardedStatus` looks up `m_interestHistory` which only lives for 30ms. Once the entry +expires and is erased, a subsequent `recordInterest` call for the same name always sees +`isForwarded = false`, losing the history of whether this node previously forwarded. This can +cause the algorithm to make incorrect suppression decisions for recurring names. + +**Fix:** Persist the forwarded status inside `EMAMeasurements` so it survives the 30ms window. + +--- + +## Medium Severity + +### M1 — EMA ignore-counter logic is ambiguous +**File:** `multicast-suppression.cpp:159–165` +**Status:** [ ] Open + +```cpp +m_ignoreDuplicateRecoring = (duplicateCount > m_lastDuplicateCount) + ? (m_ignoreDuplicateRecoring + 1) : 0; + +if (m_ignoreDuplicateRecoring > 0 && m_ignoreDuplicateRecoring < MAX_IGNORE) { + return; // skip EMA update +} +``` + +This skips the first 2 consecutive increases but lets the 3rd through. The intent is not +documented. Why 3? Why not skip all of them until the count stabilizes? + +**Fix:** Add a comment explaining the design intent, or simplify to +`if (m_ignoreDuplicateRecoring < MAX_IGNORE) return;`. + +--- + +### M2 — EMA is only updated on expiration, not in real-time +**File:** `multicast-suppression.cpp:295–303` +**Status:** [ ] Open + +Duplicate counts accumulate during the 30ms window but the EMA is fed only when the entry timer +fires. The algorithm is always at least 30ms behind actual network conditions. + +**Fix:** Consider updating EMA immediately when a duplicate threshold is crossed, or document +this as an intentional design trade-off. + +--- + +### M3 — Data suppression may drop legitimate replies +**File:** `link-service.cpp:111–120` +**Status:** [ ] Open + +The comment in `sendData` acknowledges this: if multiple consumers sent Interests at slightly +different times, suppressing a Data reply could starve consumers whose Interest was not served +by the unsuppressed copy. No solution is implemented. + +**Fix:** This requires a design decision — either track per-consumer PIT entries or document +as a known limitation with a recommendation to use unicast for consumer-sensitive traffic. + +--- + +### M4 — `ssthresh` misspelled as `ssthress` +**File:** `multicast-suppression.hpp:81–83` +**Status:** [ ] Open + +```cpp +void setSSthress(double val, int factor = 2) { m_ssthress = val/factor; } +double m_ssthress; +``` + +Also `setSSthress` is defined but never called anywhere in the codebase — dead code. + +**Fix:** Rename to `ssthresh` and remove or wire up the dead method. + +--- + +## Low Severity / Typos + +### L1 — Spelling errors in identifiers +**Status:** [ ] Open + +| Location | Current | Correct | +|---|---|---| +| `multicast-suppression.cpp:18` | `MAX_PROPOGATION_DELAY` | `MAX_PROPAGATION_DELAY` | +| `multicast-suppression.hpp:95` | `m_ignoreDuplicateRecoring` | `m_ignoreDuplicateRecording` | +| `link-service.hpp/cpp` | `cancelIfSchdeuled` | `cancelIfScheduled` | +| `multicast-suppression.hpp:81` | `setSSthress` / `m_ssthress` | `setSSthresh` / `m_ssthresh` | + +--- + +### L2 — Lambda captures should be explicit +**File:** `multicast-suppression.cpp:295, 337, 353` +**Status:** [ ] Open + +`[=]` captures everything by value including unintended variables. + +**Fix:** Explicitly list captures: `[this, vec, nameTree, name]`. + +--- + +## Test Coverage Gaps + +### T1 — No test for `cancelIfSchdeuled` name mutation +The critical C1 bug has no regression test. A name passed in should not be modified. + +### T2 — No test for `UNSET` sentinel in `NameTree` +`getSuppressionTimer` with no prior insert should return a value in `[0, 2*minSuppressionTime)`, +not something derived from the unsigned-wrapped `UNSET`. + +### T3 — No test for `getRandomNumber(0)` edge case +When suppression time is 0, the modulo by zero path is never exercised. + +### T4 — No test for EMA ignore-counter boundary +No test verifies behavior at exactly `MAX_IGNORE` consecutive increases. + +### T5 — No test for forwarded-status persistence across expiration +`EntryExpiration` test only checks `interestInflight` goes false — does not verify that EMA +received the correct `wasForwarded` value after the entry expired. + +### T6 — No test for timer key collision (interest vs data same name) +No test verifies that scheduling an interest timer and a data timer for the same name do not +interfere with each other. + +--- + +## Fix Priority + +| ID | Severity | Fix Effort | Status | +|---|---|---|---| +| C1 | Critical | Trivial | ✅ Fixed | +| C2 | Critical | Medium | ✅ Fixed | +| H1 | High | Trivial | [ ] Open | +| H4 | High | Trivial | [ ] Open | +| H2 | High | Easy | [ ] Open | +| H3 | High | Easy | [ ] Open | +| H6 | High | Easy | [ ] Open | +| H7 | High | Medium | [ ] Open | +| H5 | High | Medium | [ ] Open | +| M4 | Medium | Trivial | [ ] Open | +| L1 | Low | Trivial | [ ] Open | +| M1 | Medium | Easy | [ ] Open | +| M2 | Medium | Design | [ ] Open | +| M3 | Medium | Design | [ ] Open | +| L2 | Low | Trivial | [ ] Open | +| T1–T6 | — | Medium | [ ] Open | diff --git a/docs/multicast-suppression-status.md b/docs/multicast-suppression-status.md new file mode 100644 index 00000000..8ecc8816 --- /dev/null +++ b/docs/multicast-suppression-status.md @@ -0,0 +1,119 @@ +# Adaptive Multicast Suppression (AMS) — Current Status + +## Overview + +AMS reduces redundant packet forwarding on NDN multicast faces (UDP and Ethernet). Each node +learns per-name-prefix how many duplicates it typically sees, and adaptively adjusts how long +to wait before forwarding — suppressing more when the network is over-forwarding, less when it +is under-forwarding. + +--- + +## Key Parameters + +| Parameter | Value | Description | +|---|---|---| +| `DISCOUNT_FACTOR` (α) | 0.125 | EMA smoothing factor | +| `DUPLICATE_THRESHOLD` | 1.3 | EMA above this → increase suppression | +| `MULTIPLICATIVE_INCREASE` | 1.3× | Suppression time multiplier when EMA > threshold | +| `ADATIVE_DECREASE` | 5 ms | Additive decrease when EMA ≤ threshold | +| `DEFAULT_INSTANT_LIFETIME` | 30 ms | Window to collect duplicates (2× max propagation delay) | +| `MAX_MEASURMENT_INACTIVE_PERIOD` | 300 s | EMA record expires after 5 min of inactivity | +| `minSuppressionTime` | 15 ms | Floor for suppression timer | +| `maxSuppressionTime` | 15,000 ms | Ceiling for suppression timer | +| `MAX_IGNORE` | 3 | Consecutive rising duplicate counts ignored before EMA update | + +--- + +## Component Summary + +### `NameTree` +A trie keyed by NDN name components. Stores the learned suppression time per name prefix. +On send, a longest-prefix match gives the best known suppression time for any name. +If no match exists, falls back to `minSuppressionTime`. + +### `EMAMeasurements` +Tracks the duplicate rate for a name prefix using an exponential moving average. +- EMA rises when duplicates increase → suppression time is **multiplied** by 1.3 +- EMA falls below threshold → suppression time is **decreased** by 5 ms +- Updates are ignored for up to 3 consecutive cycles if the duplicate count is still rising + (avoids reacting to transient spikes before the network stabilizes) + +### `MulticastSuppression` +Per-face state machine. Maintains: +- `m_interestHistory` / `m_dataHistory` — 30 ms window tracking in-flight names and duplicate counts +- `m_EMA_interest` / `m_EMA_data` — long-lived EMA records per name prefix +- `m_interestNameTree` / `m_dataNameTree` — suppression timers indexed by prefix +- `m_objectExpirationTimer` — scheduler events for entry expiration + +### `LinkService` (send/receive path) +The suppression gate sits here, guarded by `m_suppressionEnabled`: + +**Send path:** +1. If interest/data already in-flight → drop immediately +2. Otherwise, look up suppression timer from `NameTree` → schedule delayed send +3. If a duplicate is overheard during the wait → cancel the scheduled send + +**Receive path:** +1. Overheard interest → cancel any pending scheduled forward of the same interest +2. Overheard data → cancel pending data forward *and* any pending interest forward for same name +3. Record into history for duplicate counting + +--- + +## Integration Points + +| File | Role | +|---|---| +| `daemon/face/multicast-suppression.hpp/.cpp` | Core AMS logic | +| `daemon/face/link-service.hpp/.cpp` | Send/receive path with suppression gate | +| `daemon/face/generic-link-service.hpp/.cpp` | `enableMulticastSuppression` option, propagates to `LinkService` | +| `daemon/face/udp-factory.cpp/.hpp` | Parses `mcast_suppression` from config, sets option on UDP multicast faces | +| `daemon/face/ethernet-factory.cpp/.hpp` | Same for Ethernet multicast faces | +| `nfd.conf.sample.in` | Documents the config knob | + +--- + +## Configuration + +Suppression is **disabled by default**. To enable, add to `nfd.conf`: + +``` +face_system { + udp { + mcast_suppression yes ; enable AMS on UDP multicast faces + } + ether { + mcast_suppression yes ; enable AMS on Ethernet multicast faces + } +} +``` + +--- + +## Test Coverage + +26 unit tests in `tests/daemon/face/multicast-suppression.t.cpp`, all passing: + +| Suite | Tests | What is covered | +|---|---|---| +| `TestNameTree` | 9 | Insert, longest-prefix match, edge cases (empty name, no match, special chars) | +| `TestEMAMeasurements` | 6 | Default/custom construction, EMA update, suppression time increase/decrease | +| `TestMulticastSuppressionClass` | 11 | Record interest/data, duplicate detection, in-flight tracking, entry expiration, delay timer | + +--- + +## Known Limitations / Open Questions + +- **Dropped packet behavior** — when a forwarding is suppressed, the upstream PIT entry is not + explicitly notified. Whether this causes retransmissions or silent drops depends on the + forwarding strategy and is not yet handled. +- **Data suppression asymmetry** — Interest suppression uses a full delay+cancel loop. Data uses + the same loop but the comment in the code notes this may cause issues if multiple consumers + sent Interests at different times (risk of suppressing legitimate Data replies). +- **EMA granularity** — EMA is tracked at `name.getPrefix(-1)` (one component above the leaf). + Very flat name spaces (short names) may cause unrelated prefixes to share a suppression timer. +- **No per-face isolation** — `MulticastSuppression` is a member of `LinkService` (one instance + per face), so state is correctly isolated per face. +- **`cancelIfSchdeuled` typo** — function name has a spelling error; low priority but worth fixing. +- **No suppression for Nacks** — Nack forwarding bypasses the suppression logic entirely. diff --git a/nfd.conf.sample.in b/nfd.conf.sample.in index bf7d9ea8..07997492 100644 --- a/nfd.conf.sample.in +++ b/nfd.conf.sample.in @@ -165,6 +165,7 @@ face_system mcast_group_v6 ff02::1234 ; UDP multicast group (IPv6) mcast_port_v6 56363 ; UDP multicast port number (IPv6) mcast_ad_hoc no ; set to 'yes' to make all UDP multicast faces "ad hoc", default 'no' + mcast_suppression no ; set to 'yes' to enable adaptive multicast suppression (AMS), default 'no' ; Whitelist and blacklist can contain, in no particular order: ; - interface names, including wildcard patterns (e.g., 'ifname eth0', 'ifname en*', 'ifname wlp?s0') @@ -220,6 +221,7 @@ face_system @IF_HAVE_LIBPCAP@ mcast yes ; set to 'no' to disable Ethernet multicast, default 'yes' @IF_HAVE_LIBPCAP@ mcast_group 01:00:5E:00:17:AA ; Ethernet multicast group @IF_HAVE_LIBPCAP@ mcast_ad_hoc no ; set to 'yes' to make all Ethernet multicast faces "ad hoc", default 'no' + @IF_HAVE_LIBPCAP@ mcast_suppression no ; set to 'yes' to enable adaptive multicast suppression (AMS), default 'no' @IF_HAVE_LIBPCAP@ @IF_HAVE_LIBPCAP@ ; Whitelist and blacklist can contain, in no particular order: @IF_HAVE_LIBPCAP@ ; - interface names, including wildcard patterns (e.g., 'ifname eth0', 'ifname en*', 'ifname wlp?s0') diff --git a/tests/daemon/face/multicast-suppression.t.cpp b/tests/daemon/face/multicast-suppression.t.cpp new file mode 100644 index 00000000..1a9f83c4 --- /dev/null +++ b/tests/daemon/face/multicast-suppression.t.cpp @@ -0,0 +1,316 @@ +/* -*- Mode:C++; c-file-style:"gnu"; indent-tabs-mode:nil; -*- */ +/* + * Copyright (c) 2014-2026, Regents of the University of California, + * Arizona Board of Regents, + * Colorado State University, + * University Pierre & Marie Curie, Sorbonne University, + * Washington University in St. Louis, + * Beijing Institute of Technology, + * The University of Memphis. + * + * This file is part of NFD (Named Data Networking Forwarding Daemon). + * See AUTHORS.md for complete list of NFD authors and contributors. + * + * NFD 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. + * + * NFD 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 + * NFD, e.g., in COPYING.md file. If not, see . + */ + +#include "face/multicast-suppression.hpp" + +#include "tests/test-common.hpp" +#include "tests/daemon/global-io-fixture.hpp" + +namespace nfd::tests { + +using namespace nfd::face::ams; + +BOOST_AUTO_TEST_SUITE(Face) + +BOOST_AUTO_TEST_SUITE(TestNameTree) + +BOOST_AUTO_TEST_CASE(DefaultConstruction) +{ + NameTree tree; + BOOST_CHECK_EQUAL(tree.isLeaf, false); +} + +BOOST_AUTO_TEST_CASE(InsertAndExactLookup) +{ + NameTree tree; + tree.insert("/a/b", 100.0); + + double val = tree.longestPrefixMatch("/a/b"); + BOOST_CHECK_CLOSE(val, 100.0, 0.001); +} + +BOOST_AUTO_TEST_CASE(LongerNameMatchesPrefix) +{ + NameTree tree; + tree.insert("/a/b", 100.0); + + double val = tree.longestPrefixMatch("/a/b/c"); + BOOST_CHECK_CLOSE(val, 100.0, 0.001); +} + +BOOST_AUTO_TEST_CASE(MultiplePrefixes) +{ + NameTree tree; + tree.insert("/a/b", 50.0); + tree.insert("/a/b/c", 200.0); + + // longest match is /a/b/c + double val = tree.longestPrefixMatch("/a/b/c/d"); + BOOST_CHECK_CLOSE(val, 200.0, 0.001); + + // longest match is /a/b + val = tree.longestPrefixMatch("/a/b/d"); + BOOST_CHECK_CLOSE(val, 50.0, 0.001); +} + +BOOST_AUTO_TEST_CASE(NoMatch) +{ + NameTree tree; + tree.insert("/a/b", 100.0); + + double val = tree.longestPrefixMatch("/x/y"); + // should return UNSET sentinel (very large value) + BOOST_CHECK(val > 1000000.0); +} + +BOOST_AUTO_TEST_CASE(SpecialCharacters) +{ + NameTree tree; + tree.insert("/ndn/test-123/ABC", 75.0); + + double val = tree.longestPrefixMatch("/ndn/test-123/ABC/data"); + BOOST_CHECK_CLOSE(val, 75.0, 0.001); +} + +BOOST_AUTO_TEST_CASE(EmptyAndRootOnlyName) +{ + NameTree tree; + tree.insert("/a", 50.0); + + // empty name returns UNSET + double val = tree.longestPrefixMatch(""); + BOOST_CHECK(val > 1000000.0); + + // root-only "/" has no components, returns UNSET + val = tree.longestPrefixMatch("/"); + BOOST_CHECK(val > 1000000.0); +} + +BOOST_AUTO_TEST_CASE(SuppressionTimeOnCorrectNode) +{ + NameTree tree; + tree.insert("/a/b", 100.0); + + // exact match should find value on the /a/b node + double val = tree.longestPrefixMatch("/a/b"); + BOOST_CHECK_CLOSE(val, 100.0, 0.001); + + // partial match /a should NOT have a value (only /a/b was inserted) + val = tree.longestPrefixMatch("/a"); + BOOST_CHECK(val > 1000000.0); +} + +BOOST_AUTO_TEST_CASE(GetSuppressionTimer) +{ + NameTree tree; + tree.insert("/a/b", 100.0); + + auto timer = tree.getSuppressionTimer("/a/b/c"); + BOOST_CHECK(timer >= 0_ms); + BOOST_CHECK(timer < time::milliseconds(200)); +} + +BOOST_AUTO_TEST_SUITE_END() // TestNameTree + +BOOST_AUTO_TEST_SUITE(TestEMAMeasurements) + +BOOST_AUTO_TEST_CASE(DefaultConstruction) +{ + EMAMeasurements ema; + BOOST_CHECK_CLOSE(ema.getEMACurrent(), 0.0, 0.001); + BOOST_CHECK_CLOSE(ema.getEMAPrev(), 0.0, 0.001); + BOOST_CHECK_CLOSE(ema.getCurrentSuppressionTime(), 1.0, 0.001); +} + +BOOST_AUTO_TEST_CASE(CustomConstruction) +{ + EMAMeasurements ema(5.0, 2, 50.0); + BOOST_CHECK_CLOSE(ema.getEMACurrent(), 5.0, 0.001); + BOOST_CHECK_CLOSE(ema.getCurrentSuppressionTime(), 50.0, 0.001); +} + +BOOST_AUTO_TEST_CASE(FirstUpdate) +{ + EMAMeasurements ema; + // duplicateCount <= m_lastDuplicateCount (1) to avoid ignore counter + ema.addUpdateEMA(1, false); + + // first update sets EMA = duplicateCount directly (since current == 0) + BOOST_CHECK_CLOSE(ema.getEMACurrent(), 1.0, 0.001); +} + +BOOST_AUTO_TEST_CASE(ExponentialSmoothing) +{ + // start with EMA = 4.0 via constructor + EMAMeasurements ema(4.0, 1, 50.0); + + // duplicateCount=1 <= m_lastDuplicateCount(1), no ignore + // EMA = round((0.125*1 + 0.875*4.0)*10)/10 = round(36.25)/10 = 3.6 + ema.addUpdateEMA(1, false); + + BOOST_CHECK_CLOSE(ema.getEMACurrent(), 3.6, 1.0); +} + +BOOST_AUTO_TEST_CASE(SuppressionTimeHoldsAboveThreshold) +{ + // EMA=3.0, well above DUPLICATE_THRESHOLD (1.3), suppression at 50ms + EMAMeasurements ema(3.0, 1, 50.0); + double initial = ema.getCurrentSuppressionTime(); + + // EMA drops toward 1 but stays above threshold + // updateDelayTime: EMA decreasing but still > threshold → else branch, no change + ema.addUpdateEMA(1, true); + + BOOST_CHECK_CLOSE(ema.getCurrentSuppressionTime(), initial, 0.001); +} + +BOOST_AUTO_TEST_CASE(SuppressionTimeDecreases) +{ + // EMA=1.0, which is <= DUPLICATE_THRESHOLD (1.3) + EMAMeasurements ema(1.0, 1, 100.0); + + // duplicateCount=1 <= m_lastDuplicateCount(1), no ignore + // EMA stays at 1.0, which is <= threshold and <= prev + // triggers additive decrease: 100 - 5 = 95 + ema.addUpdateEMA(1, false); + + BOOST_CHECK(ema.getCurrentSuppressionTime() < 100.0); +} + +BOOST_AUTO_TEST_SUITE_END() // TestEMAMeasurements + +BOOST_FIXTURE_TEST_SUITE(TestMulticastSuppressionClass, GlobalIoTimeFixture) + +BOOST_AUTO_TEST_CASE(RecordFirstInterest) +{ + MulticastSuppression ms; + auto interest = makeInterest("/test/interest"); + + BOOST_CHECK_EQUAL(ms.recordInterest(*interest), 1); +} + +BOOST_AUTO_TEST_CASE(RecordDuplicateInterest) +{ + MulticastSuppression ms; + auto interest = makeInterest("/test/interest"); + + ms.recordInterest(*interest); + BOOST_CHECK_EQUAL(ms.recordInterest(*interest), 0); +} + +BOOST_AUTO_TEST_CASE(RecordFirstData) +{ + MulticastSuppression ms; + auto data = makeData("/test/data"); + + BOOST_CHECK_EQUAL(ms.recordData(*data), 1); +} + +BOOST_AUTO_TEST_CASE(RecordDuplicateData) +{ + MulticastSuppression ms; + auto data = makeData("/test/data"); + + ms.recordData(*data); + BOOST_CHECK_EQUAL(ms.recordData(*data), 0); +} + +BOOST_AUTO_TEST_CASE(DuplicateCount) +{ + MulticastSuppression ms; + auto interest = makeInterest("/test/dup"); + + ms.recordInterest(*interest); + ms.recordInterest(*interest); + ms.recordInterest(*interest); + + BOOST_CHECK_EQUAL(ms.getDuplicateCount(interest->getName(), 'i'), 3); +} + +BOOST_AUTO_TEST_CASE(DuplicateCountNonExistent) +{ + MulticastSuppression ms; + BOOST_CHECK_EQUAL(ms.getDuplicateCount(Name("/nonexistent"), 'i'), 0); +} + +BOOST_AUTO_TEST_CASE(InterestInflight) +{ + MulticastSuppression ms; + auto interest = makeInterest("/test/inflight"); + + BOOST_CHECK(!ms.interestInflight(*interest)); + ms.recordInterest(*interest); + BOOST_CHECK(ms.interestInflight(*interest)); +} + +BOOST_AUTO_TEST_CASE(DataInflight) +{ + MulticastSuppression ms; + auto data = makeData("/test/inflight"); + + BOOST_CHECK(!ms.dataInflight(*data)); + ms.recordData(*data); + BOOST_CHECK(ms.dataInflight(*data)); +} + +BOOST_AUTO_TEST_CASE(ForwardedStatus) +{ + MulticastSuppression ms; + auto interest = makeInterest("/test/forward"); + + BOOST_CHECK(!ms.getForwardedStatus(interest->getName(), 'i')); + + ms.recordInterest(*interest, true); + BOOST_CHECK(ms.getForwardedStatus(interest->getName(), 'i')); +} + +BOOST_AUTO_TEST_CASE(EntryExpiration) +{ + MulticastSuppression ms; + auto interest = makeInterest("/test/expire"); + + ms.recordInterest(*interest); + BOOST_CHECK(ms.interestInflight(*interest)); + + // advance past DEFAULT_INSTANT_LIFETIME (30ms) + advanceClocks(10_ms, 5); + + BOOST_CHECK(!ms.interestInflight(*interest)); +} + +BOOST_AUTO_TEST_CASE(GetDelayTimer) +{ + MulticastSuppression ms; + Name name("/test/delay/item"); + + auto timer = ms.getDelayTimer(name, 'i'); + BOOST_CHECK(timer >= 0_ms); +} + +BOOST_AUTO_TEST_SUITE_END() // TestMulticastSuppressionClass + +BOOST_AUTO_TEST_SUITE_END() // Face + +} // namespace nfd::tests From 733cd020fee902980b6c25214204283a9099f43a Mon Sep 17 00:00:00 2001 From: Saurab Dulal Date: Sun, 5 Apr 2026 08:15:29 -0700 Subject: [PATCH 2/2] docs: add README-MULTICAST-SUPPRESSION with paper citation, usage, config, and tuning guide Co-Authored-By: Claude Sonnet 4.6 --- README-MULTICAST-SUPPRESSION.md | 278 ++++++++++++++++++++++++++++++++ 1 file changed, 278 insertions(+) create mode 100644 README-MULTICAST-SUPPRESSION.md diff --git a/README-MULTICAST-SUPPRESSION.md b/README-MULTICAST-SUPPRESSION.md new file mode 100644 index 00000000..24c21302 --- /dev/null +++ b/README-MULTICAST-SUPPRESSION.md @@ -0,0 +1,278 @@ +# Adaptive Multicast Suppression (AMS) for NFD + +## Research Paper + +This implementation is based on the following published work: + +> **Reining in Redundant Traffic through Adaptive Duplicate Suppression in Multi-Access NDN Networks** +> Saurab Dulal, Lan Wang +> *10th ACM Conference on Information-Centric Networking (ACM ICN '23)* +> October 9–10, 2023, Reykjavik, Iceland +> [https://doi.org/10.1145/3623565.3623717](https://doi.org/10.1145/3623565.3623717) + +**Abstract:** +Named Data Networking (NDN) provides native support for multiparty communication. However, +the current NDN forwarder lacks a duplicate suppression mechanism for multicasting in a +multi-access network, potentially leading to network congestion and significant degradation +in overall packet delivery performance. In this paper, we introduce Adaptive Duplicate +Suppression (ADS) for one-hop multicasting in multi-access NDN networks. ADS utilizes the +duplicate count per Interest and Data name observed in the network to dynamically adjust the +suppression time that a node waits before forwarding a packet. We have implemented ADS in +the NDN forwarding daemon (NFD) and assessed its performance using Mini-NDN. Our evaluation +demonstrates that ADS can effectively reduce redundant network traffic under various network +conditions, resulting in significantly improved application goodput and reduced transfer times. + +**BibTeX:** +```bibtex +@inproceedings{dulal2023reining, + author = {Dulal, Saurab and Wang, Lan}, + title = {Reining in Redundant Traffic through Adaptive Duplicate Suppression in Multi-Access NDN Networks}, + booktitle = {Proceedings of the 10th ACM Conference on Information-Centric Networking (ACM ICN '23)}, + year = {2023}, + month = {October}, + address = {Reykjavik, Iceland}, + publisher = {ACM}, + doi = {10.1145/3623565.3623717}, + isbn = {979-8-4007-0403-1}, +} +``` + +--- + +## Overview + +Multicast forwarding in NDN can cause significant redundancy — when multiple nodes receive +the same Interest or Data, they may all attempt to forward it, flooding the network with +duplicate packets. + +**Adaptive Multicast Suppression (AMS)** addresses this by making each node learn, on a +per-name-prefix basis, how many duplicates it typically sees and adaptively adjusting a +suppression delay before forwarding. Nodes that hear a neighbor already forwarded a packet +cancel their own pending forward, reducing unnecessary traffic. + +The algorithm follows an **AIMD** (Additive Increase, Multiplicative Decrease) strategy: +- If duplicates are high (EMA > 1.3) → **multiply** the suppression delay by 1.3× (back off) +- If duplicates are low (EMA ≤ 1.3) → **decrease** the suppression delay by 5 ms (forward sooner) + +AMS is **disabled by default** and has zero overhead when off — stock NFD behavior is fully +preserved. + +--- + +## Prerequisites + +- NFD built from source (see [`docs/INSTALL.rst`](docs/INSTALL.rst)) +- `ndn-cxx` library installed +- Applies to **UDP multicast** and **Ethernet multicast** faces only + +--- + +## Building + +AMS is compiled into NFD by default. No additional build flags are needed. + +```shell +# Standard build +./waf configure +./waf + +# Build with unit tests (recommended during development) +./waf configure --with-tests +./waf +``` + +--- + +## Configuration + +Edit your `nfd.conf` file to enable AMS. The knob can be set independently for UDP and +Ethernet multicast faces. + +### Enable on UDP multicast faces + +``` +face_system +{ + udp + { + mcast yes ; multicast must be enabled + mcast_suppression yes ; enable AMS (default: no) + } +} +``` + +### Enable on Ethernet multicast faces + +``` +face_system +{ + ether + { + mcast yes ; multicast must be enabled + mcast_suppression yes ; enable AMS (default: no) + } +} +``` + +### Enable on both + +``` +face_system +{ + udp + { + mcast yes + mcast_suppression yes + } + ether + { + mcast yes + mcast_suppression yes + } +} +``` + +Restart NFD after changing the configuration: + +```shell +sudo nfd-stop && sudo nfd-start +``` + +--- + +## Verifying AMS is Active + +AMS logs all suppression decisions through NFD's logging system. To see suppression activity +in real time, set the log level for the relevant modules: + +```shell +# See suppression decisions on the LinkService +NDN_LOG=MulticastSuppression=INFO:LinkService=INFO nfd + +# Full debug output +NDN_LOG=MulticastSuppression=DEBUG:LinkService=DEBUG nfd +``` + +Key log messages to look for: + +| Message | Meaning | +|---|---| +| `Interest drop by suppression ... is in flight` | Duplicate Interest dropped immediately | +| `waiting Xms before forwarding` | Suppression delay applied before send | +| `overheard, duplicate forwarding dropped` | Scheduled forward cancelled after overhearing | +| `Suppression time updated with X` | EMA updated, suppression timer adjusted | +| `Moving average before: X after: Y` | EMA calculation result | + +--- + +## How It Works + +``` +Node receives Interest/Data to forward + │ + ▼ +Is this a multicast face AND is AMS enabled? + │ No → forward immediately (stock NFD behavior) + │ Yes + ▼ +Is the same name already in-flight? + │ Yes → drop (duplicate suppression) + │ No + ▼ +Look up suppression timer from NameTree (longest prefix match) + │ + ▼ +Schedule delayed forward (random value in [0, 2×suppressionTime]) + │ + During wait... + ├── Overhear same packet → cancel scheduled forward + └── Timer fires → forward and record into EMA history + │ + ▼ + After 30ms window expires: + update EMA, adjust suppressionTime (AIMD) + store new time back into NameTree +``` + +--- + +## Tunable Parameters + +All parameters are in `daemon/face/multicast-suppression.cpp`. Changing them requires +recompiling NFD. + +| Parameter | Default | Description | +|---|---|---| +| `DISCOUNT_FACTOR` | 0.125 | EMA smoothing factor (α). Lower = slower adaptation | +| `DUPLICATE_THRESHOLD` | 1.3 | EMA above this triggers suppression time increase | +| `MULTIPLICATIVE_INCREASE` | 1.3× | Factor to multiply suppression time when over threshold | +| `ADATIVE_DECREASE` | 5 ms | Amount to subtract from suppression time when under threshold | +| `DEFAULT_INSTANT_LIFETIME` | 30 ms | Duplicate collection window (2× max propagation delay) | +| `MAX_MEASURMENT_INACTIVE_PERIOD` | 300 s | EMA record expires after this period of inactivity | +| `minSuppressionTime` | 15 ms | Minimum suppression delay floor | +| `maxSuppressionTime` | 15,000 ms | Maximum suppression delay ceiling | +| `MAX_IGNORE` | 3 | Consecutive rising duplicate counts to ignore before updating EMA | + +--- + +## Running Unit Tests + +```shell +# Build with test support first +./waf configure --with-tests +./waf + +# Run all AMS unit tests +./build/unit-tests-daemon -t "Face/TestNameTree:Face/TestEMAMeasurements:Face/TestMulticastSuppressionClass" + +# Run with verbose output +./build/unit-tests-daemon --log_level=test_suite -t "Face/TestNameTree:Face/TestEMAMeasurements:Face/TestMulticastSuppressionClass" + +# Run a single test case +./build/unit-tests-daemon -t "Face/TestMulticastSuppressionClass/EntryExpiration" +``` + +Expected output: `*** No errors detected` with 26 tests across 3 suites. + +| Suite | Tests | Coverage | +|---|---|---| +| `TestNameTree` | 9 | Trie insert, longest-prefix match, edge cases | +| `TestEMAMeasurements` | 6 | EMA construction, update, suppression time adaptation | +| `TestMulticastSuppressionClass` | 11 | Record, duplicate detection, in-flight tracking, expiration, delay timer | + +--- + +## Source Files + +| File | Description | +|---|---| +| `daemon/face/multicast-suppression.hpp` | `NameTree`, `EMAMeasurements`, `MulticastSuppression` declarations | +| `daemon/face/multicast-suppression.cpp` | Core AMS algorithm implementation | +| `daemon/face/link-service.hpp` | `m_suppressionEnabled` flag, `ScopedEventId` storage | +| `daemon/face/link-service.cpp` | Send/receive path integration | +| `daemon/face/generic-link-service.hpp` | `enableMulticastSuppression` option in `Options` struct | +| `daemon/face/generic-link-service.cpp` | Propagates option to `LinkService` | +| `daemon/face/udp-factory.cpp` | Parses `mcast_suppression` from config for UDP faces | +| `daemon/face/ethernet-factory.cpp` | Parses `mcast_suppression` from config for Ethernet faces | +| `tests/daemon/face/multicast-suppression.t.cpp` | Unit tests | +| `docs/multicast-suppression-status.md` | Implementation status and component detail | +| `docs/multicast-suppression-review.md` | Code review findings and fix tracking | + +--- + +## Known Limitations + +- **Dropped packet behavior** — suppressed packets do not explicitly notify the upstream PIT + entry. Retransmission behavior depends on the forwarding strategy. +- **Data suppression asymmetry** — if multiple consumers sent Interests at different times, + suppressing a Data reply could starve some consumers. +- **EMA granularity** — suppression timers are tracked at `name.getPrefix(-1)`. Very short + names may cause unrelated prefixes to share a timer. +- **Nacks not suppressed** — Nack forwarding bypasses AMS entirely. + +--- + +## License + +AMS is part of NFD and is distributed under the GNU General Public License version 3. +See [`COPYING.md`](COPYING.md) for details.