ZMQ based state tables

common/Makefile.am

@@ -33,6 +33,7 @@ common_libswsscommon_la_SOURCES = \
     common/json.cpp                  \
     common/producertable.cpp         \
     common/producerstatetable.cpp    \
+    common/zmqproducerstatetable.cpp \
     common/rediscommand.cpp          \
     common/redistran.cpp             \
     common/redisselect.cpp           \
@@ -42,6 +43,7 @@ common_libswsscommon_la_SOURCES = \
     common/consumertable.cpp         \
     common/consumertablebase.cpp     \
     common/consumerstatetable.cpp    \
+    common/zmqconsumerstatetable.cpp \
     common/ipaddress.cpp             \
     common/ipprefix.cpp              \
     common/ipaddresses.cpp           \
@@ -66,6 +68,8 @@ common_libswsscommon_la_SOURCES = \
     common/redisutility.cpp          \
     common/restart_waiter.cpp        \
     common/profileprovider.cpp       \
+    common/zmqclient.cpp             \
+    common/zmqserver.cpp             \
     common/redis_table_waiter.cpp
 
 common_libswsscommon_la_CXXFLAGS = $(DBGFLAGS) $(AM_CFLAGS) $(CFLAGS_COMMON) $(LIBNL_CFLAGS) $(CODE_COVERAGE_CXXFLAGS)
@@ -76,6 +80,7 @@ common_swssloglevel_SOURCES =  \
 	common/loglevel.cpp        \
 	common/loglevel_util.cpp
 
+
 common_swssloglevel_CXXFLAGS = $(DBGFLAGS) $(AM_CFLAGS) $(CFLAGS_COMMON) $(CODE_COVERAGE_CXXFLAGS)
 common_swssloglevel_CPPFLAGS = $(DBGFLAGS) $(AM_CFLAGS) $(CFLAGS_COMMON) $(CODE_COVERAGE_CPPFLAGS)
 common_swssloglevel_LDADD = common/libswsscommon.la $(CODE_COVERAGE_LIBS)

common/armhelper.h

@@ -0,0 +1,15 @@
+#ifndef __ARM_HELPER__
+#define __ARM_HELPER__
+
+#if defined(__arm__) || defined(__aarch64__)
+#define WARNINGS_NO_CAST_ALIGN \
+    _Pragma ("GCC diagnostic push") \
+    _Pragma ("GCC diagnostic ignored \"-Wcast-align\"")
+#define WARNINGS_RESET \
+    _Pragma ("GCC diagnostic pop")
+#else
+#define WARNINGS_NO_CAST_ALIGN
+#define WARNINGS_RESET
+#endif
+
+#endif

common/binaryserializer.h

