Enforce the order when the shared headroom pool is enabled (#2729)

What I did

Enforce the order when the shared headroom pool is enabled.
This is to backport #2699 to 202211.

Signed-off-by: Stephen Sun [email protected]

Why I did it

The current flow to enable the shared headroom pool

Configure the shared headroom pool size or over-subscribe ratio
Update lossless buffer profiles with xon == size
Calculate and update the shared headroom pool size.
In step 2, the lossless buffer profiles have been updated to values as if the shared headroom pool is enabled. However, it is enabled only in step 3, which is inconsistent between steps 2 and 3. Therefore, we open the PR to guarantee the order.

The new flow

A user configures the shared headroom pool size or over-subscribe ratio
The dynamic buffer manager invokes the vendor-specific Lua plugin to calculate the shared headroom pool size
This is the step introduced in this PR to guarantee the shared headroom pool will be enabled in advance
On Nvidia platform, a non-zero shared headroom pool is returned in this stage if the user configures the over-subscribe ratio
If a non-zero shared headroom pool is returned, the dynamic buffer manager pushes the shared headroom pool size to APPL_DB.ingress_lossless_pool and blocks until it has been updated into APPL_STATE_DB.ingress_lossless_pool (which indicates the buffer orchagent finishes handling it)
The buffer manager updates the lossless buffer profiles
The buffer manager invokes the Lua plugin to calculate the shared headroom pool size.
The flow continues as normal.
How I verified it

Manually test and regression test

Author: stephenxs

cfgmgr/buffer_pool_mellanox.lua

@@ -324,7 +324,7 @@ for name in pairs(profiles) do
                size = size + lossypg_reserved
             end
             if size ~= 0 then
-                if shp_enabled and shp_size == 0 then
+                if shp_size == 0 then
                     local xon = tonumber(redis.call('HGET', name, 'xon'))
                     local xoff = tonumber(redis.call('HGET', name, 'xoff'))
                     if xon ~= nil and xoff ~= nil and xon + xoff > size then
@@ -346,6 +346,12 @@ accumulative_occupied_buffer = accumulative_occupied_buffer + lossypg_extra_for_
 
 -- Accumulate sizes for private headrooms
 local accumulative_private_headroom = 0
+local force_enable_shp = false
+if accumulative_xoff > 0 and shp_enabled ~= true then
+    force_enable_shp = true
+    shp_size = 655360
+    shp_enabled = true
+end
 if shp_enabled then
     accumulative_private_headroom = lossless_port_count * private_headroom
     accumulative_occupied_buffer = accumulative_occupied_buffer + accumulative_private_headroom
@@ -391,6 +397,9 @@ end
 
 if shp_enabled and shp_size == 0 then
     shp_size = math.ceil(accumulative_xoff / over_subscribe_ratio)
+    if shp_size == 0 then
+        shp_size = 655360
+    end
 end
 
 local pool_size
@@ -432,6 +441,7 @@ table.insert(result, "debug:mgmt_pool:" .. mgmt_pool_size)
 if shp_enabled then
     table.insert(result, "debug:accumulative_private_headroom:" .. accumulative_private_headroom)
     table.insert(result, "debug:accumulative xoff:" .. accumulative_xoff)
+    table.insert(result, "debug:force enabled shp:" .. tostring(force_enable_shp))
 end
 table.insert(result, "debug:accumulative_mgmt_pg:" .. accumulative_management_pg)
 table.insert(result, "debug:egress_mirror:" .. accumulative_egress_mirror_overhead)

cfgmgr/buffermgrd.cpp

@@ -189,6 +189,8 @@ int main(int argc, char **argv)
             WarmStart::initialize("buffermgrd", "swss");
             WarmStart::checkWarmStart("buffermgrd", "swss");
 
+            DBConnector applStateDb("APPL_STATE_DB", 0);
+
             vector<TableConnector> buffer_table_connectors = {
                 TableConnector(&cfgDb, CFG_PORT_TABLE_NAME),
                 TableConnector(&cfgDb, CFG_PORT_CABLE_LEN_TABLE_NAME),
@@ -202,7 +204,7 @@ int main(int argc, char **argv)
                 TableConnector(&stateDb, STATE_BUFFER_MAXIMUM_VALUE_TABLE),
                 TableConnector(&stateDb, STATE_PORT_TABLE_NAME)
             };
-            cfgOrchList.emplace_back(new BufferMgrDynamic(&cfgDb, &stateDb, &applDb, buffer_table_connectors, peripherial_table_ptr, zero_profiles_ptr));
+            cfgOrchList.emplace_back(new BufferMgrDynamic(&cfgDb, &stateDb, &applDb, &applStateDb, buffer_table_connectors, peripherial_table_ptr, zero_profiles_ptr));
         }
         else if (!pg_lookup_file.empty())
         {

cfgmgr/buffermgrdyn.cpp

@@ -26,7 +26,7 @@
 using namespace std;
 using namespace swss;
 
-BufferMgrDynamic::BufferMgrDynamic(DBConnector *cfgDb, DBConnector *stateDb, DBConnector *applDb, const vector<TableConnector> &tables, shared_ptr<vector<KeyOpFieldsValuesTuple>> gearboxInfo, shared_ptr<vector<KeyOpFieldsValuesTuple>> zeroProfilesInfo) :
+BufferMgrDynamic::BufferMgrDynamic(DBConnector *cfgDb, DBConnector *stateDb, DBConnector *applDb, DBConnector *applStateDb, const vector<TableConnector> &tables, shared_ptr<vector<KeyOpFieldsValuesTuple>> gearboxInfo, shared_ptr<vector<KeyOpFieldsValuesTuple>> zeroProfilesInfo) :
         Orch(tables),
         m_platform(),
         m_bufferDirections{BUFFER_INGRESS, BUFFER_EGRESS},
@@ -38,6 +38,7 @@ BufferMgrDynamic::BufferMgrDynamic(DBConnector *cfgDb, DBConnector *stateDb, DBC
         m_cfgDefaultLosslessBufferParam(cfgDb, CFG_DEFAULT_LOSSLESS_BUFFER_PARAMETER),
         m_cfgDeviceMetaDataTable(cfgDb, CFG_DEVICE_METADATA_TABLE_NAME),
         m_applBufferPoolTable(applDb, APP_BUFFER_POOL_TABLE_NAME),
+        m_applStateBufferPoolTable(applStateDb, APP_BUFFER_POOL_TABLE_NAME),
         m_applBufferProfileTable(applDb, APP_BUFFER_PROFILE_TABLE_NAME),
         m_applBufferObjectTables{ProducerStateTable(applDb, APP_BUFFER_PG_TABLE_NAME), ProducerStateTable(applDb, APP_BUFFER_QUEUE_TABLE_NAME)},
         m_applBufferProfileListTables{ProducerStateTable(applDb, APP_BUFFER_PORT_INGRESS_PROFILE_LIST_NAME), ProducerStateTable(applDb, APP_BUFFER_PORT_EGRESS_PROFILE_LIST_NAME)},
@@ -1960,6 +1961,13 @@ task_process_status BufferMgrDynamic::handleDefaultLossLessBufferParam(KeyOpFiel
     {
         bool isSHPEnabled = isNonZero(m_overSubscribeRatio);
         bool willSHPBeEnabled = isNonZero(newRatio);
+        if (m_portInitDone && (!isSHPEnabled) && willSHPBeEnabled)
+        {
+            if (!isSharedHeadroomPoolEnabledInSai())
+            {
+                return task_process_status::task_need_retry;
+            }
+        }
         SWSS_LOG_INFO("Recalculate shared buffer pool size due to over subscribe ratio has been updated from %s to %s",
                       m_overSubscribeRatio.c_str(), newRatio.c_str());
         m_overSubscribeRatio = newRatio;
@@ -1968,6 +1976,24 @@ task_process_status BufferMgrDynamic::handleDefaultLossLessBufferParam(KeyOpFiel
 
     return task_process_status::task_success;
 }
+bool BufferMgrDynamic::isSharedHeadroomPoolEnabledInSai()
+{
+    string xoff;
+    recalculateSharedBufferPool();
+    if (!isNonZero(m_bufferPoolLookup[INGRESS_LOSSLESS_PG_POOL_NAME].xoff))
+    {
+        return true;
+    }
+    m_applBufferPoolTable.flush();
+    m_applStateBufferPoolTable.hget(INGRESS_LOSSLESS_PG_POOL_NAME, "xoff", xoff);
+    if (!isNonZero(xoff))
+    {
+        SWSS_LOG_INFO("Shared headroom pool is enabled but has not been applied to SAI, retrying");
+        return false;
+    }
+
+    return true;
+}
 
 task_process_status BufferMgrDynamic::handleCableLenTable(KeyOpFieldsValuesTuple &tuple)
 {
@@ -2416,6 +2442,14 @@ task_process_status BufferMgrDynamic::handleBufferPoolTable(KeyOpFieldsValuesTup
             {
                 bool isSHPEnabledBySize = isNonZero(m_configuredSharedHeadroomPoolSize);
 
+                if (m_portInitDone && (!isSHPEnabledBySize) && willSHPBeEnabledBySize)
+                {
+                    if (!isSharedHeadroomPoolEnabledInSai())
+                    {
+                        return task_process_status::task_need_retry;
+                    }
+                }
+
                 m_configuredSharedHeadroomPoolSize = newSHPSize;
                 refreshSharedHeadroomPool(false, isSHPEnabledBySize != willSHPBeEnabledBySize);
             }

cfgmgr/buffermgrdyn.h

@@ -147,7 +147,7 @@ typedef std::map<std::string, std::string> gearbox_delay_t;
 class BufferMgrDynamic : public Orch
 {
 public:
-    BufferMgrDynamic(DBConnector *cfgDb, DBConnector *stateDb, DBConnector *applDb, const std::vector<TableConnector> &tables, std::shared_ptr<std::vector<KeyOpFieldsValuesTuple>> gearboxInfo, std::shared_ptr<std::vector<KeyOpFieldsValuesTuple>> zeroProfilesInfo);
+    BufferMgrDynamic(DBConnector *cfgDb, DBConnector *stateDb, DBConnector *applDb, DBConnector *applStateDb, const std::vector<TableConnector> &tables, std::shared_ptr<std::vector<KeyOpFieldsValuesTuple>> gearboxInfo, std::shared_ptr<std::vector<KeyOpFieldsValuesTuple>> zeroProfilesInfo);
     using Orch::doTask;
 
 private:
@@ -204,6 +204,7 @@ class BufferMgrDynamic : public Orch
 
     // BUFFER_POOL table and cache
     ProducerStateTable m_applBufferPoolTable;
+    Table m_applStateBufferPoolTable;
     Table m_stateBufferPoolTable;
     buffer_pool_lookup_t m_bufferPoolLookup;
 
@@ -294,6 +295,7 @@ class BufferMgrDynamic : public Orch
     task_process_status allocateProfile(const std::string &speed, const std::string &cable, const std::string &mtu, const std::string &threshold, const std::string &gearbox_model, long lane_count, std::string &profile_name);
     void releaseProfile(const std::string &profile_name);
     bool isHeadroomResourceValid(const std::string &port, const buffer_profile_t &profile, const std::string &new_pg);
+    bool isSharedHeadroomPoolEnabledInSai();
     void refreshSharedHeadroomPool(bool enable_state_updated_by_ratio, bool enable_state_updated_by_size);
     task_process_status checkBufferProfileDirection(const std::string &profiles, buffer_direction_t dir);
     std::string constructZeroProfileListFromNormalProfileList(const std::string &normalProfileList, const std::string &port);

orchagent/bufferorch.cpp

@@ -323,6 +323,7 @@ task_process_status BufferOrch::processBufferPool(KeyOpFieldsValuesTuple &tuple)
     string map_type_name = APP_BUFFER_POOL_TABLE_NAME;
     string object_name = kfvKey(tuple);
     string op = kfvOp(tuple);
+    string xoff;
 
     SWSS_LOG_DEBUG("object name:%s", object_name.c_str());
     if (m_buffer_type_maps[map_type_name]->find(object_name) != m_buffer_type_maps[map_type_name]->end())
@@ -417,6 +418,7 @@ task_process_status BufferOrch::processBufferPool(KeyOpFieldsValuesTuple &tuple)
                 attr.value.u64 = (uint64_t)stoul(value);
                 attr.id = SAI_BUFFER_POOL_ATTR_XOFF_SIZE;
                 attribs.push_back(attr);
+                xoff = value;
             }
             else
             {
@@ -469,6 +471,15 @@ task_process_status BufferOrch::processBufferPool(KeyOpFieldsValuesTuple &tuple)
             // "FLEX_COUNTER_STATUS"
             m_countersDb->hset(COUNTERS_BUFFER_POOL_NAME_MAP, object_name, sai_serialize_object_id(sai_object));
         }
+
+        // Only publish the result when shared headroom pool is enabled and it has been successfully applied to SAI
+        if (!xoff.empty())
+        {
+            vector<FieldValueTuple> fvs;
+            fvs.emplace_back("xoff", xoff);
+            SWSS_LOG_INFO("Publishing the result after applying the shared headroom pool size %s to SAI", xoff.c_str());
+            m_publisher.publish(APP_BUFFER_POOL_TABLE_NAME, object_name, fvs, ReturnCode(SAI_STATUS_SUCCESS), true);
+        }
     }
     else if (op == DEL_COMMAND)
     {
@@ -499,6 +510,9 @@ task_process_status BufferOrch::processBufferPool(KeyOpFieldsValuesTuple &tuple)
         auto it_to_delete = (m_buffer_type_maps[map_type_name])->find(object_name);
         (m_buffer_type_maps[map_type_name])->erase(it_to_delete);
         m_countersDb->hdel(COUNTERS_BUFFER_POOL_NAME_MAP, object_name);
+
+        vector<FieldValueTuple> fvs;
+        m_publisher.publish(APP_BUFFER_POOL_TABLE_NAME, object_name, fvs, ReturnCode(SAI_STATUS_SUCCESS), true);
     }
     else
     {

tests/mock_tests/Makefile.am

@@ -172,7 +172,7 @@ tests_intfmgrd_INCLUDES = $(tests_INCLUDES) -I$(top_srcdir)/cfgmgr -I$(top_srcdi
 tests_intfmgrd_CFLAGS = $(DBGFLAGS) $(AM_CFLAGS) $(CFLAGS_COMMON) $(CFLAGS_GTEST) $(CFLAGS_SAI)
 tests_intfmgrd_CPPFLAGS = $(DBGFLAGS) $(AM_CFLAGS) $(CFLAGS_COMMON) $(CFLAGS_GTEST) $(CFLAGS_SAI) $(tests_intfmgrd_INCLUDES)
 tests_intfmgrd_LDADD = $(LDADD_GTEST) $(LDADD_SAI) -lnl-genl-3 -lhiredis -lhiredis \
-        -lswsscommon -lswsscommon -lgtest -lgtest_main -lzmq -lnl-3 -lnl-route-3 -lpthread
+        -lswsscommon -lswsscommon -lgtest -lgtest_main -lzmq -lnl-3 -lnl-route-3 -lpthread -lgmock -lgmock_main
 
 ## fpmsyncd unit tests
 

tests/mock_tests/buffermgrdyn_ut.cpp

@@ -22,6 +22,7 @@ namespace buffermgrdyn_test
     shared_ptr<swss::DBConnector> m_app_db = make_shared<swss::DBConnector>("APPL_DB", 0);
     shared_ptr<swss::DBConnector> m_config_db = make_shared<swss::DBConnector>("CONFIG_DB", 0);
     shared_ptr<swss::DBConnector> m_state_db = make_shared<swss::DBConnector>("STATE_DB", 0);
+    shared_ptr<swss::DBConnector> m_app_state_db = make_shared<swss::DBConnector>("APPL_STATE_DB", 0);
 
     BufferMgrDynamic *m_dynamicBuffer;
     SelectableTimer m_selectableTable(timespec({ .tv_sec = BUFFERMGR_TIMER_PERIOD, .tv_nsec = 0 }), 0);
@@ -180,7 +181,7 @@ namespace buffermgrdyn_test
                 TableConnector(m_state_db.get(), STATE_PORT_TABLE_NAME)
             };
 
-            m_dynamicBuffer = new BufferMgrDynamic(m_config_db.get(), m_state_db.get(), m_app_db.get(), buffer_table_connectors, nullptr, zero_profile);
+            m_dynamicBuffer = new BufferMgrDynamic(m_config_db.get(), m_state_db.get(), m_app_db.get(), m_app_state_db.get(), buffer_table_connectors, nullptr, zero_profile);
         }
 
         void InitPort(const string &port="Ethernet0", const string &admin_status="up")

tests/mock_tests/bufferorch_ut.cpp

@@ -7,9 +7,11 @@
 #include "ut_helper.h"
 #include "mock_orchagent_main.h"
 #include "mock_table.h"
+#include "mock_response_publisher.h"
 
 extern string gMySwitchType;
 
+extern std::unique_ptr<MockResponsePublisher> gMockResponsePublisher;
 
 namespace bufferorch_test
 {
@@ -19,6 +21,7 @@ namespace bufferorch_test
     sai_port_api_t *pold_sai_port_api;
 
     shared_ptr<swss::DBConnector> m_app_db;
+    shared_ptr<swss::DBConnector> m_app_state_db;
     shared_ptr<swss::DBConnector> m_config_db;
     shared_ptr<swss::DBConnector> m_state_db;
     shared_ptr<swss::DBConnector> m_chassis_app_db;
@@ -113,6 +116,7 @@ namespace bufferorch_test
             m_app_db = make_shared<swss::DBConnector>("APPL_DB", 0);
             m_config_db = make_shared<swss::DBConnector>("CONFIG_DB", 0);
             m_state_db = make_shared<swss::DBConnector>("STATE_DB", 0);
+            m_app_state_db = make_shared<swss::DBConnector>("APPL_STATE_DB", 0);
             if(gMySwitchType == "voq")
                 m_chassis_app_db = make_shared<swss::DBConnector>("CHASSIS_APP_DB", 0);
 
@@ -317,6 +321,24 @@ namespace bufferorch_test
         }
     };
 
+    TEST_F(BufferOrchTest, BufferOrchTestSharedHeadroomPool)
+    {
+        gMockResponsePublisher = std::make_unique<MockResponsePublisher>();
+
+        Table bufferPoolTable = Table(m_app_db.get(), APP_BUFFER_POOL_TABLE_NAME);
+        Table bufferPoolStateTable = Table(m_app_state_db.get(), APP_BUFFER_POOL_TABLE_NAME);
+
+        bufferPoolTable.set("ingress_lossless_pool",
+                            {
+                                {"xoff", "10240"}
+                            });
+        gBufferOrch->addExistingData(&bufferPoolTable);
+        EXPECT_CALL(*gMockResponsePublisher, publish(APP_BUFFER_POOL_TABLE_NAME, "ingress_lossless_pool", std::vector<FieldValueTuple>{{"xoff", "10240"}}, ReturnCode(SAI_STATUS_SUCCESS), true)).Times(1);
+        static_cast<Orch *>(gBufferOrch)->doTask();
+
+        gMockResponsePublisher.reset();
+    }
+
     TEST_F(BufferOrchTest, BufferOrchTestBufferPgReferencingObjRemoveThenAdd)
     {
         vector<string> ts;

tests/mock_tests/fake_response_publisher.cpp

@@ -2,19 +2,36 @@
 #include <vector>
 
 #include "response_publisher.h"
+#include "mock_response_publisher.h"
+
+/* This mock plugs into this fake response publisher implementation
+ * when needed to test code that uses response publisher. */
+std::unique_ptr<MockResponsePublisher> gMockResponsePublisher;
 
 ResponsePublisher::ResponsePublisher() : m_db("APPL_STATE_DB", 0) {}
 
 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) {}
+    const std::vector<swss::FieldValueTuple>& state_attrs, bool replace)
+{
+    if (gMockResponsePublisher)
+    {
+        gMockResponsePublisher->publish(table, key, intent_attrs, status, state_attrs, replace);
+    }
+}
 
 void ResponsePublisher::publish(
     const std::string& table, const std::string& key,
     const std::vector<swss::FieldValueTuple>& intent_attrs,
-    const ReturnCode& status, bool replace) {}
+    const ReturnCode& status, bool replace)
+{
+    if (gMockResponsePublisher)
+    {
+        gMockResponsePublisher->publish(table, key, intent_attrs, status, replace);
+    }
+}
 
 void ResponsePublisher::writeToDB(
     const std::string& table, const std::string& key,