Add update_only in SubscriberStateTable

.azure-pipelines/build-swss-template.yml

@@ -136,6 +136,7 @@ jobs:
   - script: |
       set -ex
       rm ../*.deb || true
+      sudo sed -i 's/SubscriberStateTable::SubscriberStateTable.*/SubscriberStateTable::SubscriberStateTable(DBConnector *db, const std::string \&tableName, int popBatchSize, int pri, bool update_only)/g' orchagent/p4orch/tests/fake_subscriberstatetable.cpp
       ./autogen.sh
       dpkg-buildpackage -us -uc -b -j$(nproc)
       mv ../*.deb $(Build.ArtifactStagingDirectory)

common/subscriberstatetable.cpp

@@ -14,8 +14,8 @@ using namespace std;
 
 namespace swss {
 
-SubscriberStateTable::SubscriberStateTable(DBConnector *db, const string &tableName, int popBatchSize, int pri)
-    : ConsumerTableBase(db, tableName, popBatchSize, pri), m_table(db, tableName)
+SubscriberStateTable::SubscriberStateTable(DBConnector *db, const string &tableName, int popBatchSize, int pri, bool update_only)
+    : ConsumerTableBase(db, tableName, popBatchSize, pri), m_table(db, tableName), m_update_only(update_only)
 {
     m_keyspace = "__keyspace@";
 
@@ -38,6 +38,14 @@ SubscriberStateTable::SubscriberStateTable(DBConnector *db, const string &tableN
             continue;
         }
 
+        if (m_update_only)
+        {
+            for (auto fv : kfvFieldsValues(kco))
+            {
+                m_cache[key][fvField(fv)] = fvValue(fv);
+            }
+        }
+
         m_buffer.push_back(kco);
     }
 }
@@ -144,6 +152,10 @@ void SubscriberStateTable::pops(deque<KeyOpFieldsValuesTuple> &vkco, const strin
         {
             kfvKey(kco) = key;
             kfvOp(kco) = DEL_COMMAND;
+            if (m_update_only)
+            {
+                m_cache.erase(key);
+            }
         }
         else
         {
@@ -152,6 +164,24 @@ void SubscriberStateTable::pops(deque<KeyOpFieldsValuesTuple> &vkco, const strin
                 SWSS_LOG_NOTICE("Miss table key %s, possibly outdated", table_entry.c_str());
                 continue;
             }
+            if (m_update_only)
+            {
+                bool update = false;
+                for (auto fv : kfvFieldsValues(kco))
+                {
+                    if (m_cache.find(key) == m_cache.end() ||
+                        m_cache.at(key).find(fvField(fv)) == m_cache.at(key).end() ||
+                        m_cache.at(key).at(fvField(fv)) != fvValue(fv))
+                    {
+                        update = true;
+                        m_cache[key][fvField(fv)] = fvValue(fv);
+                    }
+                }
+                if (!update)
+                {
+                    kfvFieldsValues(kco) = std::vector<FieldValueTuple>{};
+                }
+            }
             kfvKey(kco) = key;
             kfvOp(kco) = SET_COMMAND;
         }

common/subscriberstatetable.h

@@ -3,6 +3,7 @@
 #include <string>
 #include <deque>
 #include <memory.h>
+#include <unordered_map>
 #include "dbconnector.h"
 #include "consumertablebase.h"
 
@@ -11,7 +12,7 @@ namespace swss {
 class SubscriberStateTable : public ConsumerTableBase
 {
 public:
-    SubscriberStateTable(DBConnector *db, const std::string &tableName, int popBatchSize = DEFAULT_POP_BATCH_SIZE, int pri = 0);
+    SubscriberStateTable(DBConnector *db, const std::string &tableName, int popBatchSize = DEFAULT_POP_BATCH_SIZE, int pri = 0, bool update_only = false);
 
     /* Get all elements available */
     void pops(std::deque<KeyOpFieldsValuesTuple> &vkco, const std::string &prefix = EMPTY_PREFIX);
