ZMQ lib change.

common/c-api/util.h

@@ -202,6 +202,7 @@ template <class T> static inline T &getReference(std::shared_ptr<T> &t) {
 // T is anything that has a .size() method and which can be iterated over for
 // swss::KeyOpFieldValuesTuple, eg vector or deque
 template <class T> static inline SWSSKeyOpFieldValuesArray makeKeyOpFieldValuesArray(T &&in) {
+    SWSS_LOG_DEBUG("Entering makeKeyOpFieldValuesArray method");
     SWSSKeyOpFieldValues *data = new SWSSKeyOpFieldValues[in.size()];
 
     size_t i = 0;
@@ -211,6 +212,7 @@ template <class T> static inline SWSSKeyOpFieldValuesArray makeKeyOpFieldValuesA
     SWSSKeyOpFieldValuesArray out;
     out.len = (uint64_t)in.size();
     out.data = data;
+//    SWSS_LOG_DEBUG("2::: out.len value is: %ld", out.len);
     return out;
 }
 
@@ -255,9 +257,11 @@ static inline std::vector<swss::KeyOpFieldsValuesTuple>
 takeKeyOpFieldValuesArray(SWSSKeyOpFieldValuesArray in) {
     std::vector<swss::KeyOpFieldsValuesTuple> out;
     for (uint64_t i = 0; i < in.len; i++) {
+//        SWSS_LOG_DEBUG("i value is: %ld", i);
         SWSSKeyOpFieldValues kfv = in.data[i];
         out.push_back(takeKeyOpFieldValues(std::move(kfv)));
     }
+//        SWSS_LOG_DEBUG("out.len value is: %ld", out.size());
     return out;
 }
 

common/c-api/zmqclient.cpp

@@ -25,6 +25,7 @@ void SWSSZmqClient_connect(SWSSZmqClient zmqc) {
 
 void SWSSZmqClient_sendMsg(SWSSZmqClient zmqc, const char *dbName, const char *tableName,
                            SWSSKeyOpFieldValuesArray arr) {
+    SWSS_LOG_DEBUG("Inside SWSSZmqClient_sendMsg");
     SWSSTry({
         vector<KeyOpFieldsValuesTuple> kcos = takeKeyOpFieldValuesArray(arr);
         ((ZmqClient *)zmqc)

common/zmqclient.cpp

@@ -18,13 +18,20 @@ namespace swss {
 ZmqClient::ZmqClient(const std::string& endpoint)
 :ZmqClient(endpoint, "")
 {
+    initialize(endpoint);
 }
 
 ZmqClient::ZmqClient(const std::string& endpoint, const std::string& vrf)
 {
     initialize(endpoint, vrf);
 }
 
+ZmqClient::ZmqClient(const std::string& endpoint, uint32_t waitTimeMs) :
+    m_waitTimeMs(waitTimeMs)
+{
+    initialize(endpoint);
+}
+
 ZmqClient::~ZmqClient()
 {
     std::lock_guard<std::mutex> lock(m_socketMutex);
@@ -55,6 +62,17 @@ void ZmqClient::initialize(const std::string& endpoint, const std::string& vrf)
 
     connect();
 }
+
+void ZmqClient::initialize(const std::string& endpoint)
+{
+    m_connected = false;
+    m_endpoint = endpoint;
+    m_context = nullptr;
+    m_socket = nullptr;
+    m_sendbuffer.resize(MQ_RESPONSE_MAX_COUNT);
+
+    connect();
+}
 
 bool ZmqClient::isConnected()
 {
@@ -63,6 +81,7 @@ bool ZmqClient::isConnected()
 
 void ZmqClient::connect()
 {
+SWSS_LOG_ERROR("DIV:: Inside function client connect");
     if (m_connected)
     {
         SWSS_LOG_DEBUG("Already connected to endpoint: %s", m_endpoint.c_str());
@@ -88,6 +107,7 @@ void ZmqClient::connect()
     m_context = zmq_ctx_new();
     m_socket = zmq_socket(m_context, ZMQ_PUSH);
 
+    SWSS_LOG_DEBUG("m_socket in client connect() is: %p\n", m_socket);
     // timeout all pending send package, so zmq will not block in dtor of this class: http://api.zeromq.org/master:zmq-setsockopt
     int linger = 0;
     zmq_setsockopt(m_socket, ZMQ_LINGER, &linger, sizeof(linger));
@@ -119,6 +139,7 @@ void ZmqClient::sendMsg(
         const std::string& tableName,
         const std::vector<KeyOpFieldsValuesTuple>& kcos)
 {
+SWSS_LOG_ERROR("DIV:: Inside function client sendMsg");
     int serializedlen = (int)BinarySerializer::serializeBuffer(
                                                         m_sendbuffer.data(),
                                                         m_sendbuffer.size(),
@@ -137,16 +158,18 @@ void ZmqClient::sendMsg(
     int zmq_err = 0;
     int retry_delay = 10;
     int rc = 0;
-    for (int i = 0; i <=  MQ_MAX_RETRY; ++i)
+    for (int i = 0; i <= MQ_MAX_RETRY; ++i)
     {
         {
             // ZMQ socket is not thread safe: http://api.zeromq.org/2-1:zmq
             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, m_sendbuffer.data(), serializedlen, ZMQ_NOBLOCK);
+        rc = zmq_send(m_socket, m_sendbuffer.data(), serializedlen, ZMQ_NOBLOCK);
         }
-
+//SWSS_LOG_DEBUG("Before Sleep() in client sendmsg");
+//       usleep(10 * 1000);
+//SWSS_LOG_DEBUG("After Sleep() in client sendmsg");
         if (rc >= 0)
         {
             SWSS_LOG_DEBUG("zmq sended %d bytes", serializedlen);
@@ -164,7 +187,7 @@ void ZmqClient::sendMsg(
             //       For example when ZMQ socket still not receive reply message from last sended package.
             //       There was state machine inside ZMQ socket, when the socket is not in ready to send state, this error will happen.
             // for more detail, please check: http://api.zeromq.org/2-1:zmq-send
-            SWSS_LOG_DEBUG("zmq send retry, endpoint: %s, error: %d", m_endpoint.c_str(), zmq_err);
+            SWSS_LOG_WARN("zmq send retry, endpoint: %s, error: %d", m_endpoint.c_str(), zmq_err);
 
             retry_delay = 0;
         }
@@ -183,7 +206,7 @@ void ZmqClient::sendMsg(
         else
         {
             // for other error, send failed immediately.
-            auto message =  "zmq send failed, endpoint: " + m_endpoint + ", error: " + to_string(rc);
+            auto message =  "cli: zmq send failed, endpoint: " + m_endpoint + ", error: " + to_string(rc);
             SWSS_LOG_ERROR("%s", message.c_str());
             throw system_error(make_error_code(errc::io_error), message);
         }
@@ -192,9 +215,18 @@ void ZmqClient::sendMsg(
     }
 
     // failed after retry
-    auto message =  "zmq send failed, endpoint: " + m_endpoint + ", zmqerrno: " + to_string(zmq_err) + ":" + zmq_strerror(zmq_err) + ", msg length:" + to_string(serializedlen);
+    auto message =  "cli: zmq send failed, endpoint: " + m_endpoint + ", zmqerrno: " + to_string(zmq_err) + ":" + zmq_strerror(zmq_err) + ", msg length:" + to_string(serializedlen);
     SWSS_LOG_ERROR("%s", message.c_str());
     throw system_error(make_error_code(errc::io_error), message);
 }
 
+bool ZmqClient::wait(const std::string& dbName,
+                     const std::string& tableName,
+                     const std::vector<std::shared_ptr<KeyOpFieldsValuesTuple>>& kcos)
+{
+SWSS_LOG_ERROR("DIV:: Inside function wait");
+    SWSS_LOG_ENTER();
+
+    return false;
+}
 }

common/zmqclient.h

@@ -12,8 +12,10 @@ namespace swss {
 class ZmqClient
 {
 public:
+
     ZmqClient(const std::string& endpoint);
     ZmqClient(const std::string& endpoint, const std::string& vrf);
+    ZmqClient(const std::string& endpoint, uint32_t waitTimeMs);
     ~ZmqClient();
 
     bool isConnected();
@@ -23,8 +25,14 @@ class ZmqClient
     void sendMsg(const std::string& dbName,
                  const std::string& tableName,
                  const std::vector<KeyOpFieldsValuesTuple>& kcos);
+
+    bool wait(const std::string& dbName,
+              const std::string& tableName,
+              const std::vector<std::shared_ptr<KeyOpFieldsValuesTuple>>& kcos);
+
 private:
     void initialize(const std::string& endpoint, const std::string& vrf);
+    void initialize(const std::string& endpoint);
 
     std::string m_endpoint;
 
@@ -36,9 +44,11 @@ class ZmqClient
 
     bool m_connected;
 
+    uint32_t m_waitTimeMs;
+
     std::mutex m_socketMutex;
-    
-    std::vector<char> m_sendbuffer;
+
+    std::vector<char> m_sendbuffer;    
 };
 
 }

common/zmqconsumerstatetable.cpp

@@ -41,6 +41,7 @@ ZmqConsumerStateTable::ZmqConsumerStateTable(DBConnector *db, const std::string
 
 void ZmqConsumerStateTable::handleReceivedData(const std::vector<std::shared_ptr<KeyOpFieldsValuesTuple>> &kcos)
 {
+        SWSS_LOG_DEBUG("Entering ZmqConsumerStateTable::handleReceivedData");
     for (auto kco : kcos)
     {
         std::shared_ptr<KeyOpFieldsValuesTuple> clone = nullptr;
@@ -53,6 +54,7 @@ void ZmqConsumerStateTable::handleReceivedData(const std::vector<std::shared_ptr
         {
             std::lock_guard<std::mutex> lock(m_receivedQueueMutex);
             m_receivedOperationQueue.push(kco);
+SWSS_LOG_DEBUG("Called m_receivedOperationQueue.push in handleReceivedData()");
         }
 
         if (m_asyncDBUpdater != nullptr)
@@ -73,6 +75,7 @@ void ZmqConsumerStateTable::pops(std::deque<KeyOpFieldsValuesTuple> &vkco, const
 
         // For new data append to m_dataQueue during pops, will not be include in result.
         count = m_receivedOperationQueue.size();
+        //SWSS_LOG_DEBUG("count value: %ld", count);
         if (!count)
         {
             return;
@@ -82,6 +85,7 @@ void ZmqConsumerStateTable::pops(std::deque<KeyOpFieldsValuesTuple> &vkco, const
     vkco.clear();
     for (size_t ie = 0; ie < count; ie++)
     {
+        //SWSS_LOG_DEBUG("count inside for loop: %ld", count);
         auto& kco = *(m_receivedOperationQueue.front());
         vkco.push_back(std::move(kco));
 

common/zmqproducerstatetable.cpp

@@ -148,6 +148,7 @@ void ZmqProducerStateTable::del(const std::vector<std::string> &keys)
 
 void ZmqProducerStateTable::send(const std::vector<KeyOpFieldsValuesTuple> &kcos)
 {
+SWSS_LOG_DEBUG("DIV:: Inside function ZmqProducerStateTable::send");
     m_zmqClient.sendMsg(
                         m_dbName,
                         m_tableNameStr,
@@ -164,6 +165,14 @@ void ZmqProducerStateTable::send(const std::vector<KeyOpFieldsValuesTuple> &kcos
     }
 }
 
+bool ZmqProducerStateTable::wait(const std::string& dbName,
+              const std::string& tableName,
+              const std::vector<std::shared_ptr<KeyOpFieldsValuesTuple>>& kcos)
+{
+SWSS_LOG_DEBUG("DIV:: Inside function ZmqProducerStateTable::wait");
+    return m_zmqClient.wait(dbName, tableName, kcos);
+}
+
 size_t ZmqProducerStateTable::dbUpdaterQueueSize()
 {
     if (m_asyncDBUpdater == nullptr)

common/zmqproducerstatetable.h

@@ -18,6 +18,7 @@ class ZmqProducerStateTable : public ProducerStateTable
 public:
     ZmqProducerStateTable(DBConnector *db, const std::string &tableName, ZmqClient &zmqClient, bool dbPersistence = true);
     ZmqProducerStateTable(RedisPipeline *pipeline, const std::string &tableName, ZmqClient &zmqClient, bool buffered = false, bool dbPersistence = true);
+//    ~ZmqProducerStateTable() = default;
 
     /* Implements set() and del() commands using notification messages */
     virtual void set(const std::string &key,
@@ -37,6 +38,12 @@ class ZmqProducerStateTable : public ProducerStateTable
     // Batched send that can include both SET and DEL requests.
     virtual void send(const std::vector<KeyOpFieldsValuesTuple> &kcos);
 
+    // This method should only be used if the ZmqClient enables one-to-one sync.
+
+    virtual bool wait(const std::string& dbName,
+              const std::string& tableName,
+              const std::vector<std::shared_ptr<KeyOpFieldsValuesTuple>>& kcos);
+
     size_t dbUpdaterQueueSize();
 private:
     void initialize(DBConnector *db, const std::string &tableName, bool dbPersistence);