New p4orch development for performance.

Change-Id: I867ce9d1d4a641b35493fb81d60813083d4404f9

Author: mint570

cfgmgr/Makefile.am

@@ -3,7 +3,7 @@ CFLAGS_SAI = -I /usr/include/sai
 LIBNL_CFLAGS = -I/usr/include/libnl3
 LIBNL_LIBS = -lnl-genl-3 -lnl-route-3 -lnl-3
 SAIMETA_LIBS = -lsaimeta -lsaimetadata -lzmq
-COMMON_LIBS = -lswsscommon
+COMMON_LIBS = -lswsscommon -lpthread
 
 bin_PROGRAMS = vlanmgrd teammgrd portmgrd intfmgrd buffermgrd vrfmgrd nbrmgrd vxlanmgrd sflowmgrd natmgrd coppmgrd tunnelmgrd macsecmgrd fabricmgrd
 

orchagent/orch.h

@@ -271,7 +271,7 @@ class Orch
     void addExecutor(Executor* executor);
     Executor *getExecutor(std::string executorName);
 
-    ResponsePublisher m_publisher;
+    ResponsePublisher m_publisher{"APPL_STATE_DB"};
 private:
     void addConsumer(swss::DBConnector *db, std::string tableName, int pri = default_orch_pri);
 };

orchagent/p4orch/p4orch.cpp

@@ -154,6 +154,8 @@ void P4Orch::doTask(Consumer &consumer)
     {
         manager->drain();
     }
+
+    m_publisher.flush();
 }
 
 void P4Orch::doTask(swss::SelectableTimer &timer)

orchagent/p4orch/p4orch.h

@@ -83,5 +83,8 @@ class P4Orch : public Orch
     // Notification consumer for port state change
     swss::NotificationConsumer *m_portStatusNotificationConsumer;
 
+    // Sepcial publisher that writes to APPL DB instead of APPL STATE DB.
+    ResponsePublisher m_publisher{"APPL_DB", /*bool buffered=*/true, /*db_write_thread=*/true};
+
     friend class p4orch::test::WcmpManagerTest;
 };

orchagent/p4orch/tests/return_code_test.cpp

@@ -119,6 +119,7 @@ TEST(ReturnCodeTest, SaiCodeToReturnCodeMapping)
         {SAI_STATUS_TABLE_FULL, StatusCode::SWSS_RC_FULL},
         {SAI_STATUS_NOT_IMPLEMENTED, StatusCode::SWSS_RC_UNIMPLEMENTED},
         {SAI_STATUS_OBJECT_IN_USE, StatusCode::SWSS_RC_IN_USE},
+        {SAI_STATUS_NOT_EXECUTED, StatusCode::SWSS_RC_NOT_EXECUTED},
         {SAI_STATUS_FAILURE, StatusCode::SWSS_RC_UNKNOWN},
         {SAI_STATUS_INVALID_ATTRIBUTE_0, StatusCode::SWSS_RC_INVALID_PARAM},
         {SAI_STATUS_INVALID_ATTRIBUTE_10, StatusCode::SWSS_RC_INVALID_PARAM},

orchagent/response_publisher.cpp

@@ -64,26 +64,52 @@ void RecordResponse(const std::string &response_channel, const std::string &key,
 
 } // namespace
 
-ResponsePublisher::ResponsePublisher(bool buffered)
-    : m_db(std::make_unique<swss::DBConnector>("APPL_STATE_DB", 0)),
-      m_pipe(std::make_unique<swss::RedisPipeline>(m_db.get())), m_buffered(buffered)
+ResponsePublisher::ResponsePublisher(const std::string &dbName, bool buffered, bool db_write_thread)
+    : m_db(std::make_unique<swss::DBConnector>(dbName, 0)), m_buffered(buffered)
 {
+    if (m_buffered)
+    {
+        m_ntf_pipe = std::make_unique<swss::RedisPipeline>(m_db.get());
+        m_pipe = std::make_unique<swss::RedisPipeline>(m_db.get());
+    }
+    else
+    {
+        m_ntf_pipe = std::make_unique<swss::RedisPipeline>(m_db.get(), 1);
+        m_pipe = std::make_unique<swss::RedisPipeline>(m_db.get(), 1);
+    }
+    if (db_write_thread)
+    {
+        m_update_thread = std::unique_ptr<std::thread>(new std::thread(&ResponsePublisher::dbUpdateThread, this));
+    }
 }
 
