Improve hset and hdel performance with RedisPipeline.

common/Makefile.am

@@ -7,7 +7,6 @@ EXTRA_DIST = \
     consumer_table_pops.lua \
     producer_state_table_apply_view.lua \
     table_dump.lua \
-    redis_multi.lua \
     portcounter.lua \
     fdb_flush.lua \
     fdb_flush.v2.lua

common/configdb.cpp

@@ -365,9 +365,9 @@ void ConfigDBPipeConnector_Native::_set_entry(RedisTransactioner& pipe, std::str
     {
         auto original = get_entry(table, key);
 
-        RedisCommand shmset;
-        shmset.formatHMSET(_hash, data.begin(), data.end());
-        pipe.enqueue(shmset, REDIS_REPLY_STATUS);
+        RedisCommand shset;
+        shset.formatHSET(_hash, data.begin(), data.end());
+        pipe.enqueue(shset, REDIS_REPLY_INTEGER);
 
         for (auto& it: original)
         {
@@ -403,9 +403,9 @@ void ConfigDBPipeConnector_Native::_mod_entry(RedisTransactioner& pipe, string t
     }
     else
     {
-        RedisCommand shmset;
-        shmset.formatHMSET(_hash, data.begin(), data.end());
-        pipe.enqueue(shmset, REDIS_REPLY_STATUS);
+        RedisCommand shset;
+        shset.formatHSET(_hash, data.begin(), data.end());
+        pipe.enqueue(shset, REDIS_REPLY_INTEGER);
     }
 }
 // Write multiple tables into config db.

common/dbconnector.cpp

@@ -10,7 +10,7 @@
 
 #include "common/dbconnector.h"
 #include "common/redisreply.h"
-#include "common/redisapi.h"
+#include "common/redispipeline.h"
 #include "common/pubsub.h"
 
 using json = nlohmann::json;
@@ -880,55 +880,28 @@ void DBConnector::hmset(const std::unordered_map<std::string, std::vector<std::p
 {
     SWSS_LOG_ENTER();
 
-    // make sure this will be object (not null) when multi hash is empty
-    json j = json::object();
-
-    // pack multi hash to json (takes bout 70 ms for 10k to construct)
-    for (const auto& kvp: multiHash)
+    RedisPipeline pipe(this);
+    for (auto& hash : multiHash)
     {
-        json o;
-
-        for (const auto &item: kvp.second)
-        {
-            o[std::get<0>(item)] = std::get<1>(item);
-        }
-
-        j[kvp.first] = o;
+        RedisCommand hset;
+        hset.formatHSET(hash.first, hash.second.begin(), hash.second.end());
+        pipe.push(hset, REDIS_REPLY_INTEGER);
     }
 
-    std::string strJson = j.dump();
-
-    lazyLoadRedisScriptFile(this, "redis_multi.lua", m_shaRedisMulti);
-    RedisCommand command;
-    command.format(
-        "EVALSHA %s 1 %s %s",
-        m_shaRedisMulti.c_str(),
-        strJson.c_str(),
-        "mhset");
-
-    RedisReply r(this, command, REDIS_REPLY_NIL);
+    pipe.flush();
 }
 
 void DBConnector::del(const std::vector<std::string>& keys)
 {
     SWSS_LOG_ENTER();
 
-    json j = json::array();
-
-    for (const auto& key: keys)
+    RedisPipeline pipe(this);
+    for (auto& key : keys)
     {
-        j.push_back(key);
+        RedisCommand del;
+        del.formatDEL(key);
+        pipe.push(del, REDIS_REPLY_INTEGER);
     }
 
-    std::string strJson = j.dump();
-
-    lazyLoadRedisScriptFile(this, "redis_multi.lua", m_shaRedisMulti);
-    RedisCommand command;
-    command.format(
-        "EVALSHA %s 1 %s %s",
-        m_shaRedisMulti.c_str(),
-        strJson.c_str(),
-        "mdel");
-
-    RedisReply r(this, command, REDIS_REPLY_NIL);
+    pipe.flush();
 }

common/dbconnector.h

@@ -251,8 +251,6 @@ class DBConnector : public RedisContext
     int m_dbId;
     std::string m_dbName;
     std::string m_namespace;
-
-    std::string m_shaRedisMulti;
 };
 
 template <typename ReturnType>
@@ -284,9 +282,9 @@ void DBConnector::hgetall(const std::string &key, OutputIterator result)
 template<typename InputIterator>
 void DBConnector::hmset(const std::string &key, InputIterator start, InputIterator stop)
 {
-    RedisCommand shmset;
-    shmset.formatHMSET(key, start, stop);
-    RedisReply r(this, shmset, REDIS_REPLY_STATUS);
+    RedisCommand shset;
+    shset.formatHSET(key, start, stop);
+    RedisReply r(this, shset, REDIS_REPLY_INTEGER);
 }
 
 }

common/rediscommand.cpp

@@ -59,11 +59,17 @@ void RedisCommand::format(const vector<string> &commands)
     formatArgv(static_cast<int>(args.size()), args.data(), NULL);
 }
 
