Skip to content

Add lazy binding support to ZmqServer. #1068

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
qiluo-msft merged 9 commits into from
Aug 21, 2025
Merged

Add lazy binding support to ZmqServer. #1068

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
cba1312
Add lazzy bind method
mssonicbld Aug 18, 2025
036cfd1
Improve code
mssonicbld Aug 18, 2025
0ae3034
Add test case
mssonicbld Aug 18, 2025
c06701c
Update code
mssonicbld Aug 18, 2025
ea29aab
Fix code
mssonicbld Aug 18, 2025
dc0483c
Fix testcase
mssonicbld Aug 18, 2025
7ea480d
Update zmqserver.cpp
liuh-80 Aug 18, 2025
93d5d01
Improve code
mssonicbld Aug 18, 2025
d31ebf0
Improve code
mssonicbld Aug 18, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
56 changes: 45 additions & 11 deletions common/zmqserver.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,34 +13,57 @@ using namespace std;
namespace swss {

ZmqServer::ZmqServer(const std::string& endpoint)
: ZmqServer(endpoint, "")
: ZmqServer(endpoint, "", false)
{
}

ZmqServer::ZmqServer(const std::string& endpoint, const std::string& vrf)
: m_endpoint(endpoint),
m_vrf(vrf)
: ZmqServer(endpoint, vrf, false)
{
connect();
m_buffer.resize(MQ_RESPONSE_MAX_COUNT);
m_runThread = true;
m_mqPollThread = std::make_shared<std::thread>(&ZmqServer::mqPollThread, this);
}

ZmqServer::ZmqServer(const std::string& endpoint, const std::string& vrf, bool lazyBind)
: m_mqPollThread(nullptr),
m_endpoint(endpoint),
m_vrf(vrf),
m_context(nullptr),
m_socket(nullptr)
{
if (!lazyBind)
{
bind();
}

SWSS_LOG_DEBUG("ZmqServer ctor endpoint: %s", endpoint.c_str());
}

ZmqServer::~ZmqServer()
{
m_runThread = false;
m_mqPollThread->join();
if (m_mqPollThread)
{
m_mqPollThread->join();
}

if (m_socket)
{
zmq_close(m_socket);
}

zmq_close(m_socket);
zmq_ctx_destroy(m_context);
if (m_context)
{
zmq_ctx_destroy(m_context);
}
}

void ZmqServer::connect()
void ZmqServer::bind()
{
SWSS_LOG_ENTER();
if (m_socket)
{
SWSS_LOG_THROW("ZmqServer has already been bound to the endpoint: %s", m_endpoint.c_str());
Copy link
Copy Markdown
Contributor

@r12f r12f Aug 19, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

do we need to consider this bind as no-op instead of throw?

Copy link
Copy Markdown
Contributor Author

@liuh-80 liuh-80 Aug 20, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

A ZMQ server should only be bound once.
To catch potential issues during development, I recommend throwing an exception if a second bind attempt is detected. This will help identify incorrect usage early and ensure proper lifecycle management of the server.

Copy link
Copy Markdown
Contributor

@r12f r12f Aug 20, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

gotcha! so it is an explicit lazy binding instead of implicit one. makes sense!

}

m_context = zmq_ctx_new();
m_socket = zmq_socket(m_context, ZMQ_PULL);

Expand All @@ -60,6 +83,10 @@ void ZmqServer::connect()
m_endpoint.c_str(),
zmq_errno());
}

SWSS_LOG_DEBUG("ZmqServer bind to endpoint: %s", m_endpoint.c_str());

startMqPollThread();
}

void ZmqServer::registerMessageHandler(
Expand Down Expand Up @@ -114,6 +141,13 @@ void ZmqServer::handleReceivedData(const char* buffer, const size_t size)
handler->handleReceivedData(kcos);
}

void ZmqServer::startMqPollThread()
{
m_buffer.resize(MQ_RESPONSE_MAX_COUNT);
m_runThread = true;
m_mqPollThread = std::make_shared<std::thread>(&ZmqServer::mqPollThread, this);
}

void ZmqServer::mqPollThread()
{
SWSS_LOG_ENTER();
Expand Down
8 changes: 5 additions & 3 deletions common/zmqserver.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,7 @@ class ZmqServer

ZmqServer(const std::string& endpoint);
ZmqServer(const std::string& endpoint, const std::string& vrf);
ZmqServer(const std::string& endpoint, const std::string& vrf, bool lazyBind);
~ZmqServer();

void registerMessageHandler(
Expand All @@ -42,12 +43,13 @@ class ZmqServer
void sendMsg(const std::string& dbName, const std::string& tableName,
const std::vector<swss::KeyOpFieldsValuesTuple>& values);

private:

void connect();
void bind();

private:
void handleReceivedData(const char* buffer, const size_t size);

void startMqPollThread();

void mqPollThread();

ZmqMessageHandler* findMessageHandler(const std::string dbName, const std::string tableName);
Expand Down
43 changes: 43 additions & 0 deletions tests/zmq_state_ut.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -569,3 +569,46 @@ TEST(ZmqWithResponseClientError, test)
// Wait will timeout without server reply.
EXPECT_FALSE(p.wait(dbName, tableName, kcosPtr));
}

TEST(ZmqServerLazzyBind, test)
{
std::string testTableName = "ZMQ_PROD_CONS_UT";
std::string pushEndpoint = "tcp://localhost:1234";
std::string pullEndpoint = "tcp://*:1234";
DBConnector db(TEST_DB, 0, true);
ZmqClient client(pushEndpoint, 3000);
ZmqProducerStateTable p(&db, testTableName, client, true);
std::vector<KeyOpFieldsValuesTuple> kcos;
auto testKey = "testkey";
kcos.push_back(KeyOpFieldsValuesTuple{testKey, SET_COMMAND, std::vector<FieldValueTuple>{}});
std::vector<std::shared_ptr<KeyOpFieldsValuesTuple>> kcosPtr;
p.send(kcos);

// initialize ZMQ server with lazzy bind
DBConnector server_db(TEST_DB, 0, true);
ZmqServer server(pullEndpoint, "", true);
ZmqConsumerStateTable c(&db, testTableName, server, 128, 0, false);
server.bind();

std::deque<KeyOpFieldsValuesTuple> vkco;
int received = 0;
while (received < 1)
{
c.pops(vkco);
while (!vkco.empty())
{
auto &kco = vkco.front();
auto key = kfvKey(kco);
auto op = kfvOp(kco);
auto fvs = kfvFieldsValues(kco);

EXPECT_EQ(key, testKey);

received += 1;
vkco.pop_front();
}
}

EXPECT_EQ(received, 1);
}

Loading