Skip to content

Commit 1593cc6

Browse files
divyagayathri-hcldivyagayathri-hcl
authored
ZMQ lib change. (#958)
* zmq wait * ut pass case * swig_verify * Revert "swig_verify" * ... * zmqwait * const * ... * constant * Without debugs
1 parent 505381e commit 1593cc6

File tree

8 files changed

+204
-45
lines changed

8 files changed

+204
-45
lines changed

‎common/zmqclient.cpp‎

Lines changed: 15 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,12 @@ ZmqClient::ZmqClient(const std::string& endpoint, const std::string& vrf)
2525
initialize(endpoint, vrf);
2626
}
2727

28+
ZmqClient::ZmqClient(const std::string& endpoint, uint32_t waitTimeMs)
29+
: m_waitTimeMs(waitTimeMs)
30+
{
31+
initialize(endpoint);
32+
}
33+
2834
ZmqClient::~ZmqClient()
2935
{
3036
std::lock_guard<std::mutex> lock(m_socketMutex);
@@ -55,7 +61,7 @@ void ZmqClient::initialize(const std::string& endpoint, const std::string& vrf)
5561

5662
connect();
5763
}
58-
64+
5965
bool ZmqClient::isConnected()
6066
{
6167
return m_connected;
@@ -137,7 +143,7 @@ void ZmqClient::sendMsg(
137143
int zmq_err = 0;
138144
int retry_delay = 10;
139145
int rc = 0;
140-
for (int i = 0; i <= MQ_MAX_RETRY; ++i)
146+
for (int i = 0; i <= MQ_MAX_RETRY; ++i)
141147
{
142148
{
143149
// ZMQ socket is not thread safe: http://api.zeromq.org/2-1:zmq
@@ -146,7 +152,6 @@ void ZmqClient::sendMsg(
146152
// Use none block mode to use all bandwidth: http://api.zeromq.org/2-1%3Azmq-send
147153
rc = zmq_send(m_socket, m_sendbuffer.data(), serializedlen, ZMQ_NOBLOCK);
148154
}
149-
150155
if (rc >= 0)
151156
{
152157
SWSS_LOG_DEBUG("zmq sended %d bytes", serializedlen);
@@ -197,4 +202,11 @@ void ZmqClient::sendMsg(
197202
throw system_error(make_error_code(errc::io_error), message);
198203
}
199204

205+
// TODO: To be implemented later, required for ZMQ_CLIENT & ZMQ_SERVER
206+
// socket types in response path.
207+
bool ZmqClient::wait(
208+
const std::string &dbName, const std::string &tableName,
209+
const std::vector<std::shared_ptr<KeyOpFieldsValuesTuple>> &kcos) {
210+
return false;
211+
}
200212
}

‎common/zmqclient.h‎

Lines changed: 11 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -12,8 +12,10 @@ namespace swss {
1212
class ZmqClient
1313
{
1414
public:
15+
1516
ZmqClient(const std::string& endpoint);
1617
ZmqClient(const std::string& endpoint, const std::string& vrf);
18+
ZmqClient(const std::string& endpoint, uint32_t waitTimeMs);
1719
~ZmqClient();
1820

1921
bool isConnected();
@@ -23,8 +25,13 @@ class ZmqClient
2325
void sendMsg(const std::string& dbName,
2426
const std::string& tableName,
2527
const std::vector<KeyOpFieldsValuesTuple>& kcos);
28+
29+
bool wait(const std::string& dbName,
30+
const std::string& tableName,
31+
const std::vector<std::shared_ptr<KeyOpFieldsValuesTuple>>& kcos);
32+
2633
private:
27-
void initialize(const std::string& endpoint, const std::string& vrf);
34+
void initialize(const std::string& endpoint, const std::string& vrf = "");
2835

2936
std::string m_endpoint;
3037

@@ -36,8 +43,10 @@ class ZmqClient
3643

3744
bool m_connected;
3845

46+
uint32_t m_waitTimeMs;
47+
3948
std::mutex m_socketMutex;
40-
49+
4150
std::vector<char> m_sendbuffer;
4251
};
4352

‎common/zmqproducerstatetable.cpp‎

Lines changed: 18 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -1,18 +1,18 @@
1-
#include <stdlib.h>
2-
#include <tuple>
3-
#include <sstream>
4-
#include <utility>
1+
#include "zmqproducerstatetable.h"
2+
#include "binaryserializer.h"
3+
#include "redisapi.h"
4+
#include "redispipeline.h"
5+
#include "redisreply.h"
6+
#include "table.h"
7+
#include "zmqconsumerstatetable.h"
58
#include <algorithm>
69
#include <chrono>
710
#include <cmath>
11+
#include <sstream>
12+
#include <stdlib.h>
13+
#include <tuple>
14+
#include <utility>
815
#include <zmq.h>
9-
#include "redisreply.h"
10-
#include "table.h"
11-
#include "redisapi.h"
12-
#include "redispipeline.h"
13-
#include "zmqproducerstatetable.h"
14-
#include "zmqconsumerstatetable.h"
15-
#include "binaryserializer.h"
1616

1717
using namespace std;
1818

@@ -164,6 +164,13 @@ void ZmqProducerStateTable::send(const std::vector<KeyOpFieldsValuesTuple> &kcos
164164
}
165165
}
166166

167+
bool ZmqProducerStateTable::wait(const std::string& dbName,
168+
const std::string& tableName,
169+
const std::vector<std::shared_ptr<KeyOpFieldsValuesTuple>>& kcos)
170+
{
171+
return m_zmqClient.wait(dbName, tableName, kcos);
172+
}
173+
167174
size_t ZmqProducerStateTable::dbUpdaterQueueSize()
168175
{
169176
if (m_asyncDBUpdater == nullptr)

‎common/zmqproducerstatetable.h‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -37,6 +37,11 @@ class ZmqProducerStateTable : public ProducerStateTable
3737
// Batched send that can include both SET and DEL requests.
3838
virtual void send(const std::vector<KeyOpFieldsValuesTuple> &kcos);
3939

40+
// To wait for the response from the peer.
41+
virtual bool wait(const std::string& dbName,
42+
const std::string& tableName,
43+
const std::vector<std::shared_ptr<KeyOpFieldsValuesTuple>>& kcos);
44+
4045
size_t dbUpdaterQueueSize();
4146
private:
4247
void initialize(DBConnector *db, const std::string &tableName, bool dbPersistence);

‎common/zmqserver.cpp‎

Lines changed: 39 additions & 29 deletions
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,4 @@
1+
#include <unistd.h>
12
#include <string>
23
#include <deque>
34
#include <limits>
@@ -20,6 +21,7 @@ ZmqServer::ZmqServer(const std::string& endpoint, const std::string& vrf)
2021
: m_endpoint(endpoint),
2122
m_vrf(vrf)
2223
{
24+
connect();
2325
m_buffer.resize(MQ_RESPONSE_MAX_COUNT);
2426
m_runThread = true;
2527
m_mqPollThread = std::make_shared<std::thread>(&ZmqServer::mqPollThread, this);
@@ -31,6 +33,33 @@ ZmqServer::~ZmqServer()
3133
{
3234
m_runThread = false;
3335
m_mqPollThread->join();
36+
37+
zmq_close(m_socket);
38+
zmq_ctx_destroy(m_context);
39+
}
40+
41+
void ZmqServer::connect()
42+
{
43+
SWSS_LOG_ENTER();
44+
m_context = zmq_ctx_new();
45+
m_socket = zmq_socket(m_context, ZMQ_PULL);
46+
47+
// Increase recv buffer for use all bandwidth: http://api.zeromq.org/4-2:zmq-setsockopt
48+
int high_watermark = MQ_WATERMARK;
49+
zmq_setsockopt(m_socket, ZMQ_RCVHWM, &high_watermark, sizeof(high_watermark));
50+
51+
if (!m_vrf.empty())
52+
{
53+
zmq_setsockopt(m_socket, ZMQ_BINDTODEVICE, m_vrf.c_str(), m_vrf.length());
54+
}
55+
56+
int rc = zmq_bind(m_socket, m_endpoint.c_str());
57+
if (rc != 0)
58+
{
59+
SWSS_LOG_THROW("zmq_bind failed on endpoint: %s, zmqerrno: %d",
60+
m_endpoint.c_str(),
61+
zmq_errno());
62+
}
3463
}
3564

3665
void ZmqServer::registerMessageHandler(
@@ -90,40 +119,18 @@ void ZmqServer::mqPollThread()
90119
SWSS_LOG_ENTER();
91120
SWSS_LOG_NOTICE("mqPollThread begin");
92121

93-
// Producer/Consumer state table are n:1 mapping, so need use PUSH/PULL pattern http://api.zeromq.org/master:zmq-socket
94-
void* context = zmq_ctx_new();;
95-
void* socket = zmq_socket(context, ZMQ_PULL);
96-
97-
// Increase recv buffer for use all bandwidth: http://api.zeromq.org/4-2:zmq-setsockopt
98-
int high_watermark = MQ_WATERMARK;
99-
zmq_setsockopt(socket, ZMQ_RCVHWM, &high_watermark, sizeof(high_watermark));
100-
101-
if (!m_vrf.empty())
102-
{
103-
zmq_setsockopt(socket, ZMQ_BINDTODEVICE, m_vrf.c_str(), m_vrf.length());
104-
}
105-
106-
int rc = zmq_bind(socket, m_endpoint.c_str());
107-
if (rc != 0)
108-
{
109-
SWSS_LOG_THROW("zmq_bind failed on endpoint: %s, zmqerrno: %d, message: %s",
110-
m_endpoint.c_str(),
111-
zmq_errno(),
112-
strerror(zmq_errno()));
113-
}
114-
115122
// zmq_poll will use less CPU
116123
zmq_pollitem_t poll_item;
117124
poll_item.fd = 0;
118-
poll_item.socket = socket;
125+
poll_item.socket = m_socket;
119126
poll_item.events = ZMQ_POLLIN;
120127
poll_item.revents = 0;
121128

122129
SWSS_LOG_NOTICE("bind to zmq endpoint: %s", m_endpoint.c_str());
123130
while (m_runThread)
124131
{
125132
// receive message
126-
rc = zmq_poll(&poll_item, 1, 1000);
133+
auto rc = zmq_poll(&poll_item, 1, 1000);
127134
if (rc == 0 || !(poll_item.revents & ZMQ_POLLIN))
128135
{
129136
// timeout or other event
@@ -132,7 +139,7 @@ void ZmqServer::mqPollThread()
132139
}
133140

134141
// receive message
135-
rc = zmq_recv(socket, m_buffer.data(), MQ_RESPONSE_MAX_COUNT, ZMQ_DONTWAIT);
142+
rc = zmq_recv(m_socket, m_buffer.data(), MQ_RESPONSE_MAX_COUNT, ZMQ_DONTWAIT);
136143
if (rc < 0)
137144
{
138145
int zmq_err = zmq_errno();
@@ -160,11 +167,14 @@ void ZmqServer::mqPollThread()
160167
// deserialize and write to redis:
161168
handleReceivedData(m_buffer.data(), rc);
162169
}
163-
164-
zmq_close(socket);
165-
zmq_ctx_destroy(context);
166-
167170
SWSS_LOG_NOTICE("mqPollThread end");
168171
}
169172

173+
// TODO: To be implemented later, required for ZMQ_CLIENT & ZMQ_SERVER
174+
// socket types in response path.
175+
void ZmqServer::sendMsg(
176+
const std::string &dbName, const std::string &tableName,
177+
const std::vector<swss::KeyOpFieldsValuesTuple> &values) {
178+
return;
179+
}
170180
}

‎common/zmqserver.h‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -39,7 +39,13 @@ class ZmqServer
3939
const std::string tableName,
4040
ZmqMessageHandler* handler);
4141

42+
void sendMsg(const std::string& dbName, const std::string& tableName,
43+
const std::vector<swss::KeyOpFieldsValuesTuple>& values);
44+
4245
private:
46+
47+
void connect();
48+
4349
void handleReceivedData(const char* buffer, const size_t size);
4450

4551
void mqPollThread();
@@ -56,6 +62,10 @@ class ZmqServer
5662

5763
std::string m_vrf;
5864

65+
void* m_context;
66+
67+
void* m_socket;
68+
5969
std::map<std::string, std::map<std::string, ZmqMessageHandler*>> m_HandlerMap;
6070
};
6171

‎tests/c_api_ut.cpp‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -395,6 +395,7 @@ TEST(c_api, ZmqConsumerProducerStateTable) {
395395
SWSSZmqProducerStateTable_set(pst, arr.data[i].key, arr.data[i].fieldValues);
396396
else
397397
SWSSZmqClient_sendMsg(cli, "TEST_DB", "mytable", arr);
398+
sleep(2);
398399

399400
SWSSSelectResult result;
400401
SWSSZmqConsumerStateTable_readData(cst, 1500, true, &result);
@@ -434,6 +435,7 @@ TEST(c_api, ZmqConsumerProducerStateTable) {
434435
SWSSZmqProducerStateTable_del(pst, arr.data[i].key);
435436
else
436437
SWSSZmqClient_sendMsg(cli, "TEST_DB", "mytable", arr);
438+
sleep(2);
437439

438440
SWSSZmqConsumerStateTable_readData(cst, 500, true, &result);
439441
SWSSZmqConsumerStateTable_pops(cst, &arr);

0 commit comments

Comments
 (0)