Fix incorret table data issue with consecutive producerState del/set … (sonic-net#240)

* Fix incorret table data issue with consecutive producerState del/set operations
* Use _DEL_SET suffix for set of deleted keys in producerStateTable/consumerStateTable

Author: jipanyang

common/consumer_state_table_pops.lua

@@ -6,14 +6,18 @@ local keys = redis.call('SPOP', KEYS[1], ARGV[1])
 local n = table.getn(keys)
 for i = 1, n do
    local key = keys[i]
+   -- Check if there was request to delete the key, clear it in table first
+   local num = redis.call('SREM', KEYS[3], key)
+   if num == 1 then
+      redis.call('DEL', tablename..key)
+   end
+   -- Push the new set of field/value for this key in table
    local fieldvalues = redis.call('HGETALL', stateprefix..tablename..key)
    table.insert(ret, {key, fieldvalues})
    for i = 1, #fieldvalues, 2 do
       redis.call('HSET', tablename..key, fieldvalues[i], fieldvalues[i + 1])
    end
+   -- Clean up the key in temporary state table
    redis.call('DEL', stateprefix..tablename..key)
-   if #fieldvalues == 0 then
-      redis.call('DEL', tablename..key)
-   end
 end
 return ret

common/consumerstatetable.cpp

@@ -38,10 +38,11 @@ void ConsumerStateTable::pops(std::deque<KeyOpFieldsValuesTuple> &vkco, const st
 
     RedisCommand command;
     command.format(
-        "EVALSHA %s 2 %s %s: %d %s",
+        "EVALSHA %s 3 %s %s: %s %d %s",
         sha.c_str(),
         getKeySetName().c_str(),
         getTableName().c_str(),
+        getDelKeySetName().c_str(),
         POP_BATCH_SIZE,
         getStateHashPrefix().c_str());
 

common/producerstatetable.cpp

@@ -36,13 +36,15 @@ ProducerStateTable::ProducerStateTable(RedisPipeline *pipeline, const string &ta
 
     string luaDel =
         "redis.call('SADD', KEYS[2], ARGV[2])\n"
+        "redis.call('SADD', KEYS[4], ARGV[2])\n"
         "redis.call('DEL', KEYS[3])\n"
         "redis.call('PUBLISH', KEYS[1], ARGV[1])\n";
     m_shaDel = m_pipe->loadRedisScript(luaDel);
 
     string luaClear =
         "redis.call('DEL', KEYS[1])\n"
-        "redis.call('DEL', KEYS[2])\n";
+        "redis.call('DEL', KEYS[2])\n"
+        "redis.call('DEL', KEYS[3])\n";
     m_shaClear = m_pipe->loadRedisScript(luaClear);
 }
 
@@ -100,13 +102,15 @@ void ProducerStateTable::del(const string &key, const string &op /*= DEL_COMMAND
     vector<string> args;
     args.emplace_back("EVALSHA");
     args.emplace_back(m_shaDel);
-    args.emplace_back("3");
+    args.emplace_back("4");
     args.emplace_back(getChannelName());
     args.emplace_back(getKeySetName());
     args.emplace_back(getStateHashPrefix() + getKeyName(key));
+    args.emplace_back(getDelKeySetName());
     args.emplace_back("G");
     args.emplace_back(key);
     args.emplace_back("''");
+    args.emplace_back("''");
 
     // Transform data structure
     vector<const char *> args1;
@@ -145,9 +149,10 @@ void ProducerStateTable::clear()
     vector<string> args;
     args.emplace_back("EVALSHA");
     args.emplace_back(m_shaClear);
-    args.emplace_back("2");
+    args.emplace_back("3");
     args.emplace_back(getKeySetName());
     args.emplace_back(getStateHashPrefix() + getTableName());
+    args.emplace_back(getDelKeySetName());
 
     // Transform data structure
     vector<const char *> args1;

common/table.h

@@ -194,13 +194,16 @@ class TableName_KeyValueOpQueues {
 class TableName_KeySet {
 private:
     std::string m_key;
+    std::string m_delkey;
 public:
     TableName_KeySet(const std::string &tableName)
         : m_key(tableName + "_KEY_SET")
+        , m_delkey(tableName + "_DEL_SET")
     {
     }
 
     std::string getKeySetName() const { return m_key; }
+    std::string getDelKeySetName() const { return m_delkey; }
     std::string getStateHashPrefix() const { return "_"; }
 };
 

tests/redis_state_ut.cpp

@@ -374,6 +374,134 @@ TEST(ConsumerStateTable, set_del_set)
     EXPECT_EQ(qlen, 0U);
 }
 
+TEST(ConsumerStateTable, set_pop_del_set_pop_get)
+{
+    clearDB();
+
+    /* Prepare producer */
+    int index = 0;
+    string tableName = "UT_REDIS_THREAD_" + to_string(index);
+    DBConnector db(TEST_DB, "localhost", 6379, 0);
+    ProducerStateTable p(&db, tableName);
+    string key = "TheKey";
+    int maxNumOfFields = 2;
+
+    /* First set operation */
+    {
+        vector<FieldValueTuple> fields;
+        for (int j = 0; j < maxNumOfFields; j++)
+        {
+            FieldValueTuple t(field(j), value(j));
+            fields.push_back(t);
+        }
+        p.set(key, fields);
+    }
+
+    /* Prepare consumer */
+    ConsumerStateTable c(&db, tableName);
+    Select cs;
+    Selectable *selectcs;
+    cs.addSelectable(&c);
+
+    /* First pop operation */
+    {
+        int ret = cs.select(&selectcs);
+        EXPECT_EQ(ret, Select::OBJECT);
+        KeyOpFieldsValuesTuple kco;
+        c.pop(kco);
+        EXPECT_EQ(kfvKey(kco), key);
+        EXPECT_EQ(kfvOp(kco), "SET");
+
+        auto fvs = kfvFieldsValues(kco);
+        EXPECT_EQ(fvs.size(), (unsigned int)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(j)], value(j));
+        }
+    }
+
+    /* Del operation and second set operation will be merged */
+
+    /* Del operation */
+    p.del(key);
+
+    /* Second set operation */
+    {
+        vector<FieldValueTuple> fields;
+        for (int j = 0; j < maxNumOfFields * 2; j += 2)
+        {
+            FieldValueTuple t(field(j), value(j));
+            fields.push_back(t);
+        }
+        p.set(key, fields);
+    }
+
+    /* Second pop operation */
+    {
+        int ret = cs.select(&selectcs);
+        EXPECT_EQ(ret, Select::OBJECT);
+        KeyOpFieldsValuesTuple kco;
+        c.pop(kco);
+        EXPECT_EQ(kfvKey(kco), key);
+        EXPECT_EQ(kfvOp(kco), "SET");
+
+        auto fvs = kfvFieldsValues(kco);
+
+        /* size of fvs should be maxNumOfFields, no "field 1" left from first set*/
+        EXPECT_EQ(fvs.size(), (unsigned int)maxNumOfFields);
+
+        map<string, string> mm;
+        for (auto fv: fvs)
+        {
+            mm[fvField(fv)] = fvValue(fv);
+        }
+
+        for (int j = 0; j < maxNumOfFields * 2; j += 2)
+        {
+            EXPECT_EQ(mm[field(j)], value(j));
+        }
+    }
+
+    /* Get data directly from table in redis DB*/
+    Table t(&db, tableName);
+    vector<FieldValueTuple> values;
+    t.get(key, values);
+    /* size of values should be maxNumOfFields, no "field 1" left from first set */
+    EXPECT_EQ(values.size(), (unsigned int)maxNumOfFields);
+
+    /*
+     * Third pop operation, consumer received two consectuive signals.
+     * data depleted upon first one
+     */
+    {
+        int ret = cs.select(&selectcs, 1000);
+        EXPECT_EQ(ret, Select::OBJECT);
+        KeyOpFieldsValuesTuple kco;
+        c.pop(kco);
+        EXPECT_EQ(kfvKey(kco), "");
+    }
+
+    /* Third select operation */
+    {
+        int ret = cs.select(&selectcs, 1000);
+        EXPECT_EQ(ret, Select::TIMEOUT);
+    }
+
+    /* State Queue should be empty */
+    RedisCommand keys;
+    keys.format("KEYS %s*", (c.getStateHashPrefix() + tableName).c_str());
+    RedisReply r(&db, keys, REDIS_REPLY_ARRAY);
+    auto qlen = r.getContext()->elements;
+    EXPECT_EQ(qlen, 0U);
+}
+
 TEST(ConsumerStateTable, singlethread)
 {
     clearDB();