Improve memory usage by move ZMQ serialize buffer from ZmqProducerSta…

…teTable to ZmqClient (#955)

#### Why I did it
Every ZmqProducerStateTable  will allocate 16MB buffer, this can be improve by share same buffer in ZmqClient.

#### How I did it
Improve memory usage by move ZMQ serialize buffer from ZmqProducerStateTable to ZmqClient.

##### Work item tracking

#### How to verify it
Pass all test cases.

#### Which release branch to backport (provide reason below if selected)

<!--
- Note we only backport fixes to a release branch, *not* features!
- Please also provide a reason for the backporting below.
- e.g.
- [x] 202006
-->

- [ ] 201811
- [ ] 201911
- [ ] 202006
- [ ] 202012
- [ ] 202106
- [ ] 202111

#### Description for the changelog
Improve memory usage by move ZMQ serialize buffer from ZmqProducerStateTable to ZmqClient.

#### Link to config_db schema for YANG module changes
<!--
Provide a link to config_db schema for the table for which YANG model
is defined
Link should point to correct section on https://github.com/Azure/SONiC/wiki/Configuration.
-->

#### A picture of a cute animal (not mandatory but encouraged)

common/c-api/zmqclient.cpp

@@ -27,9 +27,7 @@ void SWSSZmqClient_sendMsg(SWSSZmqClient zmqc, const char *dbName, const char *t
                            SWSSKeyOpFieldValuesArray arr) {
     SWSSTry({
         vector<KeyOpFieldsValuesTuple> kcos = takeKeyOpFieldValuesArray(arr);
-        size_t bufSize = BinarySerializer::serializedSize(dbName, tableName, kcos);
-        vector<char> v(bufSize);
         ((ZmqClient *)zmqc)
-            ->sendMsg(string(dbName), string(tableName), kcos, v);
+            ->sendMsg(string(dbName), string(tableName), kcos);
     });
 }

common/zmqclient.cpp

@@ -51,6 +51,7 @@ void ZmqClient::initialize(const std::string& endpoint, const std::string& vrf)
     m_context = nullptr;
     m_socket = nullptr;
     m_vrf = vrf;
+    m_sendbuffer.resize(MQ_RESPONSE_MAX_COUNT);
 
     connect();
 }
@@ -116,12 +117,11 @@ void ZmqClient::connect()
 void ZmqClient::sendMsg(
         const std::string& dbName,
         const std::string& tableName,
-        const std::vector<KeyOpFieldsValuesTuple>& kcos,
-        std::vector<char>& sendbuffer)
+        const std::vector<KeyOpFieldsValuesTuple>& kcos)
 {
     int serializedlen = (int)BinarySerializer::serializeBuffer(
-                                                        sendbuffer.data(),
-                                                        sendbuffer.size(),
+                                                        m_sendbuffer.data(),
+                                                        m_sendbuffer.size(),
                                                         dbName,
                                                         tableName,
                                                         kcos);
@@ -144,7 +144,7 @@ void ZmqClient::sendMsg(
             std::lock_guard<std::mutex> lock(m_socketMutex);
 
             // Use none block mode to use all bandwidth: http://api.zeromq.org/2-1%3Azmq-send
-            rc = zmq_send(m_socket, sendbuffer.data(), serializedlen, ZMQ_NOBLOCK);
+            rc = zmq_send(m_socket, m_sendbuffer.data(), serializedlen, ZMQ_NOBLOCK);
         }
 
         if (rc >= 0)

common/zmqclient.h

@@ -22,8 +22,7 @@ class ZmqClient
 
     void sendMsg(const std::string& dbName,
                  const std::string& tableName,
-                 const std::vector<KeyOpFieldsValuesTuple>& kcos,
-                 std::vector<char>& sendbuffer);
+                 const std::vector<KeyOpFieldsValuesTuple>& kcos);
 private:
     void initialize(const std::string& endpoint, const std::string& vrf);
 
@@ -38,6 +37,8 @@ class ZmqClient
     bool m_connected;
 
     std::mutex m_socketMutex;
+
+    std::vector<char> m_sendbuffer;
 };
 
 }

common/zmqproducerstatetable.cpp

@@ -38,8 +38,6 @@ ZmqProducerStateTable::ZmqProducerStateTable(RedisPipeline *pipeline, const stri
 
 void ZmqProducerStateTable::initialize(DBConnector *db, const std::string &tableName, bool dbPersistence)
 {
-    m_sendbuffer.resize(MQ_RESPONSE_MAX_COUNT);
-
     if (dbPersistence)
     {
         SWSS_LOG_DEBUG("Database persistence enabled, tableName: %s", tableName.c_str());
@@ -64,8 +62,7 @@ void ZmqProducerStateTable::set(
     m_zmqClient.sendMsg(
                         m_dbName,
                         m_tableNameStr,
-                        kcos,
-                        m_sendbuffer);
+                        kcos);
 
     if (m_asyncDBUpdater != nullptr)
     {
@@ -93,8 +90,7 @@ void ZmqProducerStateTable::del(
     m_zmqClient.sendMsg(
                         m_dbName,
                         m_tableNameStr,
-                        kcos,
-                        m_sendbuffer);
+                        kcos);
 
     if (m_asyncDBUpdater != nullptr)
     {
@@ -112,8 +108,7 @@ void ZmqProducerStateTable::set(const std::vector<KeyOpFieldsValuesTuple> &value
     m_zmqClient.sendMsg(
                         m_dbName,
                         m_tableNameStr,
-                        values,
-                        m_sendbuffer);
+                        values);
 
     if (m_asyncDBUpdater != nullptr)
     {
@@ -136,8 +131,7 @@ void ZmqProducerStateTable::del(const std::vector<std::string> &keys)
     m_zmqClient.sendMsg(
                         m_dbName,
                         m_tableNameStr,
-                        kcos,
-                        m_sendbuffer);
+                        kcos);
 
     if (m_asyncDBUpdater != nullptr)
     {
@@ -157,8 +151,7 @@ void ZmqProducerStateTable::send(const std::vector<KeyOpFieldsValuesTuple> &kcos
     m_zmqClient.sendMsg(
                         m_dbName,
                         m_tableNameStr,
-                        kcos,
-                        m_sendbuffer);
+                        kcos);
 
     if (m_asyncDBUpdater != nullptr)
     {

common/zmqproducerstatetable.h

@@ -42,8 +42,6 @@ class ZmqProducerStateTable : public ProducerStateTable
     void initialize(DBConnector *db, const std::string &tableName, bool dbPersistence);
 
     ZmqClient& m_zmqClient;
-
-    std::vector<char> m_sendbuffer;
 
     const std::string m_dbName;
     const std::string m_tableNameStr;