[orchagent]: In-process SAI notification queue for ZMQ mode (Option 3)

orchagent/Makefile.am

@@ -56,6 +56,7 @@ orchagent_SOURCES = \
             orchdaemon.cpp \
             orch.cpp \
             notifications.cpp \
+            sainotificationorch.cpp \
             nhgorch.cpp \
             nhgbase.cpp \
             cbf/cbfnhgorch.cpp  \

orchagent/bfdorch.cpp

@@ -8,6 +8,7 @@
 #include "sai_serialize.h"
 #include "directory.h"
 #include "notifications.h"
+#include "sainotificationorch.h"
 #include "schema.h"
 
 using namespace std;
@@ -90,6 +91,16 @@ BfdOrch::BfdOrch(DBConnector *db, string tableName, TableConnector stateDbBfdSes
     }
     Orch::addExecutor(bfdStateNotificatier);
     register_state_change_notif = false;
+
+    if (gSaiNotificationOrch)
+    {
+        gSaiNotificationOrch->registerHandler(
+            SAI_SWITCH_NOTIFICATION_NAME_BFD_SESSION_STATE_CHANGE,
+            [this](KeyOpFieldsValuesTuple &entry)
+            {
+                handleNotification(entry);
+            });
+    }
 }
 
 BfdOrch::~BfdOrch(void)
@@ -237,39 +248,50 @@ void BfdOrch::doTask(NotificationConsumer &consumer)
         return;
     }
 
-    if (op == "bfd_session_state_change")
+    KeyOpFieldsValuesTuple entry = std::make_tuple(data, op, values);
+    handleNotification(entry);
+}
+
+void BfdOrch::handleNotification(KeyOpFieldsValuesTuple &entry)
+{
+    if (kfvOp(entry) == SAI_SWITCH_NOTIFICATION_NAME_BFD_SESSION_STATE_CHANGE)
     {
-        uint32_t count;
-        sai_bfd_session_state_notification_t *bfdSessionState = nullptr;
+        handleBfdSessionStateChangeNotification(kfvKey(entry));
+    }
+}
 
-        sai_deserialize_bfd_session_state_ntf(data, count, &bfdSessionState);
+void BfdOrch::handleBfdSessionStateChangeNotification(const std::string &data)
+{
+    uint32_t count;
+    sai_bfd_session_state_notification_t *bfdSessionState = nullptr;
 
-        for (uint32_t i = 0; i < count; i++)
-        {
-            sai_object_id_t id = bfdSessionState[i].bfd_session_id;
-            sai_bfd_session_state_t state = bfdSessionState[i].session_state;
+    sai_deserialize_bfd_session_state_ntf(data, count, &bfdSessionState);
 
-            SWSS_LOG_INFO("Get BFD session state change notification id:%" PRIx64 " state: %s", id, session_state_lookup.at(state).c_str());
+    for (uint32_t i = 0; i < count; i++)
+    {
+        sai_object_id_t id = bfdSessionState[i].bfd_session_id;
+        sai_bfd_session_state_t state = bfdSessionState[i].session_state;
 
-            if (state != bfd_session_lookup[id].state)
-            {
-                auto key = bfd_session_lookup[id].peer;
-                m_stateBfdSessionTable.hset(key, "state", session_state_lookup.at(state));
+        SWSS_LOG_INFO("Get BFD session state change notification id:%" PRIx64 " state: %s", id, session_state_lookup.at(state).c_str());
 
-                SWSS_LOG_NOTICE("BFD session state for %s changed from %s to %s", key.c_str(),
-                            session_state_lookup.at(bfd_session_lookup[id].state).c_str(), session_state_lookup.at(state).c_str());
+        if (state != bfd_session_lookup[id].state)
+        {
+            auto key = bfd_session_lookup[id].peer;
+            m_stateBfdSessionTable.hset(key, "state", session_state_lookup.at(state));
 
-                BfdUpdate update;
-                update.peer = key;
-                update.state = state;
-                notify(SUBJECT_TYPE_BFD_SESSION_STATE_CHANGE, static_cast<void *>(&update));
+            SWSS_LOG_NOTICE("BFD session state for %s changed from %s to %s", key.c_str(),
+                        session_state_lookup.at(bfd_session_lookup[id].state).c_str(), session_state_lookup.at(state).c_str());
 
-                bfd_session_lookup[id].state = state;
-            }
-        }
+            BfdUpdate update;
+            update.peer = key;
+            update.state = state;
+            notify(SUBJECT_TYPE_BFD_SESSION_STATE_CHANGE, static_cast<void *>(&update));
 
-        sai_deserialize_free_bfd_session_state_ntf(count, bfdSessionState);
+            bfd_session_lookup[id].state = state;
+        }
     }
+
+    sai_deserialize_free_bfd_session_state_ntf(count, bfdSessionState);
 }
 
 bool BfdOrch::register_bfd_state_change_notification(void)

