Merge branch 'master' into lag_id_conflict_issue

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/dash/dash_db.py

@@ -2,6 +2,7 @@
 from dvslib.dvs_common import wait_for_result
 import typing
 import pytest
+import time
 
 from dash_api.appliance_pb2 import *
 from dash_api.vnet_pb2 import *
@@ -58,9 +59,11 @@ def __setitem__(self, key: str, pairs: typing.Union[dict, list, tuple]):
         for k, v in pairs:
             pairs_str.append((to_string(k), to_string(v)))
         self.set(key, pairs_str)
+        time.sleep(1)
 
     def __delitem__(self, key: str):
         self.delete(str(key))
+        time.sleep(1)
 
 
 class Table(swsscommon.Table):

tests/dash/test_dash_acl.py

@@ -87,10 +87,12 @@ def __setitem__(self, key: str, pairs: Union[dict, list, tuple]):
             pairs_str.append((to_string(k), to_string(v)))
         self.table.set(key, pairs_str)
         self.keys.add(key)
+        time.sleep(1)
 
     def __delitem__(self, key: str):
         self.table.delete(str(key))
         self.keys.discard(key)
+        time.sleep(1)
 
     def get_keys(self):
         return self.keys

tests/dvslib/dvs_hash.py

@@ -1,5 +1,6 @@
 """Utilities for interacting with HASH objects when writing VS tests."""
 from typing import Dict, List
+import time
 
 
 class DVSHash:
@@ -21,6 +22,7 @@ def update_switch_hash(
     ) -> None:
         """Update switch hash global in Config DB."""
         self.config_db.update_entry(self.CDB_SWITCH_HASH, self.KEY_SWITCH_HASH_GLOBAL, qualifiers)
+        time.sleep(1)
 
     def get_hash_ids(
         self,

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);
+        }
+    }
 }

tests/test_ipv6_link_local.py

@@ -59,10 +59,10 @@ def test_NeighborAddRemoveIpv6LinkLocal(self, dvs, testlog):
         time.sleep(2)
 
         # Neigh entries should contain Ipv6-link-local neighbors, should be 4
-        neigh_entries = self.pdb.get_keys("NEIGH_TABLE")
-        assert (len(neigh_entries) == 4)
+        self.pdb.wait_for_n_keys("NEIGH_TABLE", 4)
 
         found_entry = False
+        neigh_entries = self.pdb.get_keys("NEIGH_TABLE")
         for key in neigh_entries:
             if (key.find("Ethernet4:2001::2") or key.find("Ethernet0:2000::2")):
                 found_entry = True