[DASH] add DASH offload manager and PA validation offload

.gitignore

@@ -73,12 +73,14 @@ swssconfig/swssconfig
 swssconfig/swssplayer
 tlm_teamd/tlm_teamd
 teamsyncd/teamsyncd
+dashoffloadmanager/dashoffloadmanager
 tests/tests
 tests/mock_tests/tests_response_publisher
 tests/mock_tests/tests_fpmsyncd
 tests/mock_tests/tests_intfmgrd
 tests/mock_tests/tests_teammgrd
 tests/mock_tests/tests_portsyncd
+tests/mock_tests/tests_dashoffloadmanager
 
 
 # Test Files #

Makefile.am

@@ -1,7 +1,7 @@
 if GCOV_ENABLED
-SUBDIRS = gcovpreload fpmsyncd neighsyncd portsyncd mclagsyncd natsyncd fdbsyncd orchagent swssconfig cfgmgr tests gearsyncd
+SUBDIRS = gcovpreload fpmsyncd neighsyncd portsyncd mclagsyncd natsyncd fdbsyncd orchagent swssconfig cfgmgr tests gearsyncd dashoffloadmanager
 else
-SUBDIRS = fpmsyncd neighsyncd portsyncd mclagsyncd natsyncd fdbsyncd orchagent swssconfig cfgmgr tests gearsyncd 
+SUBDIRS = fpmsyncd neighsyncd portsyncd mclagsyncd natsyncd fdbsyncd orchagent swssconfig cfgmgr tests gearsyncd dashoffloadmanager
 endif
 
 

configure.ac

@@ -158,6 +158,7 @@ AC_CONFIG_FILES([
     mclagsyncd/Makefile
     swssconfig/Makefile
     cfgmgr/Makefile
+    dashoffloadmanager/Makefile
     tests/Makefile
     orchagent/p4orch/tests/Makefile
 ])

dashoffloadmanager/Makefile.am

@@ -0,0 +1,31 @@
+INCLUDES = -I $(top_srcdir)/lib -I $(top_srcdir) -I $(top_srcdir)/orchagent -I $(top_srcdir)/orchagent/flex_counter
+CFLAGS_SAI = -I /usr/include/sai
+DASH_LIBS = -lprotobuf -ldashapi
+MISC_LIBS = -lzmq -lswsscommon -l:libboost_program_options.a
+SAIMETA_LIBS = -lsaimeta -lsaimetadata
+
+COMMON_ORCH_SOURCE = $(top_srcdir)/orchagent/orch.cpp \
+				$(top_srcdir)/orchagent/zmqorch.cpp \
+				$(top_srcdir)/orchagent/request_parser.cpp \
+				$(top_srcdir)/orchagent/response_publisher.cpp \
+				$(top_srcdir)/lib/recorder.cpp
+
+bin_PROGRAMS = dashoffloadmanager
+
+if DEBUG
+DBGFLAGS = -ggdb -DDEBUG
+else
+DBGFLAGS = -g
+endif
+
+dashoffloadmanager_SOURCES = main.cpp dpuinfoprovider.cpp dashoffloadmanager.cpp dashoffloadpavalidation.cpp $(COMMON_ORCH_SOURCE)
+dashoffloadmanager_CPPFLAGS = $(DBGFLAGS) $(AM_CFLAGS) $(CFLAGS_COMMON) $(CFLAGS_SAI) $(CFLAGS_ASAN)
+dashoffloadmanager_LDADD = $(LDFLAGS_ASAN) $(MISC_LIBS) $(SAIMETA_LIBS) $(DASH_LIBS)
+
+if GCOV_ENABLED
+dashoffloadmanager_SOURCES += ../gcovpreload/gcovpreload.cpp
+endif
+
+if ASAN_ENABLED
+dashoffloadmanager_SOURCES += $(top_srcdir)/lib/asan.cpp
+endif

dashoffloadmanager/dashoffloadmanager.cpp