@@ -33,6 +34,12 @@ class SubscriberStateTable : public ConsumerTableBase
 
     std::deque<std::shared_ptr<RedisReply>> m_keyspace_event_buffer;
     Table m_table;
+
+    /* If update_only is set to true, pop will return empty attribute instead of
+     * full list of attributes when there is no change for SET request. */
+    bool m_update_only;
+    /* Local cache used to check the updated attributes when m_update_only is set. */
+    std::unordered_map<std::string, std::unordered_map<std::string, std::string>> m_cache;
 };
 
 }

tests/redis_subscriber_state_ut.cpp

@@ -635,3 +635,193 @@ TEST(SubscriberStateTable, cachedData)
         EXPECT_TRUE(r);
     }
 }
+
+TEST(SubscriberStateTable, updateOnly)
+{
+    clearDB();
+
+    /* Prepare init data */
+    int index = 0;
+    int maxNumOfFields = 2;
+
+    DBConnector db("TEST_DB", 0, true);
+    Table p(&db, testTableName);
+    string key = "TheKey";
+    /* Set operation */
+    {
+        vector<FieldValueTuple> fields;
+        for (int j = 0; j < maxNumOfFields; j++)
+        {
+            FieldValueTuple t(field(index, j), value(index, j));
+            fields.push_back(t);
+        }
+        p.set(key, fields);
+    }
+
+    /* Prepare subscriber */
+    SubscriberStateTable c(&db, testTableName, TableConsumable::DEFAULT_POP_BATCH_SIZE, /*pri=*/0, /*update_only=*/true);
+    Select cs;
+    Selectable *selectcs;
+    cs.addSelectable(&c);
+    std::deque<KeyOpFieldsValuesTuple> entries;
+
+    /* Pop all the initial data */
+    {
+        c.pops(entries);
+        ASSERT_EQ(entries.size(), 1U);
+        KeyOpFieldsValuesTuple t = entries[0];
+        EXPECT_EQ(kfvKey(t), key);
+        EXPECT_EQ(kfvOp(t), SET_COMMAND);
+        auto fvs = kfvFieldsValues(t);
+        ASSERT_EQ(fvs.size(), (size_t)maxNumOfFields);
+
+        map<string, string> mm;
+        for (auto fv: fvs)
+        {
+            mm[fvField(fv)] = fvValue(fv);
+        }
+
+        for (int j = 0; j < maxNumOfFields; j++)
+        {
+            EXPECT_EQ(mm[field(index, j)], value(index, j));
+        }
+        entries.clear();
+    }
+
+    /* Add new attribute */
+    {
+        vector<FieldValueTuple> fields;
+        FieldValueTuple t(field(index, maxNumOfFields), value(index, maxNumOfFields));
+        fields.push_back(t);
+        p.set(key, fields);
+
+        int ret = cs.select(&selectcs);
+        EXPECT_EQ(ret, Select::OBJECT);
+        EXPECT_EQ(selectcs, &c);
+        c.pops(entries);
+        ASSERT_EQ(entries.size(), 1U);
+        KeyOpFieldsValuesTuple kco = entries[0];
+        EXPECT_EQ(kfvKey(kco), key);
+        EXPECT_EQ(kfvOp(kco), SET_COMMAND);
+        auto fvs = kfvFieldsValues(kco);
+        ASSERT_EQ(fvs.size(), (size_t)maxNumOfFields + 1);
+
+        map<string, string> mm;
+        for (auto fv: fvs)
+        {
+            mm[fvField(fv)] = fvValue(fv);
+        }
+
+        for (int j = 0; j <= maxNumOfFields; j++)
+        {
+            EXPECT_EQ(mm[field(index, j)], value(index, j));
+        }
+        entries.clear();
+    }
+
+    /* Update attribute */
+    {
+        vector<FieldValueTuple> fields;
+        FieldValueTuple t(field(index, 0), "new_value");
+        fields.push_back(t);
+        p.set(key, fields);
+
+        int ret = cs.select(&selectcs);
+        EXPECT_EQ(ret, Select::OBJECT);
+        EXPECT_EQ(selectcs, &c);
+        c.pops(entries);
+        ASSERT_EQ(entries.size(), 1U);
+        KeyOpFieldsValuesTuple kco = entries[0];
+        EXPECT_EQ(kfvKey(kco), key);
+        EXPECT_EQ(kfvOp(kco), SET_COMMAND);
+        auto fvs = kfvFieldsValues(kco);
+        ASSERT_EQ(fvs.size(), (size_t)maxNumOfFields + 1);
+
+        map<string, string> mm;
+        for (auto fv: fvs)
+        {
+            mm[fvField(fv)] = fvValue(fv);
+        }
+
+        for (int j = 1; j <= maxNumOfFields; j++)
+        {
+            EXPECT_EQ(mm[field(index, j)], value(index, j));
+        }
+        EXPECT_EQ(mm[field(index, 0)], "new_value");
+        entries.clear();
+    }
+
+    /* Delete key */
+    {
+        p.del(key);
+
+        int ret = cs.select(&selectcs);
+        EXPECT_EQ(ret, Select::OBJECT);
+        EXPECT_EQ(selectcs, &c);
+        c.pops(entries);
+        ASSERT_EQ(entries.size(), 1U);
+        KeyOpFieldsValuesTuple kco = entries[0];
+        EXPECT_EQ(kfvKey(kco), key);
+        EXPECT_EQ(kfvOp(kco), DEL_COMMAND);
+        auto fvs = kfvFieldsValues(kco);
+        ASSERT_EQ(fvs.size(), 0);
+        entries.clear();
+    }
+
+    /* Add new attribute */
+    {
+        vector<FieldValueTuple> fields;
+        for (int j = 0; j < maxNumOfFields; j++)
+        {
+            FieldValueTuple t(field(index, j), value(index, j));
+            fields.push_back(t);
+        }
+        p.set(key, fields);
+
+        int ret = cs.select(&selectcs);
+        EXPECT_EQ(ret, Select::OBJECT);
+        EXPECT_EQ(selectcs, &c);
+        c.pops(entries);
+        ASSERT_EQ(entries.size(), 1U);
+        KeyOpFieldsValuesTuple kco = entries[0];
+        EXPECT_EQ(kfvKey(kco), key);
+        EXPECT_EQ(kfvOp(kco), SET_COMMAND);
+        auto fvs = kfvFieldsValues(kco);
+        ASSERT_EQ(fvs.size(), (size_t)maxNumOfFields);
+
+        map<string, string> mm;
+        for (auto fv: fvs)
+        {
+            mm[fvField(fv)] = fvValue(fv);
+        }
+
+        for (int j = 0; j < maxNumOfFields; j++)
+        {
+            EXPECT_EQ(mm[field(index, j)], value(index, j));
+        }
+        entries.clear();
+    }
+
+    /* Update attribute with no change */
+    {
+        vector<FieldValueTuple> fields;
+        for (int j = 0; j < maxNumOfFields; j++)
+        {
+            FieldValueTuple t(field(index, j), value(index, j));
+            fields.push_back(t);
+        }
+        p.set(key, fields);
+
+        int ret = cs.select(&selectcs);
+        EXPECT_EQ(ret, Select::OBJECT);
+        EXPECT_EQ(selectcs, &c);
+        c.pops(entries);
+        ASSERT_EQ(entries.size(), 1U);
+        KeyOpFieldsValuesTuple kco = entries[0];
+        EXPECT_EQ(kfvKey(kco), key);
+        EXPECT_EQ(kfvOp(kco), SET_COMMAND);
+        auto fvs = kfvFieldsValues(kco);
+        ASSERT_EQ(fvs.size(), 0);
+        entries.clear();
+    }
+}