-/* Format HMSET key multiple field value command */
-void RedisCommand::formatHMSET(const std::string &key,
+/* Format HSET key multiple field value command */
+void RedisCommand::formatHSET(const std::string &key,
                                const std::vector<FieldValueTuple> &values)
 {
-    formatHMSET(key, values.begin(), values.end());
+    formatHSET(key, values.begin(), values.end());
+}
+
+void RedisCommand::formatHSET(const std::string &key,
+                              const std::map<std::string, std::string> &values)
+{
+    formatHSET(key, values.begin(), values.end());
 }
 
 /* Format HSET key field value command */
@@ -110,6 +116,12 @@ void RedisCommand::formatTTL(const std::string& key)
     return format("TTL %s", key.c_str());
 }
 
+/* Format DEL key command */
+void RedisCommand::formatDEL(const std::string& key)
+{
+    return format("DEL %s", key.c_str());
+}
+
 const char *RedisCommand::c_str() const
 {
     return temp;

common/rediscommand.h

@@ -5,6 +5,7 @@
 #include <vector>
 #include <string>
 #include <stdexcept>
+#include <map>
 
 namespace swss {
 
@@ -31,11 +32,14 @@ class RedisCommand {
 #ifndef SWIG
     __attribute__((deprecated))
 #endif
-    void formatHMSET(const std::string &key,
+    void formatHSET(const std::string &key,
                      const std::vector<FieldValueTuple> &values);
 
+    void formatHSET(const std::string &key,
+                     const std::map<std::string, std::string> &values);
+
     template<typename InputIterator>
-    void formatHMSET(const std::string &key,
+    void formatHSET(const std::string &key,
                      InputIterator start, InputIterator stop);
 
     /* Format HSET key field value command */
@@ -57,6 +61,9 @@ class RedisCommand {
     /* Format TTL key command */
     void formatTTL(const std::string& key);
 
+    /* Format DEL key command */
+    void formatDEL(const std::string& key);
+
     const char *c_str() const;
 
     size_t length() const;
@@ -66,12 +73,12 @@ class RedisCommand {
 };
 
 template<typename InputIterator>
-void RedisCommand::formatHMSET(const std::string &key,
+void RedisCommand::formatHSET(const std::string &key,
                  InputIterator start, InputIterator stop)
 {
     if (start == stop) throw std::invalid_argument("empty values");
 
-    const char* cmd = "HMSET";
+    const char* cmd = "HSET";
 
     std::vector<const char*> args = { cmd, key.c_str() };
 

common/redispipeline.h

@@ -1,6 +1,7 @@
 #pragma once
 
 #include <string>
+#include <queue>
 #include <functional>
 #include "redisreply.h"
 #include "rediscommand.h"

common/table.cpp

@@ -140,8 +140,8 @@ void Table::set(const string &key, const vector<FieldValueTuple> &values,
 
     RedisCommand cmd;
 
-    cmd.formatHMSET(getKeyName(key), values.begin(), values.end());
-    m_pipe->push(cmd, REDIS_REPLY_STATUS);
+    cmd.formatHSET(getKeyName(key), values.begin(), values.end());
+    m_pipe->push(cmd, REDIS_REPLY_INTEGER);
 
     if (ttl != DEFAULT_DB_TTL)
     {

tests/redis_ut.cpp

@@ -477,6 +477,58 @@ TEST(DBConnector, RedisClient)
     cout << "Done." << endl;
 }
 
+TEST(DBConnector, HmsetAndDel)
+{
+    DBConnector db("TEST_DB", 0, true);
+    clearDB();
+
+    unordered_map<string, vector<pair<string, string>>> multiHash;
+    vector<FieldValueTuple> values;
+    values.push_back(make_pair("field_1", "1"));
+    values.push_back(make_pair("field_2", "2"));
+
+    vector<string> keys;
+    for (int idx =0; idx<10; idx++)
+    {
+        string key = "hash_key_" + to_string(idx);
+        multiHash[key] = values;
+        keys.push_back(key);
+    }
+
+    // set multiple hash with hmset
+    db.hmset(multiHash);
+
+    // check all key exist
+    for (auto& key : keys)
+    {
+        auto fvs = db.hgetall(key);
+        EXPECT_EQ(fvs.size(), 2);
+
+        for (auto fv: fvs)
+        {
+            string value_1 = "1", value_2 = "2";
+            if (fvField(fv) == "field_1")
+            {
+                EXPECT_EQ(fvValue(fv), value_1);
+            }
+            if (fvField(fv) == "field_2")
+            {
+                EXPECT_EQ(fvValue(fv), value_2);
+            }
+        }
+    }
+
+    // delete multiple hash with del
+    db.del(keys);
+
+    // check all key deleted
+    for (auto& key : keys)
+    {
+        auto fvs = db.hgetall(key);
+        EXPECT_EQ(fvs.size(), 0);
+    }
+}
+
 TEST(DBConnector, test)
 {
     thread *producerThreads[NUMBER_OF_THREADS];