-void ResponsePublisher::publish(const std::string &table, const std::string &key,
-                                const std::vector<swss::FieldValueTuple> &intent_attrs, const ReturnCode &status,
-                                const std::vector<swss::FieldValueTuple> &state_attrs, bool replace)
+ResponsePublisher::~ResponsePublisher()
 {
-    // Write to the DB only if:
-    // 1) A write operation is being performed and state attributes are specified.
-    // 2) A successful delete operation.
-    if ((intent_attrs.size() && state_attrs.size()) || (status.ok() && !intent_attrs.size()))
+    if (m_update_thread != nullptr)
     {
-        writeToDB(table, key, state_attrs, intent_attrs.size() ? SET_COMMAND : DEL_COMMAND, replace);
+        {
+            std::lock_guard<std::mutex> lock(m_lock);
+            m_queue.push(entry{
+                .table = "",
+                .key = "",
+                .values = std::vector<swss::FieldValueTuple>{},
+                .op = "",
+                .replace = false,
+                .flush = false,
+                .shutdown = true,
+            });
+        }
+        m_signal.notify_one();
+        m_update_thread->join();
     }
+}
 
+void ResponsePublisher::publish(const std::string &table, const std::string &key,
+                                const std::vector<swss::FieldValueTuple> &intent_attrs, const ReturnCode &status,
+                                const std::vector<swss::FieldValueTuple> &state_attrs, bool replace)
+{
     std::string response_channel = "APPL_DB_" + table + "_RESPONSE_CHANNEL";
-    swss::NotificationProducer notificationProducer{m_pipe.get(), response_channel, m_buffered};
+    swss::NotificationProducer notificationProducer{m_ntf_pipe.get(), response_channel, m_buffered};
 
     auto intent_attrs_copy = intent_attrs;
     // Add error message as the first field-value-pair.
@@ -92,6 +118,14 @@ void ResponsePublisher::publish(const std::string &table, const std::string &key
     // Sends the response to the notification channel.
     notificationProducer.send(status.codeStr(), key, intent_attrs_copy);
     RecordResponse(response_channel, key, intent_attrs_copy, status.codeStr());
+
+    // Write to the DB only if:
+    // 1) A write operation is being performed and state attributes are specified.
+    // 2) A successful delete operation.
+    if ((intent_attrs.size() && state_attrs.size()) || (status.ok() && !intent_attrs.size()))
+    {
+        writeToDB(table, key, state_attrs, intent_attrs.size() ? SET_COMMAND : DEL_COMMAND, replace);
+    }
 }
 
 void ResponsePublisher::publish(const std::string &table, const std::string &key,
@@ -112,6 +146,33 @@ void ResponsePublisher::publish(const std::string &table, const std::string &key
 
 void ResponsePublisher::writeToDB(const std::string &table, const std::string &key,
                                   const std::vector<swss::FieldValueTuple> &values, const std::string &op, bool replace)
+{
+    if (m_update_thread != nullptr)
+    {
+        {
+            std::lock_guard<std::mutex> lock(m_lock);
+            m_queue.push(entry{
+                .table = table,
+                .key = key,
+                .values = values,
+                .op = op,
+                .replace = replace,
+                .flush = false,
+                .shutdown = false,
+            });
+        }
+        m_signal.notify_one();
+    }
+    else
+    {
+        writeToDBInternal(table, key, values, op, replace);
+    }
+    RecordDBWrite(table, key, values, op);
+}
+
+void ResponsePublisher::writeToDBInternal(const std::string &table, const std::string &key,
+                                          const std::vector<swss::FieldValueTuple> &values, const std::string &op,
+                                          bool replace)
 {
     swss::Table applStateTable{m_pipe.get(), table, m_buffered};
 
@@ -133,7 +194,6 @@ void ResponsePublisher::writeToDB(const std::string &table, const std::string &k
         if (!applStateTable.get(key, fv))
         {
             applStateTable.set(key, attrs);
-            RecordDBWrite(table, key, attrs, op);
             return;
         }
         for (auto it = attrs.cbegin(); it != attrs.cend();)
@@ -150,22 +210,70 @@ void ResponsePublisher::writeToDB(const std::string &table, const std::string &k
         if (attrs.size())
         {
             applStateTable.set(key, attrs);
-            RecordDBWrite(table, key, attrs, op);
         }
     }
     else if (op == DEL_COMMAND)
     {
         applStateTable.del(key);
-        RecordDBWrite(table, key, {}, op);
     }
 }
 
 void ResponsePublisher::flush()
 {
-    m_pipe->flush();
+    m_ntf_pipe->flush();
+    if (m_update_thread != nullptr)
+    {
+        {
+            std::lock_guard<std::mutex> lock(m_lock);
+            m_queue.push(entry{
+                .table = "",
+                .key = "",
+                .values = std::vector<swss::FieldValueTuple>{},
+                .op = "",
+                .replace = false,
+                .flush = true,
+                .shutdown = false,
+            });
+        }
+        m_signal.notify_one();
+    }
+    else
+    {
+        m_pipe->flush();
+    }
 }
 
 void ResponsePublisher::setBuffered(bool buffered)
 {
     m_buffered = buffered;
 }
+
+void ResponsePublisher::dbUpdateThread()
+{
+    while (true)
+    {
+        entry e;
+        {
+            std::unique_lock<std::mutex> lock(m_lock);
+            while (m_queue.empty())
+            {
+                m_signal.wait(lock);
+            }
+
+            e = m_queue.front();
+            m_queue.pop();
+        }
+        if (e.shutdown)
+        {
+            break;
+        }
+        if (e.flush)
+        {
+            m_pipe->flush();
+        }
+        else
+        {
+            writeToDBInternal(e.table, e.key, e.values, e.op, e.replace);
+        }
+    }
+}

orchagent/response_publisher.h

@@ -1,7 +1,11 @@
 #pragma once
 
+#include <condition_variable>
 #include <memory>
+#include <mutex>
+#include <queue>
 #include <string>
+#include <thread>
 #include <unordered_map>
 #include <vector>
 
@@ -17,9 +21,9 @@
 class ResponsePublisher : public ResponsePublisherInterface
 {
   public:
-    explicit ResponsePublisher(bool buffered = false);
+    explicit ResponsePublisher(const std::string &dbName, bool buffered = false, bool db_write_thread = false);
 
-    virtual ~ResponsePublisher() = default;
+    virtual ~ResponsePublisher();
 
     // Intent attributes are the attributes sent in the notification into the
     // redis channel.
@@ -57,8 +61,29 @@ class ResponsePublisher : public ResponsePublisherInterface
     void setBuffered(bool buffered);
 
   private:
+    struct entry
+    {
+        std::string table;
+        std::string key;
+        std::vector<swss::FieldValueTuple> values;
+        std::string op;
+        bool replace;
+        bool flush;
+        bool shutdown;
+    };
+
+    void dbUpdateThread();
+    void writeToDBInternal(const std::string &table, const std::string &key,
+                           const std::vector<swss::FieldValueTuple> &values, const std::string &op, bool replace);
+
     std::unique_ptr<swss::DBConnector> m_db;
+    std::unique_ptr<swss::RedisPipeline> m_ntf_pipe;
     std::unique_ptr<swss::RedisPipeline> m_pipe;
 
     bool m_buffered{false};
+    // Thread to write to DB.
+    std::unique_ptr<std::thread> m_update_thread;
+    std::queue<entry> m_queue;
+    mutable std::mutex m_lock;
+    std::condition_variable m_signal;
 };

orchagent/return_code.h

@@ -177,7 +177,24 @@ class ReturnCode
     ReturnCode(const sai_status_t &status, const std::string &message = "")
         : stream_(std::ios_base::out | std::ios_base::ate), is_sai_(true)
     {
-        if (m_saiStatusCodeLookup.find(status) == m_saiStatusCodeLookup.end())
+        // Non-ranged SAI codes that are not included in this lookup map will map to
+        // SWSS_RC_UNKNOWN. This includes the general SAI failure:
+        // SAI_STATUS_FAILURE.
+        static const auto *const saiStatusCodeLookup = new std::unordered_map<sai_status_t, StatusCode>({
+            {SAI_STATUS_SUCCESS, StatusCode::SWSS_RC_SUCCESS},
+            {SAI_STATUS_NOT_SUPPORTED, StatusCode::SWSS_RC_UNIMPLEMENTED},
+            {SAI_STATUS_NO_MEMORY, StatusCode::SWSS_RC_NO_MEMORY},
+            {SAI_STATUS_INSUFFICIENT_RESOURCES, StatusCode::SWSS_RC_FULL},
+            {SAI_STATUS_INVALID_PARAMETER, StatusCode::SWSS_RC_INVALID_PARAM},
+            {SAI_STATUS_ITEM_ALREADY_EXISTS, StatusCode::SWSS_RC_EXISTS},
+            {SAI_STATUS_ITEM_NOT_FOUND, StatusCode::SWSS_RC_NOT_FOUND},
+            {SAI_STATUS_TABLE_FULL, StatusCode::SWSS_RC_FULL},
+            {SAI_STATUS_NOT_IMPLEMENTED, StatusCode::SWSS_RC_UNIMPLEMENTED},
+            {SAI_STATUS_OBJECT_IN_USE, StatusCode::SWSS_RC_IN_USE},
+            {SAI_STATUS_NOT_EXECUTED, StatusCode::SWSS_RC_NOT_EXECUTED},
+        });
+
+        if (saiStatusCodeLookup->find(status) == saiStatusCodeLookup->end())
         {
             // Check for ranged SAI codes.
             if (SAI_RANGED_STATUS_IS_INVALID_ATTRIBUTE(status))
@@ -207,7 +224,7 @@ class ReturnCode
         }
         else
         {
-            status_ = m_saiStatusCodeLookup[status];
+            status_ = saiStatusCodeLookup->at(status);
         }
         stream_ << message;
     }
@@ -298,21 +315,6 @@ class ReturnCode
     }
 
   private:
-    // Non-ranged SAI codes that are not included in this lookup map will map to
-    // SWSS_RC_UNKNOWN. This includes the general SAI failure: SAI_STATUS_FAILURE.
-    std::unordered_map<sai_status_t, StatusCode> m_saiStatusCodeLookup = {
-        {SAI_STATUS_SUCCESS, StatusCode::SWSS_RC_SUCCESS},
-        {SAI_STATUS_NOT_SUPPORTED, StatusCode::SWSS_RC_UNIMPLEMENTED},
-        {SAI_STATUS_NO_MEMORY, StatusCode::SWSS_RC_NO_MEMORY},
-        {SAI_STATUS_INSUFFICIENT_RESOURCES, StatusCode::SWSS_RC_FULL},
-        {SAI_STATUS_INVALID_PARAMETER, StatusCode::SWSS_RC_INVALID_PARAM},
-        {SAI_STATUS_ITEM_ALREADY_EXISTS, StatusCode::SWSS_RC_EXISTS},
-        {SAI_STATUS_ITEM_NOT_FOUND, StatusCode::SWSS_RC_NOT_FOUND},
-        {SAI_STATUS_TABLE_FULL, StatusCode::SWSS_RC_FULL},
-        {SAI_STATUS_NOT_IMPLEMENTED, StatusCode::SWSS_RC_UNIMPLEMENTED},
-        {SAI_STATUS_OBJECT_IN_USE, StatusCode::SWSS_RC_IN_USE},
-    };
-
     StatusCode status_;
     std::stringstream stream_;
     // Whether the ReturnCode is generated from a SAI status code or not.

tests/mock_tests/fake_response_publisher.cpp

@@ -8,7 +8,10 @@
  * when needed to test code that uses response publisher. */
 std::unique_ptr<MockResponsePublisher> gMockResponsePublisher;
 
-ResponsePublisher::ResponsePublisher(bool buffered) : m_db(std::make_unique<swss::DBConnector>("APPL_STATE_DB", 0)), m_buffered(buffered) {}
+ResponsePublisher::ResponsePublisher(const std::string& dbName, bool buffered, bool db_write_thread) :
+    m_db(std::make_unique<swss::DBConnector>(dbName, 0)), m_buffered(buffered) {}
+
+ResponsePublisher::~ResponsePublisher() {}
 
 void ResponsePublisher::publish(
     const std::string& table, const std::string& key,

tests/mock_tests/response_publisher/response_publisher_ut.cpp

@@ -9,7 +9,7 @@ TEST(ResponsePublisher, TestPublish)
     DBConnector conn{"APPL_STATE_DB", 0};
     Table stateTable{&conn, "SOME_TABLE"};
     std::string value;
-    ResponsePublisher publisher{};
+    ResponsePublisher publisher{"APPL_STATE_DB"};
 
     publisher.publish("SOME_TABLE", "SOME_KEY", {{"field", "value"}}, ReturnCode(SAI_STATUS_SUCCESS));
     ASSERT_TRUE(stateTable.hget("SOME_KEY", "field", value));
@@ -21,7 +21,7 @@ TEST(ResponsePublisher, TestPublishBuffered)
     DBConnector conn{"APPL_STATE_DB", 0};
     Table stateTable{&conn, "SOME_TABLE"};
     std::string value;
-    ResponsePublisher publisher{};
+    ResponsePublisher publisher{"APPL_STATE_DB"};
 
     publisher.setBuffered(true);