Update orchagent to support new field pfcwd_sw_enable (sonic-net#2171)

Signed-off-by: bingwang bingwang@microsoft.com

What I did
Currently, the entry pfc_enable in table PORT_QOS_MAP is used to specify pfc and pfc_watchdog are enabled on which queues.
To avoid PFC deadlock in Dual-ToR scrnario, we are going to introduce two extra lossless queues to carry bounced back traffic.HLD.
The extra lossless queues require another two pfc_watchdogs, and the new watchdogs will be implemented by hardware due to limited resources. The hardware pfc watchdog is not covered in this PR.
To specify on which queue to enable pfc watchdog, we need to define new table pfcwd_sw_enable.

Table	                Description
pfc_enable	        Specify on which queues to enable PFC
pfcwd_sw_enable	Specify on which queues to enable software PFC watchdog
This PR is to update orchagent to support new field pfcwd_sw_enable .

As two extra lossless PGs (2 and 6) are to be added, buffermgrd is also updated in this PR to generate lossless profile for the new PGs.

Why I did it
Update orchagent to support new field pfcwd_sw_enable .

How I verified it
Verified by UT.

sudo pytest3 --dvsname=vs tests/test_pfcwd.py::TestPfcwdFunc -v --pdb
========================================================================================= test session starts =========================================================================================
platform linux -- Python 3.6.9, pytest-7.0.1, pluggy-1.0.0 -- /usr/bin/python3
cachedir: .pytest_cache
rootdir: /home/bingwang/work/sonic/sonic-buildimage-master/src/sonic-swss
plugins: flaky-3.7.0
collected 2 items                                                                                                                                                                                     

tests/test_pfcwd.py::TestPfcwdFunc::test_pfcwd_software_single_queue PASSED                                                                                                                     [ 50%]
tests/test_pfcwd.py::TestPfcwdFunc::test_pfcwd_software_multi_queue PASSED

sudo pytest test_buffer_traditional.py                
===================================================================================================================== test session starts ======================================================================================================================
platform linux -- Python 3.7.5, pytest-7.1.1, pluggy-1.0.0
rootdir: /home/bingwang/work/sonic/sonic-buildimage-master/src/sonic-swss/tests
collected 1 item                                                                                                                                                                                                                                               

test_buffer_traditional.py .

Author: bingwang-ms

cfgmgr/buffermgr.cpp

@@ -133,11 +133,11 @@ Create/update two tables: profile (in m_cfgBufferProfileTable) and port buffer (
         }
     }
 */
