From cab831748a2a9cc18b7f18f3b5e14a4374b7ab68 Mon Sep 17 00:00:00 2001
From: lichao <lichao@aiotlink.com>
Date: 星期一, 17 五月 2021 18:34:26 +0800
Subject: [PATCH] socket send using abs addr, avoid shm find by id.

---
 src/topic_node.cpp |  209 ++++++++++++++++++++++++++++++++++++---------------
 1 files changed, 147 insertions(+), 62 deletions(-)

diff --git a/src/topic_node.cpp b/src/topic_node.cpp
index 3883062..51a0ab7 100644
--- a/src/topic_node.cpp
+++ b/src/topic_node.cpp
@@ -23,9 +23,17 @@
 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); }
+inline void AddRoute(BHMsgHead &head, const ShmSocket &sock)
+{
+	auto route = head.add_route();
+	route->set_mq_id(sock.id());
+	route->set_abs_addr(sock.AbsAddr());
+}
 
 struct SrcInfo {
 	std::vector<BHAddress> route;
@@ -37,9 +45,8 @@
 } // namespace
 
 TopicNode::TopicNode(SharedMemory &shm) :
-    shm_(shm), state_(eStateUnregistered)
+    shm_(shm), state_(eStateUninited)
 {
-	Init();
 }
 
 TopicNode::~TopicNode()
@@ -54,22 +61,78 @@
 
 	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_;
-	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 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 {
+					//TODO recv offset, avoid query.
+					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()) {
+		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_);
+				AddRoute(head, socket);
+				if (msg.Fill(head, body)) {
+					socket.Send(BHTopicCenterAddress(), msg);
+				}
+			} break;
+			default:
+				break;
+			}
+			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);
+		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());
+					this->state_ = eStateUnregistered;
+				}
+			}
+			return true;
+		};
+		SockNode().Start(1, onMsg, onNodeCmd);
 		return true;
 	}
 	return false;
@@ -78,7 +141,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) {
@@ -86,7 +149,7 @@
 	} else if (nworker > 16) {
 		nworker = 16;
 	}
-	SockNode().Start();
+	// SockNode().Start();
 	ServerStart(server_cb, nworker);
 	SubscribeStartWorker(sub_cb, nworker);
 	ClientStartWorker(client_cb, nworker);
