[orchagent]: Fix notification priority dead code in Executor/Notifier

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<swss::NotificationConsumer *>(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();
-    }
-};
+#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<swss::NotificationConsumer *>(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;
+};

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 \

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 <string>
+#include <fstream>
+#include <iostream>
+#include <sstream>
+#include <memory>
+#include <set>
+#include <deque>
+#include <mutex>
+#include <thread>
+#include <condition_variable>
+#include <atomic>
+
+#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 <gtest/gtest.h>
+
+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<swss::DBConnector> m_app_db;
+
+        NotifierPriorityTest()
+        {
+            m_app_db = make_shared<swss::DBConnector>("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<swss::Selectable *, swss::Select::cmp> readySet;
+        readySet.insert(&notifier);
+        readySet.insert(&consumer);
+
+        EXPECT_EQ(*readySet.begin(), static_cast<swss::Selectable *>(&notifier));
+    }
+
+    /* 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<swss::Selectable *, swss::Select::cmp> readySet;
+        readySet.insert(&consumer);
+        readySet.insert(&notifier);
+
+        auto it = readySet.begin();
+        EXPECT_EQ(*it, static_cast<swss::Selectable *>(&notifier));
+        ++it;
+        EXPECT_EQ(*it, static_cast<swss::Selectable *>(&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<swss::Selectable *, swss::Select::cmp> readySet;
+        readySet.insert(&consumer);
+        readySet.insert(&notifier);
+
+        EXPECT_EQ(readySet.size(), 2u);
+        EXPECT_EQ(*readySet.begin(), static_cast<swss::Selectable *>(&notifier));
+    }
+
+    /* 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);
+    }
+}