orchagent/bfdorch.h

@@ -36,6 +36,8 @@ class BfdOrch: public Orch, public Subject
     uint32_t bfd_src_port(void);
 
     void notify_session_state_down(const std::string& key);
+    void handleNotification(swss::KeyOpFieldsValuesTuple &entry);
+    void handleBfdSessionStateChangeNotification(const std::string &data);
     bool register_bfd_state_change_notification(void);
     void update_port_number(std::vector<sai_attribute_t> &attrs);
     sai_status_t retry_create_bfd_session(sai_object_id_t &bfd_session_id, vector<sai_attribute_t> attrs);

orchagent/dash/dashhafloworch.cpp

@@ -11,6 +11,7 @@
 #include "macaddress.h"
 #include "swssnet.h"
 #include "schema.h"
+#include "sainotificationorch.h"
 
 #include <chrono>
 #include <cinttypes>
@@ -786,6 +787,16 @@ DashHaFlowOrch::DashHaFlowOrch(DBConnector *db, const vector<string> &tableNames
     Orch::addExecutor(flowBulkGetSessionNotifier);
 
     registerFlowBulkGetSessionNotifier();
+
+    if (gSaiNotificationOrch)
+    {
+        gSaiNotificationOrch->registerHandler(
+            SAI_SWITCH_NOTIFICATION_NAME_FLOW_BULK_GET_SESSION_EVENT,
+            [this](KeyOpFieldsValuesTuple &entry)
+            {
+                handleNotification(entry);
+            });
+    }
 }
 
 bool DashHaFlowOrch::registerFlowBulkGetSessionNotifier()
@@ -854,13 +865,22 @@ void DashHaFlowOrch::doTask(NotificationConsumer &consumer)
 
     consumer.pop(notification_name, data, values);
 
-    if (notification_name == SAI_SWITCH_NOTIFICATION_NAME_FLOW_BULK_GET_SESSION_EVENT)
+    KeyOpFieldsValuesTuple entry = std::make_tuple(data, notification_name, values);
+    handleNotification(entry);
+}
+
+void DashHaFlowOrch::handleNotification(KeyOpFieldsValuesTuple &entry)
+{
+    if (kfvOp(entry) == SAI_SWITCH_NOTIFICATION_NAME_FLOW_BULK_GET_SESSION_EVENT)
     {
-        handleSessionNotification(notification_name, data, values);
+        handleSessionNotification(
+                kfvOp(entry),
+                kfvKey(entry),
+                kfvFieldsValues(entry));
     }
     else
     {
-        SWSS_LOG_WARN("Unknown notification: %s", notification_name.c_str());
+        SWSS_LOG_WARN("Unknown notification: %s", kfvOp(entry).c_str());
     }
 }
 

orchagent/dash/dashhafloworch.h

@@ -163,6 +163,7 @@ class DashHaFlowOrch : public ZmqOrch
 
     void doTask(ConsumerBase &consumer);
     void doTask(swss::NotificationConsumer &consumer);
+    void handleNotification(swss::KeyOpFieldsValuesTuple &entry);
     void doTask(swss::SelectableTimer &timer);
     void doTaskFlowSyncSessionTable(ConsumerBase &consumer);
     void doTaskFlowDumpFilterTable(ConsumerBase &consumer);