@@ -100,25 +163,31 @@
 
 bool TopicNode::Register(ProcInfo &proc, MsgCommonReply &reply_body, const int timeout_ms)
 {
-	if (!Init()) {
-		SetLastError(eError, "BHome Node Not Inited.");
-		return false;
+	{
+		std::lock_guard<std::mutex> lk(mutex_);
+		info_ = proc;
 	}
 
-	info_ = proc;
+	if (!Init()) {
+		SetLastError(eError, kErrMsgNotInit);
+		return false;
+	}
+	auto end_time = steady_clock::now() + milliseconds(timeout_ms);
+
+	while (state_ != eStateUnregistered && steady_clock::now() < end_time) {
+		std::this_thread::yield();
+	}
+	if (state_ != eStateUnregistered) {
+		SetLastError(eError, kErrMsgNotInit);
+		return false;
+	}
 
 	auto &sock = SockNode();
 	MsgRegister body;
 	body.mutable_proc()->Swap(&proc);
-	auto AddId = [&](const MQId id) { body.add_addrs()->set_mq_id(id); };
-	AddId(SockNode().id());
-	AddId(SockServer().id());
-	AddId(SockClient().id());
-	AddId(SockSub().id());
-	AddId(SockPub().id());
 
 	auto head(InitMsgHead(GetType(body), body.proc().proc_id(), ssn()));
-	AddRoute(head, sock.id());
+	AddRoute(head, sock);
 
 	auto CheckResult = [this](MsgI &msg, BHMsgHead &head, MsgCommonReply &rbody) {
 		bool ok = head.type() == kMsgTypeCommonReply &&
@@ -151,7 +220,7 @@
 bool TopicNode::Unregister(ProcInfo &proc, MsgCommonReply &reply_body, const int timeout_ms)
 {
 	if (!IsOnline()) {
-		SetLastError(eNotRegistered, "Not Registered.");
+		SetLastError(eNotRegistered, kErrMsgNotRegistered);
 		return false;
 	}
 
@@ -163,7 +232,7 @@
 	body.mutable_proc()->Swap(&proc);
 
 	auto head(InitMsgHead(GetType(body), body.proc().proc_id(), ssn()));
-	AddRoute(head, sock.id());
+	AddRoute(head, sock);
 
 	auto CheckResult = [this](MsgI &msg, BHMsgHead &head, MsgCommonReply &rbody) {
 		bool r = head.type() == kMsgTypeCommonReply &&
@@ -190,7 +259,7 @@
 bool TopicNode::Heartbeat(ProcInfo &proc, MsgCommonReply &reply_body, const int timeout_ms)
 {
 	if (!IsOnline()) {
-		SetLastError(eNotRegistered, "Not Registered.");
+		SetLastError(eNotRegistered, kErrMsgNotRegistered);
 		return false;
 	}
 
@@ -199,7 +268,7 @@
 	body.mutable_proc()->Swap(&proc);
 
 	auto head(InitMsgHead(GetType(body), body.proc().proc_id(), ssn()));
-	AddRoute(head, sock.id());
+	AddRoute(head, sock);
 
 	if (timeout_ms == 0) {
 		return sock.Send(BHTopicCenterAddress(), head, body);
@@ -223,13 +292,13 @@
 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(), ssn()));
-	AddRoute(head, sock.id());
+	AddRoute(head, sock);
 
 	MsgI reply;
 	DEFER1(reply.Release());
@@ -242,7 +311,7 @@
 bool TopicNode::ServerRegisterRPC(MsgTopicList &topics, MsgCommonReply &reply_body, const int timeout_ms)
 {
 	if (!IsOnline()) {
-		SetLastError(eNotRegistered, "Not Registered.");
+		SetLastError(eNotRegistered, kErrMsgNotRegistered);
 		return false;
 	}
 
@@ -251,7 +320,7 @@
 	body.mutable_topics()->Swap(&topics);
 
 	auto head(InitMsgHead(GetType(body), proc_id(), ssn()));
-	AddRoute(head, sock.id());
+	AddRoute(head, sock);
 
 	if (timeout_ms == 0) {
 		return sock.Send(BHTopicCenterAddress(), head, body);
@@ -280,7 +349,7 @@
 			for (int i = 0; i < head.route_size() - 1; ++i) {
 				reply_head.add_route()->Swap(head.mutable_route(i));
 			}
-			auto remote = head.route().rbegin()->mq_id();
+			MQInfo remote = {head.route().rbegin()->mq_id(), head.route().rbegin()->abs_addr()};
 			sock.Send(remote, reply_head, reply_body);
 		}
 	};
@@ -296,10 +365,17 @@
 		MsgRequestTopic req;
 		if (!imsg.ParseBody(req)) { return; }
 
-		SrcInfo *p = new SrcInfo;
-		p->route.assign(head.route().begin(), head.route().end());
-		p->msg_id = head.msg_id();
-		acb(p, *head.mutable_proc_id(), req);
+		try {
+			SrcInfo *p = new SrcInfo;
+			if (!p) {
+				throw std::runtime_error("no memory.");
+			}
+			p->route.assign(head.route().begin(), head.route().end());
+			p->msg_id = head.msg_id();
+			acb(p, *head.mutable_proc_id(), req);
+		} catch (std::exception &e) {
+			LOG_ERROR() << "error server handle msg:" << e.what();
+		}
 	};
 
 	auto &sock = SockServer();
@@ -309,7 +385,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;
 	}
 
@@ -320,11 +396,19 @@
 	if (sock.SyncRecv(imsg, head, timeout_ms) && head.type() == kMsgTypeRequestTopic) {
 		if (imsg.ParseBody(request)) {
 			head.mutable_proc_id()->swap(proc_id);
-			SrcInfo *p = new SrcInfo;
-			p->route.assign(head.route().begin(), head.route().end());
-			p->msg_id = head.msg_id();
-			src_info = p;
-			return true;
+			try {
+				SrcInfo *p = new SrcInfo;
+				if (!p) {
+					throw std::runtime_error("no memory.");
+				}
+				p->route.assign(head.route().begin(), head.route().end());
+				p->msg_id = head.msg_id();
+				src_info = p;
+				return true;
+			} catch (std::exception &e) {
+				LOG_ERROR() << "error recv request: " << e.what();
+				return false;
+			}
 		}
 	}
 	return false;
@@ -333,7 +417,7 @@
 bool TopicNode::ServerSendReply(void *src_info, const MsgRequestTopicReply &body)
 {
 	if (!IsOnline()) {
-		SetLastError(eNotRegistered, "Not Registered.");
+		SetLastError(eNotRegistered, kErrMsgNotRegistered);
 		return false;
 	}
 
@@ -348,7 +432,8 @@
 	for (unsigned i = 0; i < p->route.size() - 1; ++i) {
 		head.add_route()->Swap(&p->route[i]);
 	}
-	return sock.Send(p->route.back().mq_id(), head, body);
+	MQInfo dest = {p->route.back().mq_id(), p->route.back().abs_addr()};
+	return sock.Send(dest, head, body);
 }
 
 bool TopicNode::ClientStartWorker(RequestResultCB const &cb, const int nworker)
@@ -371,7 +456,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;
 	}
 
@@ -379,10 +464,10 @@
 
 	out_msg_id = msg_id;
 
-	auto SendTo = [this, msg_id](const BHAddress &addr, const MsgRequestTopic &req, const RequestResultCB &cb) {
+	auto SendTo = [this, msg_id](const MQInfo &remote, const MsgRequestTopic &req, const RequestResultCB &cb) {
 		auto &sock = SockClient();
 		BHMsgHead head(InitMsgHead(GetType(req), proc_id(), ssn(), msg_id));
-		AddRoute(head, sock.id());
+		AddRoute(head, sock);
 		head.set_topic(req.topic());
 
 		if (cb) {
@@ -394,15 +479,15 @@
 					}
 				}
 			};
-			return sock.Send(addr.mq_id(), head, req, onRecv);
+			return sock.Send(remote, head, req, onRecv);
 		} else {
-			return sock.Send(addr.mq_id(), head, req);
+			return sock.Send(remote, head, req);
 		}
 	};
 
 	try {
 		BHAddress addr;
-		return (ClientQueryRPCTopic(req.topic(), addr, 3000)) && SendTo(addr, req, cb);
+		return (ClientQueryRPCTopic(req.topic(), addr, 3000)) && SendTo(MQInfo{addr.mq_id(), addr.abs_addr()}, req, cb);
 	} catch (...) {
 		SetLastError(eError, "internal error.");
 		return false;
@@ -412,7 +497,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;
 	}
 
