Merge branch 'vrrp_orch' of https://github.com/vvbrcm/sonic-swss-vrrporchd into vrrp_orch

Author: vvbrcm

cfgmgr/buffer_check_headroom_mellanox.lua

@@ -1,12 +1,16 @@
 -- KEYS - port name
 -- ARGV[1] - profile name
 -- ARGV[2] - new size
--- ARGV[3] - pg to add
+-- ARGV[3] - new xon
+-- ARGV[4] - new xoff
+-- ARGV[5] - pg to add
 
 local port = KEYS[1]
 local input_profile_name = ARGV[1]
 local input_profile_size = tonumber(ARGV[2])
-local new_pg = ARGV[3]
+local input_profile_xon = tonumber(ARGV[3])
+local input_profile_xoff = tonumber(ARGV[4])
+local new_pg = ARGV[5]
 
 local function is_port_with_8lanes(lanes)
     -- On Spectrum 3, ports with 8 lanes have doubled pipeline latency
@@ -55,17 +59,31 @@ end
 
 local asic_keys = redis.call('KEYS', 'ASIC_TABLE*')
 local pipeline_latency = tonumber(redis.call('HGET', asic_keys[1], 'pipeline_latency'))
+local cell_size = tonumber(redis.call('HGET', asic_keys[1], 'cell_size'))
+local port_reserved_shp = tonumber(redis.call('HGET', asic_keys[1], 'port_reserved_shp'))
+local port_max_shp = tonumber(redis.call('HGET', asic_keys[1], 'port_max_shp'))
 if is_port_with_8lanes(lanes) then
     -- The pipeline latency should be adjusted accordingly for ports with 2 buffer units
     pipeline_latency = pipeline_latency * 2 - 1
     egress_mirror_size = egress_mirror_size * 2
+    port_reserved_shp = port_reserved_shp * 2
 end
+
 local lossy_pg_size = pipeline_latency * 1024
 accumulative_size = accumulative_size + lossy_pg_size + egress_mirror_size
 
 -- Fetch all keys in BUFFER_PG according to the port
 redis.call('SELECT', appl_db)
 
+local is_shp_enabled
+local shp_size = tonumber(redis.call('HGET', 'BUFFER_POOL_TABLE:ingress_lossless_pool', 'xoff'))
+if shp_size == nil or shp_size == 0 then
+    is_shp_enabled = false
+else
+    is_shp_enabled = true
+end
+local accumulative_shared_headroom = 0
+
 local debuginfo = {}
 
 local function get_number_of_pgs(keyname)
@@ -122,26 +140,50 @@ end
 table.insert(debuginfo, 'debug:other overhead:' .. accumulative_size)
 for pg_key, profile in pairs(all_pgs) do
     local current_profile_size
+    local current_profile_xon
+    local current_profile_xoff
+    local buffer_profile_table_name = 'BUFFER_PROFILE_TABLE:'
     if profile ~= input_profile_name then
-        local referenced_profile_size = redis.call('HGET', 'BUFFER_PROFILE_TABLE:' .. profile, 'size')
+        local referenced_profile_size = redis.call('HGET', buffer_profile_table_name .. profile, 'size')
         if not referenced_profile_size then
-            referenced_profile_size = redis.call('HGET', '_BUFFER_PROFILE_TABLE:' .. profile, 'size')
+            buffer_profile_table_name = '_BUFFER_PROFILE_TABLE:'
+            referenced_profile_size = redis.call('HGET', buffer_profile_table_name .. profile, 'size')
             table.insert(debuginfo, 'debug:pending profile: ' .. profile)
         end
         current_profile_size = tonumber(referenced_profile_size)
+        current_profile_xon = tonumber(redis.call('HGET', buffer_profile_table_name .. profile, 'xon'))
+        current_profile_xoff = tonumber(redis.call('HGET', buffer_profile_table_name .. profile, 'xoff'))
     else
         current_profile_size = input_profile_size
+        current_profile_xon = input_profile_xon
+        current_profile_xoff = input_profile_xoff
     end
     if current_profile_size == 0 then
         current_profile_size = lossy_pg_size
     end
     accumulative_size = accumulative_size + current_profile_size * get_number_of_pgs(pg_key)
