diff --git a/orchagent/notifier.h b/orchagent/notifier.h index f4fcfab42f..8f5d652ad1 100644 --- a/orchagent/notifier.h +++ b/orchagent/notifier.h @@ -1,31 +1,73 @@ -#pragma once - -#include "orch.h" - -class Notifier : public Executor { -public: - Notifier(swss::NotificationConsumer *select, Orch *orch, const std::string &name) - : Executor(select, orch, name) - { - } - - swss::NotificationConsumer *getNotificationConsumer() const - { - return static_cast(getSelectable()); - } - - void execute() override - { - auto notificationConsumer = getNotificationConsumer(); - /* Check before triggering doTask because pop() can throw an exception if there is no data */ - if (notificationConsumer->hasData()) - { - m_orch->doTask(*notificationConsumer); - } - } - - void drain() override - { - this->execute(); - } -}; \ No newline at end of file +#pragma once + +#include "orch.h" + +class Notifier : public Executor { +public: + Notifier(swss::NotificationConsumer *select, Orch *orch, const std::string &name) + : Executor(select, orch, name) + { + } + + /* + * Delegate to wrapped NotificationConsumer so Select dispatches + * notifications before table consumers. + */ + int getPri() const override + { + return getSelectable()->getPri(); + } + + /* + * Yield the Select ready-set when the Orch stalls (defers doTask without + * popping). After STALL_THRESHOLD consecutive no-progress execute() calls, + * report no cached data so lower-priority table consumers get dispatched. + */ + bool hasCachedData() override + { + if (m_noProgressCount >= STALL_THRESHOLD) + return false; + return getSelectable()->hasCachedData(); + } + + swss::NotificationConsumer *getNotificationConsumer() const + { + return static_cast(getSelectable()); + } + + void execute() override + { + auto notificationConsumer = getNotificationConsumer(); + if (notificationConsumer->hasData()) + { + m_orch->doTask(*notificationConsumer); + + /* + * If queue drained, the Orch consumed — reset the counter. + * If the Orch deferred (e.g. allPortsReady() guard), the queue + * is unchanged and we increment toward the stall threshold. + * Partial pops from a large backlog also increment, providing + * natural fairness by eventually yielding to table consumers. + */ + if (!notificationConsumer->hasCachedData()) + m_noProgressCount = 0; + else if (m_noProgressCount < STALL_THRESHOLD) + m_noProgressCount++; + } + else + { + m_noProgressCount = 0; + } + } + + void drain() override + { + this->execute(); + } + +private: + /* execute() is called twice per main-loop iteration (Select dispatch + + * OrchDaemon sweep), so 2 gives the Orch one full iteration to consume. */ + static constexpr int STALL_THRESHOLD = 2; + int m_noProgressCount = 0; +}; diff --git a/tests/mock_tests/Makefile.am b/tests/mock_tests/Makefile.am index 4a7cc7a375..74ab663084 100644 --- a/tests/mock_tests/Makefile.am +++ b/tests/mock_tests/Makefile.am @@ -61,6 +61,7 @@ tests_SOURCES = aclorch_ut.cpp \ swssnet_ut.cpp \ flowcounterrouteorch_ut.cpp \ orchdaemon_ut.cpp \ + notifier_priority_ut.cpp \ intfsorch_ut.cpp \ evpnmhorch_ut.cpp \ vxlanorch_ut.cpp \ diff --git a/tests/mock_tests/notifier_priority_ut.cpp b/tests/mock_tests/notifier_priority_ut.cpp new file mode 100644 index 0000000000..af2c3dce46 --- /dev/null +++ b/tests/mock_tests/notifier_priority_ut.cpp @@ -0,0 +1,243 @@ +/* + * Unit tests for Notifier priority delegation and stall detection. + * + * Validates: + * 1. Notifier.getPri() delegates to NotificationConsumer (pri=100). + * 2. Table consumers keep Executor-default priority (pri=0). + * 3. Select::cmp orders Notifier before table consumers. + * 4. Stall detection: after STALL_THRESHOLD no-progress cycles, + * hasCachedData() returns false to yield to table consumers. + * 5. getPri() stays constant regardless of stall state. + */ + +/* Pre-include standard library headers that conflict with + * the #define private/protected public hack. */ +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#define protected public +#define private public +#include "orch.h" +#include "select.h" +#include "notifier.h" +#undef private +#undef protected + +#include "dbconnector.h" +#include "notificationconsumer.h" +#include "consumerstatetable.h" +#include "mock_table.h" + +#include + +namespace notifier_priority_test +{ + using namespace std; + + /* Minimal Orch that does NOT call pop() — simulates allPortsReady() guard. */ + class DeferringOrch : public Orch + { + public: + DeferringOrch(swss::DBConnector *db, const string &tableName) + : Orch(db, tableName) + { + } + + void doTask(Consumer &consumer) override + { + consumer.m_toSync.clear(); + } + + void doTask(swss::NotificationConsumer &/*consumer*/) override + { + /* no-op: simulates deferral */ + } + }; + + struct NotifierPriorityTest : public ::testing::Test + { + shared_ptr m_app_db; + + NotifierPriorityTest() + { + m_app_db = make_shared("APPL_DB", 0); + } + + void SetUp() override + { + ::testing_db::reset(); + } + + void TearDown() override + { + ::testing_db::reset(); + } + }; + + /* Notifier wrapping NotificationConsumer reports pri=100, not Executor default 0. */ + TEST_F(NotifierPriorityTest, NotifierReportsNotificationConsumerPriority) + { + DeferringOrch orch(m_app_db.get(), "DUMMY_TABLE"); + + auto *notifConsumer = new swss::NotificationConsumer(m_app_db.get(), "TEST_CHANNEL"); + EXPECT_EQ(notifConsumer->getPri(), 100); + + Notifier notifier(notifConsumer, &orch, "TEST_NOTIFICATIONS"); + EXPECT_EQ(notifier.getPri(), 100); + } + + /* Consumer (table consumer) wrapping ConsumerStateTable reports pri=0. */ + TEST_F(NotifierPriorityTest, TableConsumerReportsDefaultPriority) + { + DeferringOrch orch(m_app_db.get(), "DUMMY_TABLE"); + + auto *cst = new swss::ConsumerStateTable(m_app_db.get(), "PORT_TABLE", 1, 45); + EXPECT_EQ(cst->getPri(), 45); + + Consumer consumer(cst, &orch, "PORT_TABLE"); + EXPECT_EQ(consumer.getPri(), 0); + } + + /* Select::cmp places Notifier (pri=100) before Consumer (pri=0). */ + TEST_F(NotifierPriorityTest, SelectComparatorOrdersNotifierBeforeConsumer) + { + DeferringOrch orch(m_app_db.get(), "DUMMY_TABLE"); + + auto *notifConsumer = new swss::NotificationConsumer(m_app_db.get(), "TEST_CHANNEL"); + Notifier notifier(notifConsumer, &orch, "TEST_NOTIFICATIONS"); + + auto *cst = new swss::ConsumerStateTable(m_app_db.get(), "TEST_TABLE"); + Consumer consumer(cst, &orch, "TEST_TABLE"); + + std::set readySet; + readySet.insert(¬ifier); + readySet.insert(&consumer); + + EXPECT_EQ(*readySet.begin(), static_cast(¬ifier)); + } + + /* Priority ordering is stable regardless of insertion order. */ + TEST_F(NotifierPriorityTest, SelectComparatorPriorityOverridesInsertionOrder) + { + DeferringOrch orch(m_app_db.get(), "DUMMY_TABLE"); + + auto *cst = new swss::ConsumerStateTable(m_app_db.get(), "TEST_TABLE"); + Consumer consumer(cst, &orch, "TEST_TABLE"); + + auto *notifConsumer = new swss::NotificationConsumer(m_app_db.get(), "TEST_CHANNEL"); + Notifier notifier(notifConsumer, &orch, "TEST_NOTIFICATIONS"); + + std::set readySet; + readySet.insert(&consumer); + readySet.insert(¬ifier); + + auto it = readySet.begin(); + EXPECT_EQ(*it, static_cast(¬ifier)); + ++it; + EXPECT_EQ(*it, static_cast(&consumer)); + } + + /* Stall detection: with data in the queue, hasCachedData() returns true + * below threshold but false at/above threshold — proving the override + * suppresses re-insertion, not just that the queue is empty. */ + TEST_F(NotifierPriorityTest, StallDetectionSuppressesCachedData) + { + DeferringOrch orch(m_app_db.get(), "DUMMY_TABLE"); + + auto *notifConsumer = new swss::NotificationConsumer(m_app_db.get(), "TEST_STALL"); + Notifier notifier(notifConsumer, &orch, "TEST_STALL"); + + /* Push 2 messages so underlying hasCachedData() (size > 1) is true */ + notifConsumer->m_queue->push("[\"test_op\",\"test_data\"]"); + notifConsumer->m_queue->push("[\"test_op\",\"test_data\"]"); + ASSERT_TRUE(notifConsumer->hasCachedData()); + + /* Below threshold: Notifier delegates — returns true */ + notifier.m_noProgressCount = 0; + EXPECT_TRUE(notifier.hasCachedData()); + EXPECT_EQ(notifier.getPri(), 100); + + notifier.m_noProgressCount = Notifier::STALL_THRESHOLD - 1; + EXPECT_TRUE(notifier.hasCachedData()); + EXPECT_EQ(notifier.getPri(), 100); + + /* At threshold: override returns false despite queue having data */ + notifier.m_noProgressCount = Notifier::STALL_THRESHOLD; + EXPECT_FALSE(notifier.hasCachedData()); + EXPECT_EQ(notifier.getPri(), 100); + + /* Recovery: counter reset restores delegation */ + notifier.m_noProgressCount = 0; + EXPECT_TRUE(notifier.hasCachedData()); + EXPECT_EQ(notifier.getPri(), 100); + } + + /* A stalled Notifier still sorts before Consumer in Select::cmp. + * Stall works by suppressing re-insertion (hasCachedData=false), not priority. */ + TEST_F(NotifierPriorityTest, StalledNotifierKeepsConstantPriority) + { + DeferringOrch orch(m_app_db.get(), "DUMMY_TABLE"); + + auto *notifConsumer = new swss::NotificationConsumer(m_app_db.get(), "TEST_CHANNEL"); + Notifier notifier(notifConsumer, &orch, "TEST_NOTIFICATIONS"); + + auto *cst = new swss::ConsumerStateTable(m_app_db.get(), "TEST_TABLE"); + Consumer consumer(cst, &orch, "TEST_TABLE"); + + notifier.m_noProgressCount = Notifier::STALL_THRESHOLD; + + EXPECT_EQ(notifier.getPri(), 100); + EXPECT_EQ(consumer.getPri(), 0); + + std::set readySet; + readySet.insert(&consumer); + readySet.insert(¬ifier); + + EXPECT_EQ(readySet.size(), 2u); + EXPECT_EQ(*readySet.begin(), static_cast(¬ifier)); + } + + /* Multiple Notifiers have independent stall counters — one stalling + * does not affect another (relevant during startup with FDB + port + * notifications arriving simultaneously). */ + TEST_F(NotifierPriorityTest, MultipleNotifiersHaveIndependentStallState) + { + DeferringOrch orch(m_app_db.get(), "DUMMY_TABLE"); + + auto *nc1 = new swss::NotificationConsumer(m_app_db.get(), "CHANNEL_A"); + Notifier notifierA(nc1, &orch, "NOTIF_A"); + + auto *nc2 = new swss::NotificationConsumer(m_app_db.get(), "CHANNEL_B"); + Notifier notifierB(nc2, &orch, "NOTIF_B"); + + /* Stall A only */ + notifierA.m_noProgressCount = Notifier::STALL_THRESHOLD; + notifierB.m_noProgressCount = 0; + + EXPECT_FALSE(notifierA.hasCachedData()); + /* B still delegates normally (queue empty → false, but not due to stall) */ + EXPECT_EQ(notifierB.m_noProgressCount, 0); + + /* Both keep constant priority regardless of stall state */ + EXPECT_EQ(notifierA.getPri(), 100); + EXPECT_EQ(notifierB.getPri(), 100); + + /* Stall B independently */ + notifierB.m_noProgressCount = Notifier::STALL_THRESHOLD; + EXPECT_FALSE(notifierB.hasCachedData()); + + /* Reset A — B remains stalled */ + notifierA.m_noProgressCount = 0; + EXPECT_EQ(notifierA.m_noProgressCount, 0); + EXPECT_TRUE(notifierB.m_noProgressCount >= Notifier::STALL_THRESHOLD); + } +}