@@ -0,0 +1,154 @@
+#ifndef __BINARY_SERIALIZER__
+#define __BINARY_SERIALIZER__
+
+#include "common/armhelper.h"
+
+namespace swss {
+
+class BinarySerializer {
+public:
+    static size_t serializeBuffer(
+        const char* buffer,
+        const size_t size,
+        const std::string& key,
+        const std::vector<swss::FieldValueTuple>& values,
+        const std::string& command,
+        const std::string& dbName,
+        const std::string& tableName)
+    {
+        auto tmpSerializer = BinarySerializer(buffer, size);
+
+        tmpSerializer.setKeyAndValue(
+                                    dbName.c_str(), dbName.length(),
+                                    tableName.c_str(), tableName.length());
+        tmpSerializer.setKeyAndValue(
+                                    key.c_str(), key.length(),
+                                    command.c_str(), command.length());
+        for (auto& kvp : values)
+        {
+            auto& field = fvField(kvp);
+            auto& value = fvValue(kvp);
+            tmpSerializer.setKeyAndValue(
+                                        field.c_str(), field.length(),
+                                        value.c_str(), value.length());
+        }
+
+        return tmpSerializer.finalize();
+    }
+
+    static void deserializeBuffer(
+        const char* buffer,
+        const size_t size,
+        std::vector<swss::FieldValueTuple>& values)
+    {
+        WARNINGS_NO_CAST_ALIGN;
+        auto pkvp_count = (const size_t*)buffer;
+        WARNINGS_RESET;
+
+        size_t kvp_count = *pkvp_count;
+        auto tmp_buffer = buffer + sizeof(size_t);
+        while (kvp_count > 0)
+        {
+            kvp_count--;
+
+            // read key and value from buffer
+            WARNINGS_NO_CAST_ALIGN;
+            auto pkeylen = (const size_t*)tmp_buffer;
+            WARNINGS_RESET;
+
+            tmp_buffer += sizeof(size_t);
+            if ((size_t)(tmp_buffer - buffer + *pkeylen)  > size)
+            {
+                SWSS_LOG_THROW("serialized key data was truncated, key length: %zu, increase buffer size: %zu",
+                                                                                            *pkeylen,
+                                                                                            size);
+            }
+
+            auto pkey = tmp_buffer;
+            tmp_buffer += *pkeylen;
+
+            WARNINGS_NO_CAST_ALIGN;
+            auto pvallen = (const size_t*)tmp_buffer;
+            WARNINGS_RESET;
+
+            tmp_buffer += sizeof(size_t);
+            if ((size_t)(tmp_buffer - buffer + *pvallen)  > size)
+            {
+                SWSS_LOG_THROW("serialized value data was truncated,, value length: %zu increase buffer size: %zu",
+                                                                                            *pvallen,
+                                                                                            size);
+            }
+
+            auto pval = tmp_buffer;
+            tmp_buffer += *pvallen;
+
+            values.push_back(std::make_pair(pkey, pval));
+        }
+    }
+
+private:
+    const char* m_buffer;
+    const size_t m_buffer_size;
+    char* m_current_position;
+    size_t m_kvp_count;
+
+    BinarySerializer(const char* buffer, const size_t size)
+        : m_buffer(buffer), m_buffer_size(size)
+    {
+        resetSerializer();
+    }
+
+    void resetSerializer()
+    {
+        m_current_position = const_cast<char*>(m_buffer) + sizeof(size_t);
+        m_kvp_count = 0;
+    }
+
+    void setKeyAndValue(const char* key, size_t klen,
+                        const char* value, size_t vlen)
+    {
+        // to improve deserialize performance, copy null-terminated string. 
+        setData(key, klen + 1);
+        setData(value, vlen + 1);
+
+        m_kvp_count++;
+    }
+
+    size_t finalize()
+    {
+        // set key value pair count to message
+        WARNINGS_NO_CAST_ALIGN;
+        size_t* pkvp_count = (size_t*)const_cast<char*>(m_buffer);
+        WARNINGS_RESET;
+
+        *pkvp_count = m_kvp_count;
+
+        // return size
+        return m_current_position - m_buffer;
+    }
+
+    void setData(const char* data, size_t datalen)
+    {
+        if ((size_t)(m_current_position - m_buffer + datalen + sizeof(size_t)) > m_buffer_size)
+        {
+            SWSS_LOG_THROW("There are not enough buffer for binary serializer to serialize,\
+                             key count: %zu, data length %zu, buffer size: %zu",
+                                                m_kvp_count,
+                                                datalen,
+                                                m_buffer_size);
+        }
+
+        WARNINGS_NO_CAST_ALIGN;
+        size_t* pdatalen = (size_t*)m_current_position;
+        WARNINGS_RESET;
+
+        *pdatalen = datalen;
+        m_current_position += sizeof(size_t);
+
+        memcpy(m_current_position, data, datalen);
+        m_current_position += datalen;
+    }
+};
+
+}
+#endif

common/defaultvalueprovider.cpp

@@ -11,17 +11,7 @@
 #include "logger.h"
 #include "table.h"
 #include "json.h"
-
-#if defined(__arm__) || defined(__aarch64__)
-#define WARNINGS_NO_CAST_ALIGN \
-    _Pragma ("GCC diagnostic push") \
-    _Pragma ("GCC diagnostic ignored \"-Wcast-align\"")
-#define WARNINGS_RESET \
-    _Pragma ("GCC diagnostic pop")
-#else
-#define WARNINGS_NO_CAST_ALIGN
-#define WARNINGS_RESET
-#endif
+#include "armhelper.h"
 
 using namespace std;
 using namespace swss;

common/table.h

@@ -39,29 +39,34 @@ class TableBase {
 #endif
     TableBase(int dbId, const std::string &tableName)
         : m_tableName(tableName)
+    {
+        m_tableSeparator = getTableSeparator(dbId);
+    }
+
+    TableBase(const std::string &tableName, const std::string &tableSeparator)
+        : m_tableName(tableName), m_tableSeparator(tableSeparator)
+    {
+        static const std::string legalSeparators = ":|";
+        if (legalSeparators.find(tableSeparator) == std::string::npos)
+            throw std::invalid_argument("Invalid table name separator");
+    }
+
+    static std::string getTableSeparator(int dbId)
     {
         /* Look up table separator for the provided DB */
         auto it = tableNameSeparatorMap.find(dbId);
 
         if (it != tableNameSeparatorMap.end())
         {
-            m_tableSeparator = it->second;
+            return it->second;
         }
         else
         {
             SWSS_LOG_NOTICE("Unrecognized database ID. Using default table name separator ('%s')", TABLE_NAME_SEPARATOR_VBAR.c_str());
-            m_tableSeparator = TABLE_NAME_SEPARATOR_VBAR;
+            return TABLE_NAME_SEPARATOR_VBAR;
         }
     }
 
-    TableBase(const std::string &tableName, const std::string &tableSeparator)
-        : m_tableName(tableName), m_tableSeparator(tableSeparator)
-    {
-        static const std::string legalSeparators = ":|";
-        if (legalSeparators.find(tableSeparator) == std::string::npos)
-            throw std::invalid_argument("Invalid table name separator");
-    }
-
     std::string getTableName() const { return m_tableName; }
 
     /* Return the actual key name as a combination of tableName<table_separator>key */