@@ -0,0 +1,177 @@
+#include "dashoffloadmanager.h"
+#include "logger_utils.h"
+
+#include "dashoffloadpavalidation.h"
+
+#include <sstream>
+
+using namespace std;
+using namespace swss;
+
+#define SELECT_TIMEOUT 60000
+
+DashOffloadManager::DashOffloadManager(const DpuInfo& dpuInfo, const std::string& zmq_server_addr, const std::string& zmq_proxy_endpoint):
+    m_dpuInfo(dpuInfo),
+    m_dpuLogKey(makeDpuLogKey(dpuInfo.dpuId)),
+    m_dpuStateKey("DPU" + to_string(dpuInfo.dpuId))
+{
+    SWSS_LOG_ENTER();
+
+    DASH_MNG_LOG_NOTICE("Starting Dash Offload Manager (zmq proxy: %s -> %s)", zmq_server_addr.c_str(), zmq_proxy_endpoint.c_str());
+
+    m_zmqProxy = make_unique<ZmqServer>(zmq_server_addr);
+
+    m_zmqProxy->enableProxyMode(zmq_proxy_endpoint);
+
+    m_applDb = make_unique<swss::DBConnector>("APPL_DB", 0);
+
+    m_dpuStateDb = make_unique<swss::DBConnector>("CHASSIS_STATE_DB", 0);
+};
+
+void DashOffloadManager::start()
+{
+    SWSS_LOG_ENTER();
+
+    m_thread = make_shared<thread>(&DashOffloadManager::offload_handle, this);
+}
+
+void DashOffloadManager::join()
+{
+    m_thread->join();
+}
+
+OffloadState DashOffloadManager::getOffloadState() const
+{
+    return m_offloadState;
+}
+
+void DashOffloadManager::enableDefaultOffload()
+{
+    SWSS_LOG_ENTER();
+
+    enablePaValidationOffload();
+}
+
+void DashOffloadManager::disableOffload()
+{
+    SWSS_LOG_ENTER();
+
+    for (auto &orch : m_offloadOrchList)
+    {
+        auto orch_selecatables = orch->getSelectables();
+        for (auto &sel : orch_selecatables)
+        {
+            m_select.removeSelectable(sel);
+        }
+    }
+
+    m_offloadOrchList.clear();
+    m_offloadState.pa_validation = false;
+}
+
+void DashOffloadManager::processDpuStateUpdate(bool dpu_down)
+{
+    SWSS_LOG_ENTER();
+
+    if (dpu_down == m_dpu_state_down)
+    {
+        return;
+    }
+
+    if (dpu_down)
+    {
+        DASH_MNG_LOG_NOTICE("DPU is down - stopping the offload");
+        disableOffload();
+    } else {
+        enableDefaultOffload();
+    }
+
+    m_dpu_state_down = dpu_down;
+}
+
+void DashOffloadManager::processDpuStateTableUpdate(swss::SubscriberStateTable *table)
+{
+    SWSS_LOG_ENTER();
+
+    std::deque<KeyOpFieldsValuesTuple> entries;
+    table->pops(entries);
+
+    for (auto &entry : entries)
+    {
+        auto key = kfvKey(entry);
+
+        if (key == m_dpuStateKey)
+        {
+            for (auto i : kfvFieldsValues(entry))
+            {
+                if (fvField(i) == "dpu_control_plane_state")
+                {
+                    bool dpu_down = fvValue(i) == "down";
+                    return processDpuStateUpdate(dpu_down);
+                }
+            }
+        }
+    }
+}
+
+void DashOffloadManager::enablePaValidationOffload()
+{
+    SWSS_LOG_ENTER();
+
+    if (m_offloadState.pa_validation)
+    {
+        return;
+    }
+
+    Orch *orch = new DashPaVlidationOffloadOrch(m_dpuInfo, m_applDb.get(), m_zmqProxy.get());
+    m_select.addSelectables(orch->getSelectables());
+
+    m_offloadOrchList.push_back(unique_ptr<Orch>(orch));
+
+    m_offloadState.pa_validation = true;
+}
+
+void DashOffloadManager::offload_handle()
+{
+    SWSS_LOG_ENTER();
+
+    auto dpuStateTable = make_unique<SubscriberStateTable>(m_dpuStateDb.get(), CHASSIS_STATE_DPU_STATE_TABLE_NAME, TableConsumable::DEFAULT_POP_BATCH_SIZE, 0);
+    m_select.addSelectable(dpuStateTable.get());
+
+    while (!m_stop_thread)
+    {
+        Selectable *sel;
+        int ret;
+
+        ret = m_select.select(&sel, SELECT_TIMEOUT);
+        if (ret == Select::ERROR)
+        {
+            DASH_MNG_LOG_ERROR("Error from select - %s",  strerror(errno));
+            continue;
+        }
+
+        if (ret == Select::TIMEOUT)
+        {
+            for (auto &o : m_offloadOrchList)
+            {
+                o->doTask();
+            }
+
+            continue;
+        }
+
+        if (sel == dpuStateTable.get())
+        {
+            processDpuStateTableUpdate(dpuStateTable.get());
+        } else
+        {
+            auto *c = (Executor *)sel;
+            c->execute();
+        }
+
+        for (auto &o : m_offloadOrchList)
+        {
+            o->doTask();
+        }
+    }
+}

dashoffloadmanager/dashoffloadmanager.h

@@ -0,0 +1,48 @@
+#pragma once
+
+#include "dbconnector.h"
+#include "zmqserver.h"
+#include "select.h"
+#include "subscriberstatetable.h"
+#include "orch.h"
+#include <thread>
+
+#include "dpuinfoprovider.h"
+
+struct OffloadState
+{
+    bool pa_validation = false;
+};
+
+class DashOffloadManager
+{
+public:
+    DashOffloadManager(const DpuInfo& dpuInfo, const std::string& zmq_server_addr, const std::string& zmq_proxy_endpoint);
+    void start();
+    void join();
+    OffloadState getOffloadState() const;
+
+private:
+    DpuInfo m_dpuInfo;
+    std::string m_dpuLogKey;
+    std::string m_dpuStateKey;
+    bool m_dpu_state_down = true;
+    OffloadState m_offloadState;
+
+    swss::Select m_select;
+    std::shared_ptr<std::thread> m_thread;
+    std::atomic<bool> m_stop_thread { false };
+    std::unique_ptr<swss::ZmqServer> m_zmqProxy;
+    std::unique_ptr<swss::DBConnector> m_applDb;
+    std::unique_ptr<swss::DBConnector> m_dpuStateDb;
+    std::vector<std::unique_ptr<Orch>> m_offloadOrchList;
+
+    void offload_handle();
+    void enableDefaultOffload();
+    void disableOffload();
+    void processDpuStateTableUpdate(swss::SubscriberStateTable *table);
+    void processOffloadTablePaValidation(swss::KeyOpFieldsValuesTuple &entry);
+    void processDpuStateUpdate(bool dpu_down);
+
+    void enablePaValidationOffload();
+};