[PFC WD]: Initial support

meta/saiserialize.cpp

@@ -708,6 +708,14 @@ std::string sai_serialize_port_stat(
     return sai_serialize_enum(counter, &sai_metadata_enum_sai_port_stat_t);
 }
 
+std::string sai_serialize_queue_stat(
+        _In_ const sai_queue_stat_t counter)
+{
+    SWSS_LOG_ENTER();
+
+    return sai_serialize_enum(counter, &sai_metadata_enum_sai_queue_stat_t);
+}
+
 std::string sai_serialize_switch_oper_status(
         _In_ sai_object_id_t switch_id,
         _In_ sai_switch_oper_status_t status)
@@ -2701,3 +2709,22 @@ void sai_deserialize_free_port_oper_status_ntf(
 
     delete port_oper_status;
 }
+
+void sai_deserialize_port_stat(
+        _In_ const std::string& s,
+        _Out_ sai_port_stat_t& stat)
+{
+    SWSS_LOG_ENTER();
+
+    sai_deserialize_enum(s, &sai_metadata_enum_sai_port_stat_t, (int32_t&)stat);
+}
+
+void sai_deserialize_queue_stat(
+        _In_ const std::string& s,
+        _Out_ sai_queue_stat_t& stat)
+{
+    SWSS_LOG_ENTER();
+
+    sai_deserialize_enum(s, &sai_metadata_enum_sai_queue_stat_t, (int32_t&)stat);
+}
+

meta/saiserialize.h

