From dc12826dd61ce18fac3a9561c5843d30a0cf9660 Mon Sep 17 00:00:00 2001
From: lichao <lichao@aiotlink.com>
Date: 星期五, 02 四月 2021 15:48:53 +0800
Subject: [PATCH] add request topic cache; refactor req/rep center.

---
 src/reqrep_center.cpp |  136 +++++++++++++++++++++++++++++----------------
 1 files changed, 88 insertions(+), 48 deletions(-)

diff --git a/src/reqrep_center.cpp b/src/reqrep_center.cpp
index 5f1873e..0b6ddea 100644
--- a/src/reqrep_center.cpp
+++ b/src/reqrep_center.cpp
@@ -17,24 +17,96 @@
  */
 #include "reqrep_center.h"
 #include "bh_util.h"
-using namespace bhome_shm;
+#include "msg.h"
+#include <chrono>
+#include <memory>
+#include <mutex>
+#include <unordered_map>
 
-struct A {
-	void F(int){};
-};
+using namespace bhome_shm;
 
 namespace
 {
-inline uint64_t Now()
-{
-	time_t t;
-	return time(&t);
-}
+auto Now = []() { time_t t; return time(&t); };
 
+class NodeCenter
+{
+public:
+	typedef std::string ProcAddr;
+	typedef bhome::msg::ProcInfo ProcInfo;
+
+	template <class Iter>
+	bool Register(ProcInfo &info, const ProcAddr &src_mq, Iter topics_begin, Iter topics_end)
+	{
+		try {
+			Node node(new NodeInfo);
+			node->addr_ = src_mq;
+			node->proc_.Swap(&info);
+			node->state_.timestamp_ = Now();
+			nodes_[node->proc_.id()] = node;
+			for (auto it = topics_begin; it != topics_end; ++it) {
+				topic_map_[*it] = node;
+			}
+			return true;
+		} catch (...) {
+			return false;
+		}
+	}
+	void Heartbeat(ProcInfo &info, const ProcAddr &src_mq)
+	{
+		auto pos = nodes_.find(info.name());
+		if (pos != nodes_.end() && pos->second->addr_ == src_mq) { // both name and mq should be the same.
+			NodeInfo &ni = *pos->second;
+			ni.state_.timestamp_ = Now();
+			if (!info.public_info().empty()) {
+				ni.proc_.set_public_info(info.public_info());
+			}
+			if (!info.private_info().empty()) {
+				ni.proc_.set_private_info(info.private_info());
+			}
+		}
+	}
+	bool QueryTopic(const std::string &topic, ProcAddr &addr)
+	{
+		auto pos = topic_map_.find(topic);
+		if (pos != topic_map_.end()) {
+			Node node(pos->second.lock());
+			if (node) {
+				addr = node->addr_;
+				return true;
+			} else { // dead, remove record.
+				topic_map_.erase(pos);
+				return false;
+			}
+		} else {
+			return false;
+		}
+	}
+
+private:
+	struct ProcState {
+		time_t timestamp_ = 0;
+		uint32_t flag_ = 0; // reserved
+	};
+	typedef std::string ProcId;
+	struct NodeInfo {
+		ProcState state_; // state
+		ProcAddr addr_;   // registered_mqid.
+		ProcInfo proc_;   //
+	};
+	typedef std::shared_ptr<NodeInfo> Node;
+	typedef std::weak_ptr<NodeInfo> WeakNode;
+	std::unordered_map<std::string, WeakNode> topic_map_;
+	std::unordered_map<ProcId, Node> nodes_;
+};
 } // namespace
+
 bool ReqRepCenter::Start(const int nworker)
 {
-	auto onRecv = [&](BHMsg &msg) {
+	auto center_ptr = std::make_shared<Synced<NodeCenter>>();
+	auto onRecv = [center_ptr, this](BHMsg &msg) {
+		auto &center = *center_ptr;
+
 #ifndef NDEBUG
 		static std::atomic<time_t> last(0);
 		time_t now = 0;
@@ -50,54 +122,22 @@
 
 		auto OnRegister = [&]() {
 			DataProcRegister reg;
-			if (!reg.ParseFromString(msg.body())) {
-				return;
+			if (reg.ParseFromString(msg.body()) && reg.has_proc()) {
+				center->Register(*reg.mutable_proc(), src_mq, reg.topics().begin(), reg.topics().end());
 			}
-			ProcInfo pi;
-			pi.server_mqid_ = src_mq;
-			pi.proc_id_ = reg.proc().name();
-			pi.ext_info_ = reg.proc().info();
-			pi.timestamp_ = Now();
-
-			std::lock_guard<std::mutex> lock(mutex_);
-			for (auto &t : reg.topics()) {
-				topic_mq_[t] = pi.server_mqid_;
-			}
-			procs_[pi.proc_id_] = pi;
 		};
 
 		auto OnHeartbeat = [&]() {
 			DataProcHeartbeat hb;
-			if (!hb.ParseFromString(msg.body())) {
-				return;
-			}
-
-			std::lock_guard<std::mutex> lock(mutex_);
-			auto pos = procs_.find(hb.proc().name());
-			if (pos != procs_.end() && pos->second.server_mqid_ == src_mq) { // both name and mq should be the same.
-				pos->second.timestamp_ = Now();
-				pos->second.ext_info_ = hb.proc().info();
+			if (hb.ParseFromString(msg.body()) && hb.has_proc()) {
+				center->Heartbeat(*hb.mutable_proc(), src_mq);
 			}
 		};
 
 		auto OnQueryTopic = [&]() {
 			DataProcQueryTopic query;
-			if (!query.ParseFromString(msg.body())) {
-				return;
-			}
-
-			std::string dest;
-			auto FindDest = [&]() {
-				std::lock_guard<std::mutex> lock(mutex_);
-				auto pos = topic_mq_.find(query.topic());
-				if (pos != topic_mq_.end()) {
-					dest = pos->second;
-					return true;
-				} else {
-					return false;
-				}
-			};
-			if (FindDest()) {
+			NodeCenter::ProcAddr dest;
+			if (query.ParseFromString(msg.body()) && center->QueryTopic(query.topic(), dest)) {
 				MQId remote;
 				memcpy(&remote, msg.route().rbegin()->mq_id().data(), sizeof(remote));
 				MsgI imsg;

--
Gitblit v1.8.0