-task_process_status BufferMgr::doSpeedUpdateTask(string port, bool admin_up)
+task_process_status BufferMgr::doSpeedUpdateTask(string port)
 {
-    vector<FieldValueTuple> fvVectorPg, fvVectorProfile;
     string cable;
     string speed;
+    string pfc_enable;
 
     if (m_cableLenLookup.count(port) == 0)
     {
@@ -152,47 +152,70 @@ task_process_status BufferMgr::doSpeedUpdateTask(string port, bool admin_up)
         return task_process_status::task_success;
     }
 
-    speed = m_speedLookup[port];
+    if (m_portStatusLookup.count(port) == 0)
+    {
+        // admin_statue is not available yet. This can happen when notification of `PORT_QOS_MAP` table
+        // comes first. 
+        SWSS_LOG_INFO("pfc_enable status is not available for port %s", port.c_str());
+        return task_process_status::task_need_retry;
+    }
+
+    if (m_portPfcStatus.count(port) == 0)
+    {
+        // PORT_QOS_MAP is not ready yet. The notification is cleared, and buffer pg
+        // will be handled when `pfc_enable` in `PORT_QOS_MAP` table is available
+        SWSS_LOG_INFO("pfc_enable status is not available for port %s", port.c_str());
+        return task_process_status::task_success;
+    }
+    pfc_enable = m_portPfcStatus[port];
 
-    string buffer_pg_key = port + m_cfgBufferPgTable.getTableNameSeparator() + LOSSLESS_PGS;
+    speed = m_speedLookup[port];
     // key format is pg_lossless_<speed>_<cable>_profile
     string buffer_profile_key = "pg_lossless_" + speed + "_" + cable + "_profile";
     string profile_ref = buffer_profile_key;
+    
+    vector<string> lossless_pgs = tokenize(pfc_enable, ',');
 
-    m_cfgBufferPgTable.get(buffer_pg_key, fvVectorPg);
-
-    if (!admin_up && m_platform == "mellanox")
+    if (m_portStatusLookup[port] == "down" && m_platform == "mellanox")
     {
-        // Remove the entry in BUFFER_PG table if any
-        if (!fvVectorPg.empty())
+        for (auto lossless_pg : lossless_pgs)
         {
-            for (auto &prop : fvVectorPg)
+            // Remove the entry in BUFFER_PG table if any
+            vector<FieldValueTuple> fvVectorPg;
+            string buffer_pg_key = port + m_cfgBufferPgTable.getTableNameSeparator() + lossless_pg;
+
+            m_cfgBufferPgTable.get(buffer_pg_key, fvVectorPg);
+            if (!fvVectorPg.empty())
             {
-                if (fvField(prop) == "profile")
+                for (auto &prop : fvVectorPg)
                 {
-                    if (fvValue(prop) == profile_ref)
+                    if (fvField(prop) == "profile")
                     {
-                        SWSS_LOG_NOTICE("Removing PG %s from port %s which is administrative down", buffer_pg_key.c_str(), port.c_str());
-                        m_cfgBufferPgTable.del(buffer_pg_key);
-                    }
-                    else
-                    {
-                        SWSS_LOG_NOTICE("Not default profile %s is configured on PG %s, won't reclaim buffer", fvValue(prop).c_str(), buffer_pg_key.c_str());
+                        if (fvValue(prop) == profile_ref)
+                        {
+                            SWSS_LOG_NOTICE("Removing PG %s from port %s which is administrative down", buffer_pg_key.c_str(), port.c_str());
+                            m_cfgBufferPgTable.del(buffer_pg_key);
+                        }
+                        else
+                        {
+                            SWSS_LOG_NOTICE("Not default profile %s is configured on PG %s, won't reclaim buffer", fvValue(prop).c_str(), buffer_pg_key.c_str());
+                        }
                     }
                 }
             }
         }
 
         return task_process_status::task_success;
     }
-
+    
     if (m_pgProfileLookup.count(speed) == 0 || m_pgProfileLookup[speed].count(cable) == 0)
     {
-        SWSS_LOG_ERROR("Unable to create/update PG profile for port %s. No PG profile configured for speed %s and cable length %s",
-                       port.c_str(), speed.c_str(), cable.c_str());
-        return task_process_status::task_invalid_entry;
+            SWSS_LOG_ERROR("Unable to create/update PG profile for port %s. No PG profile configured for speed %s and cable length %s",
+                        port.c_str(), speed.c_str(), cable.c_str());
+            return task_process_status::task_invalid_entry;
     }
 
+    vector<FieldValueTuple> fvVectorProfile;
     // check if profile already exists - if yes - skip creation
     m_cfgBufferProfileTable.get(buffer_profile_key, fvVectorProfile);
     // Create record in BUFFER_PROFILE table
@@ -213,9 +236,10 @@ task_process_status BufferMgr::doSpeedUpdateTask(string port, bool admin_up)
 
         fvVectorProfile.push_back(make_pair("pool", INGRESS_LOSSLESS_PG_POOL_NAME));
         fvVectorProfile.push_back(make_pair("xon", m_pgProfileLookup[speed][cable].xon));
-        if (m_pgProfileLookup[speed][cable].xon_offset.length() > 0) {
+        if (m_pgProfileLookup[speed][cable].xon_offset.length() > 0)
+        {
             fvVectorProfile.push_back(make_pair("xon_offset",
-                                         m_pgProfileLookup[speed][cable].xon_offset));
+                                                m_pgProfileLookup[speed][cable].xon_offset));
         }
         fvVectorProfile.push_back(make_pair("xoff", m_pgProfileLookup[speed][cable].xoff));
         fvVectorProfile.push_back(make_pair("size", m_pgProfileLookup[speed][cable].size));
@@ -227,20 +251,28 @@ task_process_status BufferMgr::doSpeedUpdateTask(string port, bool admin_up)
         SWSS_LOG_NOTICE("Reusing existing profile '%s'", buffer_profile_key.c_str());
     }
 
-    /* Check if PG Mapping is already then log message and return. */
-    for (auto& prop : fvVectorPg)
+    for (auto lossless_pg : lossless_pgs)
     {
-        if ((fvField(prop) == "profile") && (profile_ref == fvValue(prop)))
+        vector<FieldValueTuple> fvVectorPg;
+        string buffer_pg_key = port + m_cfgBufferPgTable.getTableNameSeparator() + lossless_pg;
+
+        m_cfgBufferPgTable.get(buffer_pg_key, fvVectorPg);
+
+        /* Check if PG Mapping is already then log message and return. */
+        for (auto& prop : fvVectorPg)
         {
-            SWSS_LOG_NOTICE("PG to Buffer Profile Mapping %s already present", buffer_pg_key.c_str());
-            return task_process_status::task_success;
+            if ((fvField(prop) == "profile") && (profile_ref == fvValue(prop)))
+            {
+                SWSS_LOG_NOTICE("PG to Buffer Profile Mapping %s already present", buffer_pg_key.c_str());
+                continue;
+            }
         }
-    }
 