@@ -423,14 +508,14 @@
 		if (ClientQueryRPCTopic(request.topic(), addr, timeout_ms)) {
 			LOG_TRACE() << "node: " << SockNode().id() << ", topic dest: " << addr.mq_id();
 			BHMsgHead head(InitMsgHead(GetType(request), proc_id(), ssn()));
-			AddRoute(head, sock.id());
+			AddRoute(head, sock);
 			head.set_topic(request.topic());
 
 			MsgI reply_msg;
 			DEFER1(reply_msg.Release(););
 			BHMsgHead reply_head;
 
-			if (sock.SendAndRecv(addr.mq_id(), head, request, reply_msg, reply_head, timeout_ms) &&
+			if (sock.SendAndRecv({addr.mq_id(), addr.abs_addr()}, head, request, reply_msg, reply_head, timeout_ms) &&
 			    reply_head.type() == kMsgTypeRequestTopicReply &&
 			    reply_msg.ParseBody(out_reply)) {
 				reply_head.mutable_proc_id()->swap(out_proc_id);
@@ -443,7 +528,7 @@
 	return false;
 }
 
-int TopicNode::QueryRPCTopics(const Topic &topic, std::vector<NodeAddress> &addr, const int timeout_ms)
+int TopicNode::QueryTopicServers(const Topic &topic, std::vector<NodeAddress> &addr, const int timeout_ms)
 {
 	int n = 0;
 	MsgQueryTopic query;
@@ -463,7 +548,7 @@
 bool TopicNode::ClientQueryRPCTopic(const Topic &topic, BHAddress &addr, const int timeout_ms)
 {
 	if (!IsOnline()) {
-		SetLastError(eNotRegistered, "Not Registered.");
+		SetLastError(eNotRegistered, kErrMsgNotRegistered);
 		return false;
 	}
 
@@ -471,7 +556,7 @@
 		return true;
 	}
 	std::vector<NodeAddress> lst;
-	if (QueryRPCTopics(topic, lst, timeout_ms)) {
+	if (QueryTopicServers(topic, lst, timeout_ms)) {
 		addr = lst.front().addr();
 		if (addr.mq_id() != 0) {
 			topic_query_cache_.Store(topic, addr);
@@ -487,14 +572,14 @@
 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(), ssn()));
-		AddRoute(head, sock.id());
+		AddRoute(head, sock);
 
 		if (timeout_ms == 0) {
 			return sock.Send(BHTopicBusAddress(), head, pub);
@@ -518,7 +603,7 @@
 bool TopicNode::Subscribe(MsgTopicList &topics, MsgCommonReply &reply_body, const int timeout_ms)
 {
 	if (!IsOnline()) {
-		SetLastError(eNotRegistered, "Not Registered.");
+		SetLastError(eNotRegistered, kErrMsgNotRegistered);
 		return false;
 	}
 
@@ -528,7 +613,7 @@
 		sub.mutable_topics()->Swap(&topics);
 
 		BHMsgHead head(InitMsgHead(GetType(sub), proc_id(), ssn()));
-		AddRoute(head, sock.id());
+		AddRoute(head, sock);
 		if (timeout_ms == 0) {
 			return sock.Send(BHTopicBusAddress(), head, sub);
 		} else {
@@ -567,7 +652,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