From db322f33ba13592f2492317e3f1a070454c97059 Mon Sep 17 00:00:00 2001
From: lichao <lichao@aiotlink.com>
Date: 星期四, 13 五月 2021 19:34:46 +0800
Subject: [PATCH] center alloc all msgs.

---
 src/topic_node.cpp |  170 ++++++++++++++++++++++++++++++++++++++++++++------------
 1 files changed, 132 insertions(+), 38 deletions(-)

diff --git a/src/topic_node.cpp b/src/topic_node.cpp
index 9398318..35228b4 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); }
@@ -37,47 +40,133 @@
 } // namespace
 
 TopicNode::TopicNode(SharedMemory &shm) :
-    shm_(shm), sockets_(eSockEnd), state_(eStateUnregistered)
+    shm_(shm), 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; };
-	for (auto &p : sockets_) {
-		p->Start(default_ignore_msg);
-	}
 }
 
 TopicNode::~TopicNode()
 {
+	LOG_DEBUG() << "~TopicNode()";
 	Stop();
-	SockNode().Stop();
-	if (state() == eStateUnregistered) {
-		for (auto &p : sockets_) { p->Remove(); }
+}
+
+bool TopicNode::Init()
+{
+	std::lock_guard<std::mutex> lk(mutex_);
+
+	if (Valid()) {
+		return true;
+	} else if (info_.proc_id().empty()) {
+		return false;
 	}
+
+	if (ssn_id_ == 0) {
+		ssn_id_ = ShmMsgQueue::NewId();
+	}
+	LOG_DEBUG() << "Node Init, id " << ssn_id_;
+	auto NodeInit = [&]() {
+		auto SendInitCmd = [&]() {
+			int64_t init_cmd = ssn_id_ << 4 | EncodeCmd(eCmdNodeInit);
+			auto end_time = steady_clock::now() + 3s;
+			bool r = false;
+			do {
+				r = ShmMsgQueue::TrySend(shm(), BHTopicCenterAddress(), init_cmd);
+			} while (!r && steady_clock::now() < end_time);
+			return r;
+		};
+		if (SendInitCmd()) {
+			LOG_DEBUG() << "node send init ok";
+			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()) {
+		NodeInit();
+	}
+	if (!sockets_.empty()) {
+		LOG_DEBUG() << "node sockets ok";
+		auto onNodeCmd = [this](ShmSocket &socket, int64_t &val) {
+			LOG_DEBUG() << "node recv cmd: " << DecodeCmd(val);
+			switch (DecodeCmd(val)) {
+			case eCmdNodeInitReply: {
+				MsgI msg(val >> 4);
+				DEFER1(msg.Release());
+				MsgProcInit body;
+				auto head = InitMsgHead(GetType(body), info_.proc_id(), ssn_id_);
+				head.add_route()->set_mq_id(ssn_id_);
+				if (msg.Fill(head, body)) {
+					socket.Send(BHTopicCenterAddress(), msg);
+				}
+			} break;
+			default:
+				break;
+			}
+			return true;
+		};
+
+		// recv msgs to avoid memory leak.
+		auto onMsg = [this](ShmSocket &sock, MsgI &imsg, BHMsgHead &head) {
+			LOG_DEBUG() << "node recv type: " << head.type();
+			if (head.type() == kMsgTypeProcInitReply) {
+				LOG_DEBUG() << "got proc init reply";
+				MsgProcInitReply reply;
+				if (imsg.ParseBody(reply)) {
+					SetProcIndex(reply.proc_index());
+				}
+			}
+			return true;
+		};
+		SockNode().Start(1, onMsg, onNodeCmd);
+		LOG_DEBUG() << "sockets ok.";
+		return true;
+	}
+	return false;
 }
 
 void TopicNode::Start(ServerAsyncCB const &server_cb, SubDataCB const &sub_cb, RequestResultCB &client_cb, int nworker)
 {
+	if (!Init()) {
+		SetLastError(eError, kErrMsgNotInit);
+		return;
+	}
 	if (nworker < 1) {
 		nworker = 1;
 	} else if (nworker > 16) {
 		nworker = 16;
 	}
-	SockNode().Start();
+	// SockNode().Start();
 	ServerStart(server_cb, nworker);
 	SubscribeStartWorker(sub_cb, nworker);
 	ClientStartWorker(client_cb, nworker);
 }
 void TopicNode::Stop()
 {
+	LOG_DEBUG() << "Node Stopping";
 	for (auto &p : sockets_) { p->Stop(); }
+	LOG_INFO() << "Node Stopped";
 }
 
 bool TopicNode::Register(ProcInfo &proc, MsgCommonReply &reply_body, const int timeout_ms)
 {
-	info_ = proc;
+	{
+		std::lock_guard<std::mutex> lk(mutex_);
+		info_ = proc;
+	}
+
+	if (!Init()) {
+		SetLastError(eError, kErrMsgNotInit);
+		return false;
+	}
 
 	auto &sock = SockNode();
 	MsgRegister body;
@@ -89,7 +178,7 @@
 	AddId(SockSub().id());
 	AddId(SockPub().id());
 
-	auto head(InitMsgHead(GetType(body), body.proc().proc_id()));
+	auto head(InitMsgHead(GetType(body), body.proc().proc_id(), ssn()));
 	AddRoute(head, sock.id());
 
 	auto CheckResult = [this](MsgI &msg, BHMsgHead &head, MsgCommonReply &rbody) {
@@ -122,6 +211,11 @@
 }
 bool TopicNode::Unregister(ProcInfo &proc, MsgCommonReply &reply_body, const int timeout_ms)
 {
+	if (!IsOnline()) {
+		SetLastError(eNotRegistered, kErrMsgNotRegistered);
+		return false;
+	}
+
 	info_.Clear();
 	state_cas(eStateOnline, eStateOffline);
 
@@ -129,7 +223,7 @@
 	MsgUnregister body;
 	body.mutable_proc()->Swap(&proc);
 
-	auto head(InitMsgHead(GetType(body), body.proc().proc_id()));
+	auto head(InitMsgHead(GetType(body), body.proc().proc_id(), ssn()));
 	AddRoute(head, sock.id());
 
 	auto CheckResult = [this](MsgI &msg, BHMsgHead &head, MsgCommonReply &rbody) {
@@ -157,7 +251,7 @@
 bool TopicNode::Heartbeat(ProcInfo &proc, MsgCommonReply &reply_body, const int timeout_ms)
 {
 	if (!IsOnline()) {
-		SetLastError(eNotRegistered, "Not Registered.");
+		SetLastError(eNotRegistered, kErrMsgNotRegistered);
 		return false;
 	}
 
@@ -165,7 +259,7 @@
 	MsgHeartbeat body;
 	body.mutable_proc()->Swap(&proc);
 
-	auto head(InitMsgHead(GetType(body), body.proc().proc_id()));
+	auto head(InitMsgHead(GetType(body), body.proc().proc_id(), ssn()));
 	AddRoute(head, sock.id());
 
 	if (timeout_ms == 0) {
@@ -190,12 +284,12 @@
 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();
 
-	BHMsgHead head(InitMsgHead(GetType(query), proc_id()));
+	BHMsgHead head(InitMsgHead(GetType(query), proc_id(), ssn()));
 	AddRoute(head, sock.id());
 
 	MsgI reply;
@@ -209,7 +303,7 @@
 bool TopicNode::ServerRegisterRPC(MsgTopicList &topics, MsgCommonReply &reply_body, const int timeout_ms)
 {
 	if (!IsOnline()) {
-		SetLastError(eNotRegistered, "Not Registered.");
+		SetLastError(eNotRegistered, kErrMsgNotRegistered);
 		return false;
 	}
 
@@ -217,7 +311,7 @@
 	MsgRegisterRPC body;
 	body.mutable_topics()->Swap(&topics);
 
-	auto head(InitMsgHead(GetType(body), proc_id()));
+	auto head(InitMsgHead(GetType(body), proc_id(), ssn()));
 	AddRoute(head, sock.id());
 
 	if (timeout_ms == 0) {
@@ -242,7 +336,7 @@
 
 		MsgRequestTopicReply reply_body;
 		if (rcb(head.proc_id(), req, reply_body)) {
-			BHMsgHead reply_head(InitMsgHead(GetType(reply_body), proc_id(), head.msg_id()));
+			BHMsgHead reply_head(InitMsgHead(GetType(reply_body), proc_id(), ssn(), head.msg_id()));
 
 			for (int i = 0; i < head.route_size() - 1; ++i) {
 				reply_head.add_route()->Swap(head.mutable_route(i));
@@ -276,7 +370,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;
 	}
 
@@ -300,7 +394,7 @@
 bool TopicNode::ServerSendReply(void *src_info, const MsgRequestTopicReply &body)
 {
 	if (!IsOnline()) {
-		SetLastError(eNotRegistered, "Not Registered.");
+		SetLastError(eNotRegistered, kErrMsgNotRegistered);
 		return false;
 	}
 
@@ -311,7 +405,7 @@
 	if (!p || p->route.empty()) {
 		return false;
 	}
-	BHMsgHead head(InitMsgHead(GetType(body), proc_id(), p->msg_id));
+	BHMsgHead head(InitMsgHead(GetType(body), proc_id(), ssn(), p->msg_id));
 	for (unsigned i = 0; i < p->route.size() - 1; ++i) {
 		head.add_route()->Swap(&p->route[i]);
 	}
@@ -338,7 +432,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;
 	}
 
@@ -348,7 +442,7 @@
 
 	auto SendTo = [this, msg_id](const BHAddress &addr, const MsgRequestTopic &req, const RequestResultCB &cb) {
 		auto &sock = SockClient();
-		BHMsgHead head(InitMsgHead(GetType(req), proc_id(), msg_id));
+		BHMsgHead head(InitMsgHead(GetType(req), proc_id(), ssn(), msg_id));
 		AddRoute(head, sock.id());
 		head.set_topic(req.topic());
 
@@ -379,7 +473,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;
 	}
 
@@ -388,7 +482,8 @@
 
 		BHAddress addr;
 		if (ClientQueryRPCTopic(request.topic(), addr, timeout_ms)) {
-			BHMsgHead head(InitMsgHead(GetType(request), proc_id()));
+			LOG_TRACE() << "node: " << SockNode().id() << ", topic dest: " << addr.mq_id();
+			BHMsgHead head(InitMsgHead(GetType(request), proc_id(), ssn()));
 			AddRoute(head, sock.id());
 			head.set_topic(request.topic());
 
@@ -402,8 +497,6 @@
 				reply_head.mutable_proc_id()->swap(out_proc_id);
 				return true;
 			}
-		} else {
-			SetLastError(eNotFound, "remote not found.");
 		}
 	} catch (...) {
 		SetLastError(eError, __func__ + std::string(" internal errer."));
@@ -431,7 +524,7 @@
 bool TopicNode::ClientQueryRPCTopic(const Topic &topic, BHAddress &addr, const int timeout_ms)
 {
 	if (!IsOnline()) {
-		SetLastError(eNotRegistered, "Not Registered.");
+		SetLastError(eNotRegistered, kErrMsgNotRegistered);
 		return false;
 	}
 
@@ -446,6 +539,7 @@
 			return true;
 		}
 	}
+	SetLastError(eNotFound, "remote not found.");
 	return false;
 }
 
@@ -454,13 +548,13 @@
 bool TopicNode::Publish(const MsgPublish &pub, const int timeout_ms)
 {
 	if (!IsOnline()) {
-		SetLastError(eNotRegistered, "Not Registered.");
+		SetLastError(eNotRegistered, kErrMsgNotRegistered);
 		return false;
 	}
 
 	try {
 		auto &sock = SockPub();
-		BHMsgHead head(InitMsgHead(GetType(pub), proc_id()));
+		BHMsgHead head(InitMsgHead(GetType(pub), proc_id(), ssn()));
 		AddRoute(head, sock.id());
 
 		if (timeout_ms == 0) {
@@ -485,7 +579,7 @@
 bool TopicNode::Subscribe(MsgTopicList &topics, MsgCommonReply &reply_body, const int timeout_ms)
 {
 	if (!IsOnline()) {
-		SetLastError(eNotRegistered, "Not Registered.");
+		SetLastError(eNotRegistered, kErrMsgNotRegistered);
 		return false;
 	}
 
@@ -494,7 +588,7 @@
 		MsgSubscribe sub;
 		sub.mutable_topics()->Swap(&topics);
 
-		BHMsgHead head(InitMsgHead(GetType(sub), proc_id()));
+		BHMsgHead head(InitMsgHead(GetType(sub), proc_id(), ssn()));
 		AddRoute(head, sock.id());
 		if (timeout_ms == 0) {
 			return sock.Send(BHTopicBusAddress(), head, sub);
@@ -534,7 +628,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