From 139919d5e91eadc89eff5a5e33374d76fe558982 Mon Sep 17 00:00:00 2001 From: Runming Wu Date: Mon, 1 May 2023 19:04:49 -0700 Subject: [PATCH] Add update_only in SubscriberStateTable. If the flag is set, the API will return empty attribute (instead of full list of attributes) when there is no change for SET requests. This will be useful when CONFIG DB is updated with no actual change. After setting the flag, the user of the SubscriberStateTable will do no-op when empty attribute is return. --- common/subscriberstatetable.cpp | 34 ++++- common/subscriberstatetable.h | 9 +- tests/redis_subscriber_state_ut.cpp | 190 ++++++++++++++++++++++++++++ 3 files changed, 230 insertions(+), 3 deletions(-) diff --git a/common/subscriberstatetable.cpp b/common/subscriberstatetable.cpp index b80f24ef5..315b72165 100644 --- a/common/subscriberstatetable.cpp +++ b/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 &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 &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{}; + } + } kfvKey(kco) = key; kfvOp(kco) = SET_COMMAND; } diff --git a/common/subscriberstatetable.h b/common/subscriberstatetable.h index 432dc90fb..496cdee99 100644 --- a/common/subscriberstatetable.h +++ b/common/subscriberstatetable.h @@ -3,6 +3,7 @@ #include #include #include +#include #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 &vkco, const std::string &prefix = EMPTY_PREFIX); @@ -33,6 +34,12 @@ class SubscriberStateTable : public ConsumerTableBase std::deque> 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> m_cache; }; } diff --git a/tests/redis_subscriber_state_ut.cpp b/tests/redis_subscriber_state_ut.cpp index 34db9d48a..512e29e17 100644 --- a/tests/redis_subscriber_state_ut.cpp +++ b/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 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 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 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 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 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 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 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 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 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 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(); + } +} \ No newline at end of file