-    fvVectorPg.clear();
+        fvVectorPg.clear();
 
-    fvVectorPg.push_back(make_pair("profile", profile_ref));
-    m_cfgBufferPgTable.set(buffer_pg_key, fvVectorPg);
+        fvVectorPg.push_back(make_pair("profile", profile_ref));
+        m_cfgBufferPgTable.set(buffer_pg_key, fvVectorPg);
+    }
     return task_process_status::task_success;
 }
 
@@ -346,6 +378,47 @@ void BufferMgr::doBufferMetaTask(Consumer &consumer)
     }
 }
 
+/*
+Parse PORT_QOS_MAP to retrieve on which queue PFC is enable, and
+cached in a map
+*/
+void BufferMgr::doPortQosTableTask(Consumer &consumer)
+{
+    SWSS_LOG_ENTER();
+
+    auto it = consumer.m_toSync.begin();
+    while (it != consumer.m_toSync.end())
+    {
+        KeyOpFieldsValuesTuple tuple = it->second;
+        string port_name = kfvKey(tuple);
+        string op = kfvOp(tuple);
+        if (op == SET_COMMAND)
+        {
+            bool update_pfc_enable = false;
+            for (auto itp : kfvFieldsValues(tuple))
+            {
+                if (fvField(itp) == "pfc_enable")
+                {
+                    if (m_portPfcStatus.count(port_name) == 0 || m_portPfcStatus[port_name] != fvValue(itp))
+                    {
+                        m_portPfcStatus[port_name] = fvValue(itp);
+                        update_pfc_enable = true;
+                    }
+                    SWSS_LOG_INFO("Got pfc enable status for port %s status %s", port_name.c_str(), fvValue(itp).c_str());
+                    break;
+                }
+            }
+            if (update_pfc_enable)
+            {
+                // The return status is ignored
+                doSpeedUpdateTask(port_name);
+            }
+        }
+        it = consumer.m_toSync.erase(it);
+    }
+
+}
+
 void BufferMgr::doTask(Consumer &consumer)
 {
     SWSS_LOG_ENTER();
@@ -399,6 +472,12 @@ void BufferMgr::doTask(Consumer &consumer)
         return;
     }
 
+    if (table_name == CFG_PORT_QOS_MAP_TABLE_NAME)
+    {
+        doPortQosTableTask(consumer);
+        return;
+    }
+
     auto it = consumer.m_toSync.begin();
     while (it != consumer.m_toSync.end())
     {
@@ -422,7 +501,6 @@ void BufferMgr::doTask(Consumer &consumer)
             }
             else if (m_pgfile_processed && table_name == CFG_PORT_TABLE_NAME)
             {
-                bool admin_up = false;
                 for (auto i : kfvFieldsValues(t))
                 {
                     if (fvField(i) == "speed")
@@ -431,39 +509,34 @@ void BufferMgr::doTask(Consumer &consumer)
                     }
                     if (fvField(i) == "admin_status")
                     {
-                        admin_up = ("up" == fvValue(i));
+                        m_portStatusLookup[port] = fvValue(i);
                     }
                 }
 
                 if (m_speedLookup.count(port) != 0)
                 {
                     // create/update profile for port
-                    task_status = doSpeedUpdateTask(port, admin_up);
+                    task_status = doSpeedUpdateTask(port);
                 }
+            }
 
-                if (task_status != task_process_status::task_success)
-                {
+            switch (task_status)
+            {
+                case task_process_status::task_failed:
+                    SWSS_LOG_ERROR("Failed to process table update");
+                    return;
+                case task_process_status::task_need_retry:
+                    SWSS_LOG_INFO("Unable to process table update. Will retry...");
+                    ++it;
+                    break;
+                case task_process_status::task_invalid_entry:
+                    SWSS_LOG_ERROR("Failed to process invalid entry, drop it");
+                    it = consumer.m_toSync.erase(it);
+                    break;
+                default:
+                    it = consumer.m_toSync.erase(it);
                     break;
-                }
             }
         }
-
-        switch (task_status)
-        {
-            case task_process_status::task_failed:
-                SWSS_LOG_ERROR("Failed to process table update");
-                return;
-            case task_process_status::task_need_retry:
-                SWSS_LOG_INFO("Unable to process table update. Will retry...");
-                ++it;
-                break;
-            case task_process_status::task_invalid_entry:
-                SWSS_LOG_ERROR("Failed to process invalid entry, drop it");
-                it = consumer.m_toSync.erase(it);
-                break;
-            default:
-                it = consumer.m_toSync.erase(it);
-                break;
-        }
     }
 }

cfgmgr/buffermgr.h

