Producer/Consumer table binary support (#801)

Pterosaur · qiluo-msft · web-flow · commit 62e8d805a1e5 · 2023-07-13T12:55:11.000+08:00
We want to make the Producer/Consumer table can support binary messages, The native C string (char *) will be replaced to pointers and its lengths in all paths.
Meanwhile, the Python interfaces of SWIG can only handle the type, str, with UTF-8. So, we need to specialize the SWIG interfaces from bytes of Python to string of C++.
---------

Signed-off-by: Ze Gan &lt;ganze718@gmail.com&gt;
Co-authored-by: Qi Luo &lt;qiluo-msft@users.noreply.github.com&gt;
diff --git a/common/consumerstatetable.cpp b/common/consumerstatetable.cpp
@@ -68,16 +68,16 @@ void ConsumerStateTable::pops(std::deque<KeyOpFieldsValuesTuple> &vkco, const st
 
         auto& ctx = ctx0->element[ie];
         assert(ctx->element[0]->type == REDIS_REPLY_STRING);
-        std::string key = ctx->element[0]->str;
+        std::string key(ctx->element[0]->str, ctx->element[0]->len);
         kfvKey(kco) = key;
 
         assert(ctx->element[1]->type == REDIS_REPLY_ARRAY);
         auto ctx1 = ctx->element[1];
         for (size_t i = 0; i < ctx1->elements / 2; i++)
         {
             FieldValueTuple e;
-            fvField(e) = ctx1->element[i * 2]->str;
-            fvValue(e) = ctx1->element[i * 2 + 1]->str;
+            fvField(e).assign(ctx1->element[i * 2]->str, ctx1->element[i * 2]->len);
+            fvValue(e).assign(ctx1->element[i * 2 + 1]->str, ctx1->element[i * 2 + 1]->len);
             values.push_back(e);
         }
 
diff --git a/common/luatable.cpp b/common/luatable.cpp
@@ -54,13 +54,9 @@ bool LuaTable::get(const vector<string> &luaKeys, vector<FieldValueTuple> &value
         args.emplace_back(v);
     }
 
-    // Transform data structure
-    vector<const char *> args1;
-    transform(args.begin(), args.end(), back_inserter(args1), [](const string &s) { return s.c_str(); } );
-
     // Invoke redis command
     RedisCommand command;
-    command.formatArgv((int)args1.size(), &args1[0], NULL);
+    command.format(args);
     RedisReply r(m_db.get(), command, REDIS_REPLY_ARRAY);
     redisReply *reply = r.getContext();
 
@@ -109,13 +105,9 @@ bool LuaTable::hget(const vector<string> &luaKeys, const string &field, string &
         args.emplace_back(v);
     }
 
-    // Transform data structure
-    vector<const char *> args1;
-    transform(args.begin(), args.end(), back_inserter(args1), [](const string &s) { return s.c_str(); } );
-
     // Invoke redis command
     RedisCommand command;
-    command.formatArgv((int)args1.size(), &args1[0], NULL);
+    command.format(args);
     RedisReply r(m_db.get(), command);
     redisReply *reply = r.getContext();
 
diff --git a/common/producerstatetable.cpp b/common/producerstatetable.cpp
@@ -134,13 +134,9 @@ void ProducerStateTable::set(const string &key, const vector<FieldValueTuple> &v
         args.emplace_back(fvValue(iv));
     }
 
-    // Transform data structure
-    vector<const char *> args1;
-    transform(args.begin(), args.end(), back_inserter(args1), [](const string &s) { return s.c_str(); } );
-
     // Invoke redis command
     RedisCommand command;
-    command.formatArgv((int)args1.size(), &args1[0], NULL);
+    command.format(args);
     m_pipe->push(command, REDIS_REPLY_NIL);
     if (!m_buffered)
     {
@@ -171,13 +167,9 @@ void ProducerStateTable::del(const string &key, const string &op /*= DEL_COMMAND
     args.emplace_back("''");
     args.emplace_back("''");
 
-    // Transform data structure
-    vector<const char *> args1;
-    transform(args.begin(), args.end(), back_inserter(args1), [](const string &s) { return s.c_str(); } );
-
     // Invoke redis command
     RedisCommand command;
-    command.formatArgv((int)args1.size(), &args1[0], NULL);
+    command.format(args);
     m_pipe->push(command, REDIS_REPLY_NIL);
     if (!m_buffered)
     {
@@ -224,13 +216,9 @@ void ProducerStateTable::set(const std::vector<KeyOpFieldsValuesTuple>& values)
         }
     }
 
-    // Transform data structure
-    vector<const char *> args1;
-    transform(args.begin(), args.end(), back_inserter(args1), [](const string &s) { return s.c_str(); } );
-
     // Invoke redis command
     RedisCommand command;
-    command.formatArgv((int)args1.size(), &args1[0], NULL);
+    command.format(args);
     m_pipe->push(command, REDIS_REPLY_NIL);
     if (!m_buffered)
     {
@@ -265,13 +253,9 @@ void ProducerStateTable::del(const std::vector<std::string>& keys)
     }
     args.emplace_back("G");
 
-    // Transform data structure
-    vector<const char *> args1;
-    transform(args.begin(), args.end(), back_inserter(args1), [](const string &s) { return s.c_str(); } );
-
     // Invoke redis command
     RedisCommand command;
-    command.formatArgv((int)args1.size(), &args1[0], NULL);
+    command.format(args);
     m_pipe->push(command, REDIS_REPLY_NIL);
     if (!m_buffered)
     {
@@ -307,13 +291,9 @@ void ProducerStateTable::clear()
     args.emplace_back(getStateHashPrefix() + getTableName());
     args.emplace_back(getDelKeySetName());
 
-    // Transform data structure
-    vector<const char *> args1;
-    transform(args.begin(), args.end(), back_inserter(args1), [](const string &s) { return s.c_str(); } );
-
     // Invoke redis command
     RedisCommand cmd;
-    cmd.formatArgv((int)args1.size(), &args1[0], NULL);
+    cmd.format(args);
     m_pipe->push(cmd, REDIS_REPLY_NIL);
     m_pipe->flush();
 }
@@ -466,13 +446,9 @@ void ProducerStateTable::apply_temp_view()
         SWSS_LOG_DEBUG("apply_view.lua is called with following argument list: %s", ss.str().c_str());
     }
 
-    // Transform data structure
-    vector<const char *> args1;
-    transform(args.begin(), args.end(), back_inserter(args1), [](const string &s) { return s.c_str(); } );
-
     // Invoke redis command
     RedisCommand command;
-    command.formatArgv((int)args1.size(), &args1[0], NULL);
+    command.format(args);
     m_pipe->push(command, REDIS_REPLY_NIL);
     m_pipe->flush();
 
diff --git a/common/redisapi.h b/common/redisapi.h
@@ -83,16 +83,8 @@ static inline std::set<std::string> runRedisScript(RedisContext &ctx, const std:
     args.insert(args.end(), argv.begin(), argv.end());
     args.push_back("''");
 
-    // Convert to vector of char *
-    std::vector<const char *> c_args;
-    transform(
-            args.begin(),
-            args.end(),
-            std::back_inserter(c_args),
-            [](const std::string& s) { return s.c_str(); } );
-
     RedisCommand command;
-    command.formatArgv(static_cast<int>(c_args.size()), c_args.data(), NULL);
+    command.format(args);
 
     std::set<std::string> ret;
     try
diff --git a/common/rediscommand.cpp b/common/rediscommand.cpp
@@ -1,13 +1,15 @@
 #include <vector>
 #include <hiredis/hiredis.h>
 #include "rediscommand.h"
+#include "stringutility.h"
 
 using namespace std;
 
 namespace swss {
 
 RedisCommand::RedisCommand()
- : temp(NULL)
+ : temp(NULL),
+   len(0)
 {
 }
 
@@ -26,7 +28,7 @@ void RedisCommand::format(const char *fmt, ...)
 
     va_list ap;
     va_start(ap, fmt);
-    int len = redisvFormatCommand(&temp, fmt, ap);
+    len = redisvFormatCommand(&temp, fmt, ap);
     va_end(ap);
     if (len == -1) {
         throw std::bad_alloc();
@@ -43,7 +45,7 @@ void RedisCommand::formatArgv(int argc, const char **argv, const size_t *argvlen
         temp = nullptr;
     }
 
-    int len = redisFormatCommandArgv(&temp, argc, argv, argvlen);
+    len = redisFormatCommandArgv(&temp, argc, argv, argvlen);
     if (len == -1) {
         throw std::bad_alloc();
     }
@@ -52,11 +54,13 @@ void RedisCommand::formatArgv(int argc, const char **argv, const size_t *argvlen
 void RedisCommand::format(const vector<string> &commands)
 {
     vector<const char*> args;
+    vector<size_t> lens;
     for (auto& command : commands)
     {
         args.push_back(command.c_str());
+        lens.push_back(command.size());
     }
-    formatArgv(static_cast<int>(args.size()), args.data(), NULL);
+    formatArgv(static_cast<int>(args.size()), args.data(), lens.data());
 }
 
 /* Format HSET key multiple field value command */
@@ -96,12 +100,9 @@ void RedisCommand::formatHDEL(const std::string& key, const std::vector<std::str
 {
     if (fields.empty()) throw std::invalid_argument("empty values");
 
-    std::vector<const char *> args = {"HDEL", key.c_str()};
-    for (const std::string &f : fields)
-    {
-        args.push_back(f.c_str());
-    }
-    formatArgv(static_cast<int>(args.size()), args.data(), NULL);
+    std::vector<string> args = {"HDEL", key};
+    args.insert(args.end(), fields.begin(), fields.end());
+    format(args);
 }
 
 /* Format EXPIRE key field command */
@@ -122,14 +123,26 @@ void RedisCommand::formatDEL(const std::string& key)
     return format("DEL %s", key.c_str());
 }
 
+int RedisCommand::appendTo(redisContext *ctx) const
+{
+    return redisAppendFormattedCommand(ctx, c_str(), length());
+}
+
+std::string RedisCommand::toPrintableString() const
+{
+    return binary_to_printable(temp, len);
+}
+
 const char *RedisCommand::c_str() const
 {
     return temp;
 }
 
 size_t RedisCommand::length() const
 {
-    return strlen(temp);
+    if (len <= 0)
+        return 0;
+    return static_cast<size_t>(len);
 }
 
 }
diff --git a/common/rediscommand.h b/common/rediscommand.h
@@ -6,6 +6,7 @@
 #include <string>
 #include <stdexcept>
 #include <map>
+#include <hiredis/hiredis.h>
 
 namespace swss {
 
@@ -17,6 +18,7 @@ typedef std::tuple<std::string, std::string, std::vector<FieldValueTuple> > KeyO
 #define kfvOp     std::get<1>
 #define kfvFieldsValues std::get<2>
 
+
 class RedisCommand {
 public:
     RedisCommand();
@@ -64,12 +66,18 @@ class RedisCommand {
     /* Format DEL key command */
     void formatDEL(const std::string& key);
 
+    int appendTo(redisContext *ctx) const;
+
+    std::string toPrintableString() const;
+
+protected:
     const char *c_str() const;
 
     size_t length() const;
 
 private:
     char *temp;
+    int len;
 };
 
 template<typename InputIterator>
@@ -80,15 +88,15 @@ void RedisCommand::formatHSET(const std::string &key,
 
     const char* cmd = "HSET";
 
-    std::vector<const char*> args = { cmd, key.c_str() };
+    std::vector<std::string> args = { cmd, key.c_str() };
 
     for (auto i = start; i != stop; i++)
     {
-        args.push_back(fvField(*i).c_str());
-        args.push_back(fvValue(*i).c_str());
+        args.push_back(fvField(*i));
+        args.push_back(fvValue(*i));
     }
 
-    formatArgv((int)args.size(), args.data(), NULL);
+    format(args);
 }
 
 }
diff --git a/common/redispipeline.h b/common/redispipeline.h
@@ -49,7 +49,7 @@ class RedisPipeline {
             case REDIS_REPLY_STATUS:
             case REDIS_REPLY_INTEGER:
             {
-                int rc = redisAppendFormattedCommand(m_db->getContext(), command.c_str(), command.length());
+                int rc = command.appendTo(m_db->getContext());
                 if (rc != REDIS_OK)
                 {
                     // The only reason of error is REDIS_ERR_OOM (Out of memory)
diff --git a/common/redisreply.cpp b/common/redisreply.cpp
@@ -78,7 +78,7 @@ inline void guard(FUNC func, const char* command)
 
 RedisReply::RedisReply(RedisContext *ctx, const RedisCommand& command)
 {
-    int rc = redisAppendFormattedCommand(ctx->getContext(), command.c_str(), command.length());
+    int rc = command.appendTo(ctx->getContext());
     if (rc != REDIS_OK)
     {
         // The only reason of error is REDIS_ERR_OOM (Out of memory)
@@ -89,9 +89,9 @@ RedisReply::RedisReply(RedisContext *ctx, const RedisCommand& command)
     rc = redisGetReply(ctx->getContext(), (void**)&m_reply);
     if (rc != REDIS_OK)
     {
-        throw RedisError("Failed to redisGetReply with " + string(command.c_str()), ctx->getContext());
+        throw RedisError("Failed to redisGetReply with " + command.toPrintableString(), ctx->getContext());
     }
-    guard([&]{checkReply();}, command.c_str());
+    guard([&]{checkReply();}, command.toPrintableString().c_str());
 }
 
 RedisReply::RedisReply(RedisContext *ctx, const string& command)
@@ -109,19 +109,19 @@ RedisReply::RedisReply(RedisContext *ctx, const string& command)
     {
         throw RedisError("Failed to redisGetReply with " + command, ctx->getContext());
     }
-    guard([&]{checkReply();}, command.c_str());
+    guard([&]{checkReply();}, binary_to_printable(command.c_str(), command.length()).c_str());
 }
 
 RedisReply::RedisReply(RedisContext *ctx, const RedisCommand& command, int expectedType)
     : RedisReply(ctx, command)
 {
-    guard([&]{checkReplyType(expectedType);}, command.c_str());
+    guard([&]{checkReplyType(expectedType);}, command.toPrintableString().c_str());
 }
 
 RedisReply::RedisReply(RedisContext *ctx, const string& command, int expectedType)
     : RedisReply(ctx, command)
 {
-    guard([&]{checkReplyType(expectedType);}, command.c_str());
+    guard([&]{checkReplyType(expectedType);}, binary_to_printable(command.c_str(), command.length()).c_str());
 }
 
 RedisReply::RedisReply(redisReply *reply) :
diff --git a/common/stringutility.h b/common/stringutility.h
diff --git a/pyext/swsscommon.i b/pyext/swsscommon.i
diff --git a/tests/stringutility_ut.cpp b/tests/stringutility_ut.cpp

Original file line number	Diff line number	Diff line change
`@@ -1,13 +1,15 @@`
`1`	`1`	`#include <vector>`
`2`	`2`	`#include <hiredis/hiredis.h>`
`3`	`3`	`#include "rediscommand.h"`
	`4`	`+#include "stringutility.h"`
`4`	`5`
`5`	`6`	`using namespace std;`
`6`	`7`
`7`	`8`	`namespace swss {`
`8`	`9`
`9`	`10`	`RedisCommand::RedisCommand()`
`10`		`- : temp(NULL)`
	`11`	`+ : temp(NULL),`
	`12`	`+ len(0)`
`11`	`13`	`{`
`12`	`14`	`}`
`13`	`15`
`@@ -26,7 +28,7 @@ void RedisCommand::format(const char *fmt, ...)`
`26`	`28`
`27`	`29`	`va_list ap;`
`28`	`30`	`va_start(ap, fmt);`
`29`		`- int len = redisvFormatCommand(&temp, fmt, ap);`
	`31`	`+ len = redisvFormatCommand(&temp, fmt, ap);`
`30`	`32`	`va_end(ap);`
`31`	`33`	`if (len == -1) {`
`32`	`34`	`throw std::bad_alloc();`
`@@ -43,7 +45,7 @@ void RedisCommand::formatArgv(int argc, const char *argv, const size_t argvlen`
`43`	`45`	`temp = nullptr;`
`44`	`46`	`}`
`45`	`47`
`46`		`- int len = redisFormatCommandArgv(&temp, argc, argv, argvlen);`
	`48`	`+ len = redisFormatCommandArgv(&temp, argc, argv, argvlen);`
`47`	`49`	`if (len == -1) {`
`48`	`50`	`throw std::bad_alloc();`
`49`	`51`	`}`
`@@ -52,11 +54,13 @@ void RedisCommand::formatArgv(int argc, const char *argv, const size_t argvlen`
`52`	`54`	`void RedisCommand::format(const vector<string> &commands)`
`53`	`55`	`{`
`54`	`56`	`vector<const char*> args;`
	`57`	`+ vector<size_t> lens;`
`55`	`58`	`for (auto& command : commands)`
`56`	`59`	`{`
`57`	`60`	`args.push_back(command.c_str());`
	`61`	`+ lens.push_back(command.size());`
`58`	`62`	`}`
`59`		`- formatArgv(static_cast<int>(args.size()), args.data(), NULL);`
	`63`	`+ formatArgv(static_cast<int>(args.size()), args.data(), lens.data());`
`60`	`64`	`}`
`61`	`65`
`62`	`66`	`/* Format HSET key multiple field value command */`
`@@ -96,12 +100,9 @@ void RedisCommand::formatHDEL(const std::string& key, const std::vector<std::str`
`96`	`100`	`{`
`97`	`101`	`if (fields.empty()) throw std::invalid_argument("empty values");`
`98`	`102`
`99`		`- std::vector<const char *> args = {"HDEL", key.c_str()};`
`100`		`- for (const std::string &f : fields)`
`101`		`- {`
`102`		`- args.push_back(f.c_str());`
`103`		`- }`
`104`		`- formatArgv(static_cast<int>(args.size()), args.data(), NULL);`
	`103`	`+ std::vector<string> args = {"HDEL", key};`
	`104`	`+ args.insert(args.end(), fields.begin(), fields.end());`
	`105`	`+ format(args);`
`105`	`106`	`}`
`106`	`107`
`107`	`108`	`/* Format EXPIRE key field command */`
`@@ -122,14 +123,26 @@ void RedisCommand::formatDEL(const std::string& key)`
`122`	`123`	`return format("DEL %s", key.c_str());`
`123`	`124`	`}`
`124`	`125`
	`126`	`+int RedisCommand::appendTo(redisContext *ctx) const`
	`127`	`+{`
	`128`	`+ return redisAppendFormattedCommand(ctx, c_str(), length());`
	`129`	`+}`
	`130`	`+`
	`131`	`+std::string RedisCommand::toPrintableString() const`
	`132`	`+{`
	`133`	`+ return binary_to_printable(temp, len);`
	`134`	`+}`
	`135`	`+`
`125`	`136`	`const char *RedisCommand::c_str() const`
`126`	`137`	`{`
`127`	`138`	`return temp;`
`128`	`139`	`}`
`129`	`140`
`130`	`141`	`size_t RedisCommand::length() const`
`131`	`142`	`{`
`132`		`- return strlen(temp);`
	`143`	`+ if (len <= 0)`
	`144`	`+ return 0;`
	`145`	`+ return static_cast<size_t>(len);`
`133`	`146`	`}`
`134`	`147`
`135`	`148`	`}`
Original file line number	Diff line number	Diff line change
`@@ -49,7 +49,7 @@ class RedisPipeline {`
`49`	`49`	`case REDIS_REPLY_STATUS:`
`50`	`50`	`case REDIS_REPLY_INTEGER:`
`51`	`51`	`{`
`52`		`- int rc = redisAppendFormattedCommand(m_db->getContext(), command.c_str(), command.length());`
	`52`	`+ int rc = command.appendTo(m_db->getContext());`
`53`	`53`	`if (rc != REDIS_OK)`
`54`	`54`	`{`
`55`	`55`	`// The only reason of error is REDIS_ERR_OOM (Out of memory)`
Original file line number	Diff line number	Diff line change
`@@ -78,7 +78,7 @@ inline void guard(FUNC func, const char* command)`
`78`	`78`
`79`	`79`	`RedisReply::RedisReply(RedisContext *ctx, const RedisCommand& command)`
`80`	`80`	`{`
`81`		`- int rc = redisAppendFormattedCommand(ctx->getContext(), command.c_str(), command.length());`
	`81`	`+ int rc = command.appendTo(ctx->getContext());`
`82`	`82`	`if (rc != REDIS_OK)`
`83`	`83`	`{`
`84`	`84`	`// The only reason of error is REDIS_ERR_OOM (Out of memory)`
`@@ -89,9 +89,9 @@ RedisReply::RedisReply(RedisContext *ctx, const RedisCommand& command)`
`89`	`89`	`rc = redisGetReply(ctx->getContext(), (void**)&m_reply);`
`90`	`90`	`if (rc != REDIS_OK)`
`91`	`91`	`{`
`92`		`- throw RedisError("Failed to redisGetReply with " + string(command.c_str()), ctx->getContext());`
	`92`	`+ throw RedisError("Failed to redisGetReply with " + command.toPrintableString(), ctx->getContext());`
`93`	`93`	`}`
`94`		`- guard([&]{checkReply();}, command.c_str());`
	`94`	`+ guard([&]{checkReply();}, command.toPrintableString().c_str());`
`95`	`95`	`}`
`96`	`96`
`97`	`97`	`RedisReply::RedisReply(RedisContext *ctx, const string& command)`
`@@ -109,19 +109,19 @@ RedisReply::RedisReply(RedisContext *ctx, const string& command)`
`109`	`109`	`{`
`110`	`110`	`throw RedisError("Failed to redisGetReply with " + command, ctx->getContext());`
`111`	`111`	`}`
`112`		`- guard([&]{checkReply();}, command.c_str());`
	`112`	`+ guard([&]{checkReply();}, binary_to_printable(command.c_str(), command.length()).c_str());`
`113`	`113`	`}`
`114`	`114`
`115`	`115`	`RedisReply::RedisReply(RedisContext *ctx, const RedisCommand& command, int expectedType)`
`116`	`116`	`: RedisReply(ctx, command)`
`117`	`117`	`{`
`118`		`- guard([&]{checkReplyType(expectedType);}, command.c_str());`
	`118`	`+ guard([&]{checkReplyType(expectedType);}, command.toPrintableString().c_str());`
`119`	`119`	`}`
`120`	`120`
`121`	`121`	`RedisReply::RedisReply(RedisContext *ctx, const string& command, int expectedType)`
`122`	`122`	`: RedisReply(ctx, command)`
`123`	`123`	`{`
`124`		`- guard([&]{checkReplyType(expectedType);}, command.c_str());`
	`124`	`+ guard([&]{checkReplyType(expectedType);}, binary_to_printable(command.c_str(), command.length()).c_str());`
`125`	`125`	`}`
`126`	`126`
`127`	`127`	`RedisReply::RedisReply(redisReply *reply) :`