From cf05ea3d9f43e4e84d621e1f9d54cbef552b6e2b Mon Sep 17 00:00:00 2001
From: lichao <lichao@aiotlink.com>
Date: 星期二, 18 五月 2021 16:53:28 +0800
Subject: [PATCH] fix center init mutex.

---
 src/topic_node.h |   81 ++++++++++++++++++++++++++++++----------
 1 files changed, 61 insertions(+), 20 deletions(-)

diff --git a/src/topic_node.h b/src/topic_node.h
index 35cdde5..1dfbf43 100644
--- a/src/topic_node.h
+++ b/src/topic_node.h
@@ -20,7 +20,10 @@
 
 #include "msg.h"
 #include "socket.h"
+#include <atomic>
 #include <memory>
+#include <mutex>
+#include <vector>
 
 using namespace bhome_shm;
 using namespace bhome_msg;
@@ -39,8 +42,10 @@
 
 	// topic node
 	bool Register(ProcInfo &proc, MsgCommonReply &reply_body, const int timeout_ms);
+	bool Unregister(ProcInfo &proc, MsgCommonReply &reply_body, const int timeout_ms);
 	bool Heartbeat(ProcInfo &proc, MsgCommonReply &reply_body, const int timeout_ms);
 	bool Heartbeat(const int timeout_ms);
+	bool QueryTopicAddress(BHAddress &dest, MsgQueryTopic &query, MsgQueryTopicReply &reply_body, const int timeout_ms);
 
 	// topic rpc server
 	typedef std::function<bool(const std::string &client_proc_id, const MsgRequestTopic &request, MsgRequestTopicReply &reply)> ServerSyncCB;
@@ -54,8 +59,8 @@
 	// topic client
 	typedef std::function<void(const BHMsgHead &head, const MsgRequestTopicReply &reply)> RequestResultCB;
 	bool ClientStartWorker(RequestResultCB const &cb, const int nworker = 2);
-	bool ClientAsyncRequest(const MsgRequestTopic &request, std::string &msg_id, const RequestResultCB &rrcb = RequestResultCB());
-	bool ClientSyncRequest(const MsgRequestTopic &request, std::string &proc_id, MsgRequestTopicReply &reply, const int timeout_ms);
+	bool ClientAsyncRequest(const BHAddress &remote_addr, const MsgRequestTopic &request, std::string &msg_id, const RequestResultCB &rrcb = RequestResultCB());
+	bool ClientSyncRequest(const BHAddress &remote_addr, const MsgRequestTopic &request, std::string &proc_id, MsgRequestTopicReply &reply, const int timeout_ms);
 
 	// publish
 	bool Publish(const MsgPublish &pub, const int timeout_ms);
@@ -70,7 +75,10 @@
 	void Stop();
 
 private:
+	MQId ssn() { return SockNode().id(); }
 	bool ClientQueryRPCTopic(const Topic &topic, BHAddress &addr, const int timeout_ms);
+	typedef MsgQueryTopicReply::BHNodeAddress NodeAddress;
+	int QueryTopicServers(const Topic &topic, std::vector<NodeAddress> &addr, const int timeout_ms);
 	const std::string &proc_id() { return info_.proc_id(); }
 
 	typedef BHAddress Address;
@@ -78,7 +86,11 @@
 	{
 		class Impl
 		{
-			typedef std::unordered_map<Topic, Address> Records;
+			struct TimedRec {
+				Address addr_;
+				int64_t timestamp_;
+			};
+			typedef std::unordered_map<Topic, TimedRec> Records;
 			Records records_;
 
 		public:
@@ -86,15 +98,18 @@
 			{
 				auto pos = records_.find(topic);
 				if (pos != records_.end()) {
-					addr = pos->second;
-					return true;
-				} else {
-					return false;
+					if (NowSec() - pos->second.timestamp_ < NodeTimeoutSec() / 2) {
+						addr = pos->second.addr_;
+						return true;
+					} else {
+						LOG_TRACE() << "topic dest cache timeout.";
+					}
 				}
+				return false;
 			}
 			bool Store(const Topic &topic, const Address &addr)
 			{
-				records_[topic] = addr;
+				records_[topic] = {addr, NowSec()};
 				return true;
 			}
 		};
@@ -106,20 +121,46 @@
 	};
 
 	// some sockets may be the same one, using functions make it easy to change.
+	enum { eSockStart,
+		   eSockNode = eSockStart,
+		   eSockPub = eSockNode,
+		   eSockServer,
+		   eSockClient,
+		   eSockSub,
+		   eSockEnd,
+	};
+	std::vector<std::shared_ptr<ShmSocket>> sockets_;
 
-	ShmSocket &SockNode() { return sock_node_; }
-	ShmSocket &SockPub() { return SockNode(); }
-	ShmSocket &SockSub() { return sock_sub_; }
-	ShmSocket &SockClient() { return sock_client_; }
-	ShmSocket &SockServer() { return sock_server_; }
-	bool IsRegistered() const { return registered_.load(); }
+	ShmSocket &SockNode() { return *sockets_[eSockNode]; }
+	ShmSocket &SockPub() { return *sockets_[eSockPub]; }
+	ShmSocket &SockSub() { return *sockets_[eSockSub]; }
+	ShmSocket &SockClient() { return *sockets_[eSockClient]; }
+	ShmSocket &SockServer() { return *sockets_[eSockServer]; }
 
-	ShmSocket sock_node_;
-	ShmSocket sock_client_;
-	ShmSocket sock_server_;
-	ShmSocket sock_sub_;
-	std::atomic<bool> registered_;
-	std::atomic<bool> registered_ever_;
+	void SetProcIndex(int index)
+	{
+		proc_index_ = index;
+		for (int i = eSockStart; i < eSockEnd; ++i) {
+			sockets_[i]->SetNodeProc(index, i);
+		}
+	}
+
+	enum State {
+		eStateUninited,
+		eStateUnregistered,
+		eStateOnline,
+		eStateOffline // heartbeat fail.
+	};
+	void state(const State st) { state_.store(st); }
+	void state_cas(State expected, const State val) { state_.compare_exchange_strong(expected, val); }
+	State state() const { return state_.load(); }
+	bool IsOnline() { return state() == eStateOnline; }
+	bool Init();
+	bool Valid() const { return state() != eStateUninited; }
+	std::mutex mutex_;
+	MQId ssn_id_ = 0;
+	std::atomic<State> state_;
+	int proc_index_ = -1;
 
 	TopicQueryCache topic_query_cache_;
 };

--
Gitblit v1.8.0