From 36e6a35a886252516f168b90f7a9a7c1c5177312 Mon Sep 17 00:00:00 2001
From: lichao <lichao@aiotlink.com>
Date: 星期六, 08 五月 2021 15:57:01 +0800
Subject: [PATCH] center alloc node queue; node just find them.

---
 src/topic_node.cpp |   56 +++++++++++++++++++++++++++++++++++---------------------
 1 files changed, 35 insertions(+), 21 deletions(-)

diff --git a/src/topic_node.cpp b/src/topic_node.cpp
index 3883062..d8d6a42 100644
--- a/src/topic_node.cpp
+++ b/src/topic_node.cpp
@@ -23,6 +23,9 @@
 using namespace std::chrono;
 using namespace std::chrono_literals;
 
+const char *const kErrMsgNotInit = "BHome node NOT initialized.";
+const char *const kErrMsgNotRegistered = "BHome node NOT registered.";
+
 namespace
 {
 inline void AddRoute(BHMsgHead &head, const MQId id) { head.add_route()->set_mq_id(id); }
@@ -63,14 +66,25 @@
 	MsgI msg;
 	msg.OffsetRef() = ssn_id_;
 	if (ShmMsgQueue::TrySend(shm(), BHInitAddress(), msg)) {
-		sockets_.resize(eSockEnd);
-		for (int i = eSockStart; i < eSockEnd; ++i) {
-			sockets_[i].reset(new ShmSocket(shm_, ssn_id_ + i, kMqLen));
+
+		auto end_time = steady_clock::now() + 3s;
+		do {
+			try {
+				for (int i = eSockStart; i < eSockEnd; ++i) {
+					sockets_.emplace_back(new ShmSocket(shm_, false, ssn_id_ + i, kMqLen));
+				}
+				break;
+			} catch (...) {
+				sockets_.clear();
+				std::this_thread::sleep_for(100ms);
+			}
+		} while (steady_clock::now() < end_time);
+		if (!sockets_.empty()) {
+			// recv msgs to avoid memory leak.
+			auto default_ignore_msg = [](ShmSocket &sock, MsgI &imsg, BHMsgHead &head) { return true; };
+			SockNode().Start(default_ignore_msg);
+			return true;
 		}
-		// recv msgs to avoid memory leak.
-		auto default_ignore_msg = [](ShmSocket &sock, MsgI &imsg, BHMsgHead &head) { return true; };
-		SockNode().Start(default_ignore_msg);
-		return true;
 	}
 	return false;
 }
@@ -78,7 +92,7 @@
 void TopicNode::Start(ServerAsyncCB const &server_cb, SubDataCB const &sub_cb, RequestResultCB &client_cb, int nworker)
 {
 	if (!Init()) {
-		SetLastError(eError, "BHome Node Not Inited.");
+		SetLastError(eError, kErrMsgNotInit);
 		return;
 	}
 	if (nworker < 1) {
@@ -101,7 +115,7 @@
 bool TopicNode::Register(ProcInfo &proc, MsgCommonReply &reply_body, const int timeout_ms)
 {
 	if (!Init()) {
-		SetLastError(eError, "BHome Node Not Inited.");
+		SetLastError(eError, kErrMsgNotInit);
 		return false;
 	}
 
@@ -151,7 +165,7 @@
 bool TopicNode::Unregister(ProcInfo &proc, MsgCommonReply &reply_body, const int timeout_ms)
 {
 	if (!IsOnline()) {
-		SetLastError(eNotRegistered, "Not Registered.");
+		SetLastError(eNotRegistered, kErrMsgNotRegistered);
 		return false;
 	}
 
@@ -190,7 +204,7 @@
 bool TopicNode::Heartbeat(ProcInfo &proc, MsgCommonReply &reply_body, const int timeout_ms)
 {
 	if (!IsOnline()) {
-		SetLastError(eNotRegistered, "Not Registered.");
+		SetLastError(eNotRegistered, kErrMsgNotRegistered);
 		return false;
 	}
 
@@ -223,7 +237,7 @@
 bool TopicNode::QueryTopicAddress(BHAddress &dest, MsgQueryTopic &query, MsgQueryTopicReply &reply_body, const int timeout_ms)
 {
 	if (!IsOnline()) {
-		SetLastError(eNotRegistered, "Not Registered.");
+		SetLastError(eNotRegistered, kErrMsgNotRegistered);
 		return false;
 	}
 	auto &sock = SockNode();
@@ -242,7 +256,7 @@
 bool TopicNode::ServerRegisterRPC(MsgTopicList &topics, MsgCommonReply &reply_body, const int timeout_ms)
 {
 	if (!IsOnline()) {
-		SetLastError(eNotRegistered, "Not Registered.");
+		SetLastError(eNotRegistered, kErrMsgNotRegistered);
 		return false;
 	}
 
@@ -309,7 +323,7 @@
 bool TopicNode::ServerRecvRequest(void *&src_info, std::string &proc_id, MsgRequestTopic &request, const int timeout_ms)
 {
 	if (!IsOnline()) {
-		SetLastError(eNotRegistered, "Not Registered.");
+		SetLastError(eNotRegistered, kErrMsgNotRegistered);
 		return false;
 	}
 
@@ -333,7 +347,7 @@
 bool TopicNode::ServerSendReply(void *src_info, const MsgRequestTopicReply &body)
 {
 	if (!IsOnline()) {
-		SetLastError(eNotRegistered, "Not Registered.");
+		SetLastError(eNotRegistered, kErrMsgNotRegistered);
 		return false;
 	}
 
@@ -371,7 +385,7 @@
 bool TopicNode::ClientAsyncRequest(const BHAddress &remote_addr, const MsgRequestTopic &req, std::string &out_msg_id, const RequestResultCB &cb)
 {
 	if (!IsOnline()) {
-		SetLastError(eNotRegistered, "Not Registered.");
+		SetLastError(eNotRegistered, kErrMsgNotRegistered);
 		return false;
 	}
 
@@ -412,7 +426,7 @@
 bool TopicNode::ClientSyncRequest(const BHAddress &remote_addr, const MsgRequestTopic &request, std::string &out_proc_id, MsgRequestTopicReply &out_reply, const int timeout_ms)
 {
 	if (!IsOnline()) {
-		SetLastError(eNotRegistered, "Not Registered.");
+		SetLastError(eNotRegistered, kErrMsgNotRegistered);
 		return false;
 	}
 
@@ -463,7 +477,7 @@
 bool TopicNode::ClientQueryRPCTopic(const Topic &topic, BHAddress &addr, const int timeout_ms)
 {
 	if (!IsOnline()) {
-		SetLastError(eNotRegistered, "Not Registered.");
+		SetLastError(eNotRegistered, kErrMsgNotRegistered);
 		return false;
 	}
 
@@ -487,7 +501,7 @@
 bool TopicNode::Publish(const MsgPublish &pub, const int timeout_ms)
 {
 	if (!IsOnline()) {
-		SetLastError(eNotRegistered, "Not Registered.");
+		SetLastError(eNotRegistered, kErrMsgNotRegistered);
 		return false;
 	}
 
@@ -518,7 +532,7 @@
 bool TopicNode::Subscribe(MsgTopicList &topics, MsgCommonReply &reply_body, const int timeout_ms)
 {
 	if (!IsOnline()) {
-		SetLastError(eNotRegistered, "Not Registered.");
+		SetLastError(eNotRegistered, kErrMsgNotRegistered);
 		return false;
 	}
 
@@ -567,7 +581,7 @@
 bool TopicNode::RecvSub(std::string &proc_id, MsgPublish &pub, const int timeout_ms)
 {
 	if (!IsOnline()) {
-		SetLastError(eNotRegistered, "Not Registered.");
+		SetLastError(eNotRegistered, kErrMsgNotRegistered);
 		return false;
 	}
 

--
Gitblit v1.8.0