@@ -11,7 +11,6 @@
 namespace swss {
 
 #define INGRESS_LOSSLESS_PG_POOL_NAME "ingress_lossless_pool"
-#define LOSSLESS_PGS "3-4"
 
 #define BUFFERMGR_TIMER_PERIOD 10
 
@@ -28,6 +27,8 @@ typedef std::map<std::string, speed_map_t> pg_profile_lookup_t;
 
 typedef std::map<std::string, std::string> port_cable_length_t;
 typedef std::map<std::string, std::string> port_speed_t;
+typedef std::map<std::string, std::string> port_pfc_status_t;
+typedef std::map<std::string, std::string> port_admin_status_t;
 
 class BufferMgr : public Orch
 {
@@ -56,17 +57,22 @@ class BufferMgr : public Orch
 
     pg_profile_lookup_t m_pgProfileLookup;
     port_cable_length_t m_cableLenLookup;
+    port_admin_status_t m_portStatusLookup;
     port_speed_t m_speedLookup;
     std::string getPgPoolMode();
     void readPgProfileLookupFile(std::string);
     task_process_status doCableTask(std::string port, std::string cable_length);
-    task_process_status doSpeedUpdateTask(std::string port, bool admin_up);
+    task_process_status doSpeedUpdateTask(std::string port);
     void doBufferTableTask(Consumer &consumer, ProducerStateTable &applTable);
 
     void transformSeperator(std::string &name);
 
     void doTask(Consumer &consumer);
     void doBufferMetaTask(Consumer &consumer);
+    
+    port_pfc_status_t m_portPfcStatus;
+    void doPortQosTableTask(Consumer &consumer);
+
 };
 
 }

cfgmgr/buffermgrd.cpp

@@ -215,7 +215,8 @@ int main(int argc, char **argv)
                 CFG_BUFFER_QUEUE_TABLE_NAME,
                 CFG_BUFFER_PORT_INGRESS_PROFILE_LIST_NAME,
                 CFG_BUFFER_PORT_EGRESS_PROFILE_LIST_NAME,
-                CFG_DEVICE_METADATA_TABLE_NAME
+                CFG_DEVICE_METADATA_TABLE_NAME,
+                CFG_PORT_QOS_MAP_TABLE_NAME
             };
             cfgOrchList.emplace_back(new BufferMgr(&cfgDb, &applDb, pg_lookup_file, cfg_buffer_tables));
         }

orchagent/pfcwdorch.cpp

@@ -399,9 +399,9 @@ void PfcWdSwOrch<DropHandler, ForwardHandler>::enableBigRedSwitchMode()
             continue;
         }
 
-        if (!gPortsOrch->getPortPfc(port.m_port_id, &pfcMask))
+        if (!gPortsOrch->getPortPfcWatchdogStatus(port.m_port_id, &pfcMask))
         {
-            SWSS_LOG_ERROR("Failed to get PFC mask on port %s", port.m_alias.c_str());
+            SWSS_LOG_ERROR("Failed to get PFC watchdog mask on port %s", port.m_alias.c_str());
             return;
         }
 
@@ -443,9 +443,9 @@ void PfcWdSwOrch<DropHandler, ForwardHandler>::enableBigRedSwitchMode()
             continue;
         }
 
-        if (!gPortsOrch->getPortPfc(port.m_port_id, &pfcMask))
+        if (!gPortsOrch->getPortPfcWatchdogStatus(port.m_port_id, &pfcMask))
         {
-            SWSS_LOG_ERROR("Failed to get PFC mask on port %s", port.m_alias.c_str());
+            SWSS_LOG_ERROR("Failed to get PFC watchdog mask on port %s", port.m_alias.c_str());
             return;
         }
 
@@ -489,7 +489,7 @@ bool PfcWdSwOrch<DropHandler, ForwardHandler>::registerInWdDb(const Port& port,
 
     uint8_t pfcMask = 0;
 
-    if (!gPortsOrch->getPortPfc(port.m_port_id, &pfcMask))
+    if (!gPortsOrch->getPortPfcWatchdogStatus(port.m_port_id, &pfcMask))
     {
         SWSS_LOG_ERROR("Failed to get PFC mask on port %s", port.m_alias.c_str());
         return false;

orchagent/port.h

@@ -139,7 +139,8 @@ class Port
     std::vector<sai_object_id_t> m_queue_ids;
     std::vector<sai_object_id_t> m_priority_group_ids;
     sai_port_priority_flow_control_mode_t m_pfc_asym = SAI_PORT_PRIORITY_FLOW_CONTROL_MODE_COMBINED;
-    uint8_t   m_pfc_bitmask = 0;
+    uint8_t   m_pfc_bitmask = 0;        // PFC enable bit mask
+    uint8_t   m_pfcwd_sw_bitmask = 0;   // PFC software watchdog enable
     uint16_t  m_tpid = DEFAULT_TPID;
     uint32_t  m_nat_zone_id = 0;
     uint32_t  m_vnid = VNID_NONE;