Fix port up/bfd sessions bringup notification delay issue. (#3269)

Fix port up/bfd sessions bringup notification delay issue.
Why I did it
Fix following issue: sonic-net/sonic-buildimage#19569

How I did it
Revert change in Consumer::execute(), which introduced by this commit:
9258978#diff-96451cb89f907afccbd39ddadb6d30aa21fe6fbd01b1cbaf6362078b926f1f08

The change in this commit add a while loop:
do
{
std::deque entries;
table->pops(entries);
update_size = addToSync(entries);
} while (update_size != 0);

The addToSync sync method will return the size of entries
Which means, if there are massive routes notification, other high priority notification for example port up notification will blocked until all routes notification been handled.

orchagent/orch.cpp

@@ -242,14 +242,12 @@ void Consumer::execute()
     // ConsumerBase::execute_impl<swss::ConsumerTableBase>();
     SWSS_LOG_ENTER();
 
-    size_t update_size = 0;
     auto table = static_cast<swss::ConsumerTableBase *>(getSelectable());
-    do
-    {
-        std::deque<KeyOpFieldsValuesTuple> entries;
-        table->pops(entries);
-        update_size = addToSync(entries);
-    } while (update_size != 0);
+    std::deque<KeyOpFieldsValuesTuple> entries;
+    table->pops(entries);
+
+    // add to sync
+    addToSync(entries);
 
     drain();
 }

tests/mock_tests/consumer_ut.cpp

@@ -10,6 +10,25 @@ namespace consumer_test
 {
     using namespace std;
 
+    class TestOrch : public Orch
+    {
+    public:
+        TestOrch(swss::DBConnector *db, string tableName)
+            :Orch(db, tableName),
+            m_notification_count(0)
+        {
+        }
+
+        void doTask(Consumer& consumer)
+        {
+            std::cout << "TestOrch::doTask " << consumer.m_toSync.size() << std::endl;
+            m_notification_count += consumer.m_toSync.size();
+            consumer.m_toSync.clear();
+        }
+
+        long m_notification_count;
+    };
+
     struct ConsumerTest : public ::testing::Test
     {
         shared_ptr<swss::DBConnector> m_app_db;
@@ -322,4 +341,31 @@ namespace consumer_test
         validate_syncmap(consumer->m_toSync, 1, key, exp_kofv);
 
     }
+
+    TEST_F(ConsumerTest, ConsumerPops_notification_count)
+    {
+        int consumer_pops_batch_size = 10;
+        TestOrch test_orch(m_config_db.get(), "CFG_TEST_TABLE");
+        Consumer test_consumer(
+                new swss::ConsumerStateTable(m_config_db.get(), "CFG_TEST_TABLE", consumer_pops_batch_size, 1), &test_orch, "CFG_TEST_TABLE");
+        swss::ProducerStateTable producer_table(m_config_db.get(), "CFG_TEST_TABLE");
+
+        m_config_db->flushdb();
+        for (int notification_count = 0; notification_count< consumer_pops_batch_size*2; notification_count++)
+        {
+            std::vector<FieldValueTuple> fields;
+            FieldValueTuple t("test_field", "test_value");
+            fields.push_back(t);
+            producer_table.set(std::to_string(notification_count), fields);
+
+            cout << "ConsumerPops_notification_count:: add key: " << notification_count << endl;
+        }
+
+        // consumer should pops consumer_pops_batch_size notifications 
+        test_consumer.execute();
+        ASSERT_EQ(test_orch.m_notification_count, consumer_pops_batch_size);
+
+        test_consumer.execute();
+        ASSERT_EQ(test_orch.m_notification_count, consumer_pops_batch_size*2);
+    }
 }

tests/mock_tests/mock_consumerstatetable.cpp

@@ -7,4 +7,34 @@ namespace swss
         TableName_KeySet(tableName)
     {
     }
+
+    void ConsumerStateTable::pops(std::deque<KeyOpFieldsValuesTuple> &vkco, const std::string& /*prefix*/)
+    {
+        int count = 0;
+        swss::Table table(getDbConnector(), getTableName());
+        std::vector<std::string> keys;
+        table.getKeys(keys);
+        for (const auto &key: keys)
+        {
+            // pop with batch size
+            if (count < POP_BATCH_SIZE)
+            {
+                count++;
+            }
+            else
+            {
+                break;
+            }
+
+            KeyOpFieldsValuesTuple kco;
+            kfvKey(kco) = key;
+            kfvOp(kco) = SET_COMMAND;
+            if (!table.get(key, kfvFieldsValues(kco)))
+            {
+                continue;
+            }
+            table.del(key);
+            vkco.push_back(kco);
+        }
+    }
 }