From 3931f83205f153f2bc7fc36d1a894cdc3f14b4db Mon Sep 17 00:00:00 2001
From: lichao <lichao@aiotlink.com>
Date: 星期三, 21 四月 2021 16:52:51 +0800
Subject: [PATCH] change node socket to vector; try lock free queue.
---
src/topic_node.cpp | 23 ++++++++++-------------
1 files changed, 10 insertions(+), 13 deletions(-)
diff --git a/src/topic_node.cpp b/src/topic_node.cpp
index 70247dd..8bf9cf8 100644
--- a/src/topic_node.cpp
+++ b/src/topic_node.cpp
@@ -37,14 +37,16 @@
} // namespace
TopicNode::TopicNode(SharedMemory &shm) :
- shm_(shm), sock_node_(shm), sock_client_(shm, kMqLen), sock_server_(shm, kMqLen), sock_sub_(shm, kMqLen), state_(eStateUnregistered)
+ shm_(shm), sockets_(eSockEnd), state_(eStateUnregistered)
{
+ for (int i = eSockStart; i < eSockEnd; ++i) {
+ sockets_[i].reset(new ShmSocket(shm_, kMqLen));
+ }
// recv msgs to avoid memory leak.
auto default_ignore_msg = [](ShmSocket &sock, MsgI &imsg, BHMsgHead &head) { return true; };
- SockNode().Start(default_ignore_msg);
- SockClient().Start(default_ignore_msg);
- SockServer().Start(default_ignore_msg);
- SockSub().Start(default_ignore_msg);
+ for (auto &p : sockets_) {
+ p->Start(default_ignore_msg);
+ }
}
TopicNode::~TopicNode()
@@ -52,10 +54,7 @@
Stop();
SockNode().Stop();
if (state() == eStateUnregistered) {
- SockNode().Remove();
- SockClient().Remove();
- SockServer().Remove();
- SockSub().Remove();
+ for (auto &p : sockets_) { p->Remove(); }
}
}
@@ -66,16 +65,14 @@
} else if (nworker > 16) {
nworker = 16;
}
-
+ SockNode().Start();
ServerStart(server_cb, nworker);
SubscribeStartWorker(sub_cb, nworker);
ClientStartWorker(client_cb, nworker);
}
void TopicNode::Stop()
{
- SockSub().Stop();
- SockServer().Stop();
- SockClient().Stop();
+ for (auto &p : sockets_) { p->Stop(); }
}
bool TopicNode::Register(ProcInfo &proc, MsgCommonReply &reply_body, const int timeout_ms)
--
Gitblit v1.8.0