@@ -72,6 +72,9 @@ std::string sai_serialize_common_api(
 std::string sai_serialize_port_stat(
         _In_ const sai_port_stat_t counter);
 
+std::string sai_serialize_queue_stat(
+        _In_ const sai_queue_stat_t counter);
+
 std::string sai_serialize_switch_oper_status(
         _In_ sai_object_id_t switch_id,
         _In_ sai_switch_oper_status_t status);
@@ -196,4 +199,12 @@ void sai_deserialize_free_port_oper_status_ntf(
         _In_ uint32_t count,
         _In_ sai_port_oper_status_notification_t* portoperstatus);
 
+void sai_deserialize_port_stat(
+        _In_ const std::string& s,
+        _Out_ sai_port_stat_t& stat);
+
+void sai_deserialize_queue_stat(
+        _In_ const std::string& s,
+        _Out_ sai_queue_stat_t& stat);
+
 #endif // __SAI_SERIALIZE__

syncd/Makefile.am

@@ -20,7 +20,8 @@ syncd_SOURCES = \
 				syncd_hard_reinit.cpp \
 				syncd_notifications.cpp \
 				syncd_counters.cpp \
-				syncd_applyview.cpp
+				syncd_applyview.cpp \
+				syncd_pfc_watchdog.cpp
 
 syncd_CPPFLAGS = $(DBGFLAGS) $(AM_CPPFLAGS) $(CFLAGS_COMMON) $(SAIFLAGS)
 syncd_LDADD = -lhiredis -lswsscommon $(SAILIB) -lpthread -L$(top_srcdir)/meta/.libs -lsaimetadata -lsaimeta -ldl

syncd/syncd.cpp

@@ -1,6 +1,7 @@
 #include "syncd.h"
 #include "syncd_saiswitch.h"
 #include "sairedis.h"
+#include "syncd_pfc_watchdog.h"
 #include "swss/tokenize.h"
 #include <limits.h>
 
@@ -2330,6 +2331,62 @@ sai_status_t processEvent(
     return status;
 }
 
+void processPfcWdEvent(
+        _In_ swss::ConsumerStateTable &consumer)
+{
+    std::lock_guard<std::mutex> lock(g_mutex);
+
+    SWSS_LOG_ENTER();
+
+    swss::KeyOpFieldsValuesTuple kco;
+    consumer.pop(kco);
+
+    const auto &key = kfvKey(kco);
+    const auto &op = kfvOp(kco);
+
+    sai_object_id_t queueVid = SAI_NULL_OBJECT_ID;
+    sai_deserialize_object_id(key, queueVid);
+    sai_object_id_t queueId = translate_vid_to_rid(queueVid);
+
+    const auto values = kfvFieldsValues(kco);
+    for (const auto& valuePair : values)
+    {
+        const auto field = fvField(valuePair);
+        const auto value = fvValue(valuePair);
+
+        if (op == DEL_COMMAND)
+        {
+            PfcWatchdog::removeQueue(queueVid);
+            continue;
+        }
+
+        auto idStrings  = swss::tokenize(value, ',');
+
+        if (field == PFC_WD_PORT_COUNTER_ID_LIST)
+        {
+            std::vector<sai_port_stat_t> portCounterIds;
+            for (const auto &str : idStrings)
+            {
+                sai_port_stat_t stat;
+                sai_deserialize_port_stat(str, stat);
+                portCounterIds.push_back(stat);
+            }
+            PfcWatchdog::setPortCounterList(queueVid, queueId, portCounterIds);
+        }
+        else if (field == PFC_WD_QUEUE_COUNTER_ID_LIST)
+        {
+            std::vector<sai_queue_stat_t> queueCounterIds;
+            for (const auto &str : idStrings)
+            {
+                sai_queue_stat_t stat;
+                sai_deserialize_queue_stat(str, stat);
+                queueCounterIds.push_back(stat);
+            }
+            PfcWatchdog::setQueueCounterList(queueVid, queueId, queueCounterIds);
+        }
+    }
+}
+
 void printUsage()
 {
     std::cout << "Usage: syncd [-N] [-d] [-p profile] [-i interval] [-t [cold|warm|fast]] [-h] [-u] [-S]" << std::endl;
@@ -2901,21 +2958,23 @@ int main(int argc, char **argv)
     }
 #endif // SAITHRIFT
 
-    std::shared_ptr<swss::DBConnector> db = std::make_shared<swss::DBConnector>(ASIC_DB, swss::DBConnector::DEFAULT_UNIXSOCKET, 0);
+    std::shared_ptr<swss::DBConnector> dbAsic = std::make_shared<swss::DBConnector>(ASIC_DB, swss::DBConnector::DEFAULT_UNIXSOCKET, 0);
     std::shared_ptr<swss::DBConnector> dbNtf = std::make_shared<swss::DBConnector>(ASIC_DB, swss::DBConnector::DEFAULT_UNIXSOCKET, 0);
+    std::shared_ptr<swss::DBConnector> dbPfcWatchdog = std::make_shared<swss::DBConnector>(PFC_WD_DB, swss::DBConnector::DEFAULT_UNIXSOCKET, 0);
 
-    g_redisClient = std::make_shared<swss::RedisClient>(db.get());
+    g_redisClient = std::make_shared<swss::RedisClient>(dbAsic.get());
 
-    std::shared_ptr<swss::ConsumerTable> asicState = std::make_shared<swss::ConsumerTable>(db.get(), ASIC_STATE_TABLE);
-    std::shared_ptr<swss::NotificationConsumer> restartQuery = std::make_shared<swss::NotificationConsumer>(db.get(), "RESTARTQUERY");
+    std::shared_ptr<swss::ConsumerTable> asicState = std::make_shared<swss::ConsumerTable>(dbAsic.get(), ASIC_STATE_TABLE);
+    std::shared_ptr<swss::NotificationConsumer> restartQuery = std::make_shared<swss::NotificationConsumer>(dbAsic.get(), "RESTARTQUERY");
+    std::shared_ptr<swss::ConsumerStateTable> pfcWdState = std::make_shared<swss::ConsumerStateTable>(dbPfcWatchdog.get(), PFC_WD_STATE_TABLE);
 
     /*
      * At the end we cant use producer consumer concept since if one proces
      * will restart there may be something in the queue also "remove" from
      * response queue will also trigger another "response".
      */
 
-    getResponse  = std::make_shared<swss::ProducerTable>(db.get(), "GETRESPONSE");
+    getResponse  = std::make_shared<swss::ProducerTable>(dbAsic.get(), "GETRESPONSE");
     notifications = std::make_shared<swss::NotificationProducer>(dbNtf.get(), "NOTIFICATIONS");
 
     g_veryFirstRun = isVeryFirstRun();
@@ -3003,6 +3062,7 @@ int main(int argc, char **argv)
 
         s.addSelectable(asicState.get());
         s.addSelectable(restartQuery.get());
+        s.addSelectable(pfcWdState.get());
 
         SWSS_LOG_NOTICE("starting main loop");
 
@@ -3027,8 +3087,11 @@ int main(int argc, char **argv)
                 warmRestartHint = handleRestartQuery(*restartQuery);
                 break;
             }
-
-            if (result == swss::Select::OBJECT)
+            else if (sel == pfcWdState.get())
+            {
+                processPfcWdEvent(*(swss::ConsumerStateTable*)sel);
+            }
+            else if (result == swss::Select::OBJECT)
             {
                 processEvent(*(swss::ConsumerTable*)sel);
             }

syncd/syncd.h

@@ -31,6 +31,7 @@ extern "C" {
 #include "swss/dbconnector.h"
 #include "swss/producertable.h"
 #include "swss/consumertable.h"
+#include "swss/consumerstatetable.h"
 #include "swss/notificationconsumer.h"
 #include "swss/notificationproducer.h"
 #include "swss/selectableevent.h"
@@ -100,7 +101,6 @@ void startCountersThread(
         _In_ int intervalInSeconds);
 
 sai_status_t syncdApplyView();
-
 void check_notifications_pointers(
         _In_ uint32_t attr_count,
         _In_ sai_attribute_t *attr_list);