-    table.insert(debuginfo, 'debug:' .. pg_key .. ':' .. profile .. ':' .. current_profile_size .. ':' .. get_number_of_pgs(pg_key) .. ':accu:' .. accumulative_size)
+
+    if is_shp_enabled and current_profile_xon and current_profile_xoff then
+        if current_profile_size < current_profile_xon + current_profile_xoff then
+            accumulative_shared_headroom = accumulative_shared_headroom + (current_profile_xon + current_profile_xoff - current_profile_size) * get_number_of_pgs(pg_key)
+        end
+    end
+    table.insert(debuginfo, 'debug:' .. pg_key .. ':' .. profile .. ':' .. current_profile_size .. ':' .. get_number_of_pgs(pg_key) .. ':accu:' .. accumulative_size .. ':accu_shp:' .. accumulative_shared_headroom)
 end
 
 if max_headroom_size > accumulative_size then
-    table.insert(ret, "result:true")
-    table.insert(ret, "debug:Accumulative headroom on port " .. accumulative_size .. ", the maximum available headroom " .. max_headroom_size)
+    if is_shp_enabled then
+        local max_shp = (port_max_shp + port_reserved_shp) * cell_size
+        if accumulative_shared_headroom > max_shp then
+            table.insert(ret, "result:false")
+        else
+            table.insert(ret, "result:true")
+        end
+        table.insert(ret, "debug:Accumulative headroom on port " .. accumulative_size .. ", the maximum available headroom " .. max_headroom_size .. ", the port SHP " .. accumulative_shared_headroom .. ", max SHP " .. max_shp)
+    else
+        table.insert(ret, "result:true")
+        table.insert(ret, "debug:Accumulative headroom on port " .. accumulative_size .. ", the maximum available headroom " .. max_headroom_size)
+    end
 else
     table.insert(ret, "result:false")
     table.insert(ret, "debug:Accumulative headroom on port " .. accumulative_size .. " exceeds the maximum available headroom which is " .. max_headroom_size)

cfgmgr/buffermgrdyn.cpp

@@ -1056,14 +1056,16 @@ bool BufferMgrDynamic::isHeadroomResourceValid(const string &port, const buffer_
 
     argv.emplace_back(profile.name);
     argv.emplace_back(profile.size);
+    argv.emplace_back(profile.xon);
+    argv.emplace_back(profile.xoff);
 
     if (!new_pg.empty())
     {
         argv.emplace_back(new_pg);
     }
 
-    SWSS_LOG_INFO("Checking headroom for port %s with profile %s size %s pg %s",
-                  port.c_str(), profile.name.c_str(), profile.size.c_str(), new_pg.c_str());
+    SWSS_LOG_INFO("Checking headroom for port %s with profile %s size %s xon %s xoff %s pg %s",
+                  port.c_str(), profile.name.c_str(), profile.size.c_str(), profile.xon.c_str(), profile.xoff.c_str(), new_pg.c_str());
 
     try
     {

cfgmgr/portmgr.cpp

@@ -192,6 +192,15 @@ void PortMgr::doTask(Consumer &consumer)
                 }
             }
 
+            if (!portOk)
+            {
+                // Port configuration is handled by the orchagent. If the configuration is written to the APP DB using
+                // multiple Redis write commands, the orchagent may receive a partial configuration and create a port
+                // with incorrect settings.
+                field_values.emplace_back("mtu", mtu);
+                field_values.emplace_back("admin_status", admin_status);
+            }
+
             if (field_values.size())
             {
                 writeConfigToAppDb(alias, field_values);
@@ -201,8 +210,6 @@ void PortMgr::doTask(Consumer &consumer)
             {
                 SWSS_LOG_INFO("Port %s is not ready, pending...", alias.c_str());
 
-                writeConfigToAppDb(alias, "mtu", mtu);
-                writeConfigToAppDb(alias, "admin_status", admin_status);
                 /* Retry setting these params after the netdev is created */
                 field_values.clear();
                 field_values.emplace_back("mtu", mtu);

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