From 150e192a7ddf19783a0f4a81b87d270c0812277a Mon Sep 17 00:00:00 2001 From: lichao <lichao@aiotlink.com> Date: 星期一, 26 四月 2021 10:25:54 +0800 Subject: [PATCH] fix center handle ssn after node restart. --- utest/api_test.cpp | 124 ++++++++++++++++++++++++++++++----------- box/center.cpp | 13 ++-- 2 files changed, 96 insertions(+), 41 deletions(-) diff --git a/box/center.cpp b/box/center.cpp index 4bb9ea1..829a089 100644 --- a/box/center.cpp +++ b/box/center.cpp @@ -112,8 +112,7 @@ try { MQId ssn = head.ssn_id(); - // use src_addr as session id. - // when node restart, src_addr will change, + // when node restart, ssn will change, // and old node will be removed after timeout. auto UpdateRegInfo = [&](Node &node) { node->addrs_.insert(SrcAddr(head)); @@ -134,14 +133,15 @@ UpdateRegInfo(node); nodes_[ssn] = node; - auto old = node_addr_map_.find(head.proc_id()); - if (old != node_addr_map_.end()) { // old session + printf("new ssn %ld\n", ssn); + auto old = online_node_addr_map_.find(head.proc_id()); + if (old != online_node_addr_map_.end()) { // old session auto &old_ssn = old->second; nodes_[old_ssn]->state_.PutOffline(offline_time_); printf("put %s %ld offline\n", nodes_[old_ssn]->proc_.proc_id().c_str(), old->second); old_ssn = ssn; } else { - node_addr_map_.emplace(head.proc_id(), ssn); + online_node_addr_map_.emplace(head.proc_id(), ssn); } } return MakeReply(eSuccess); @@ -387,7 +387,6 @@ }; EraseMapRec(service_map_, node->services_); EraseMapRec(subscribe_map_, node->subscriptions_); - node_addr_map_.erase(node->proc_.proc_id()); for (auto &addr : node->addrs_) { cleaner_(addr); @@ -399,7 +398,7 @@ std::unordered_map<Topic, Clients> service_map_; std::unordered_map<Topic, Clients> subscribe_map_; std::unordered_map<Address, Node> nodes_; - std::unordered_map<std::string, Address> node_addr_map_; + std::unordered_map<std::string, Address> online_node_addr_map_; Cleaner cleaner_; // remove mqs. int64_t offline_time_; int64_t kill_time_; diff --git a/utest/api_test.cpp b/utest/api_test.cpp index dd59b09..debe8ad 100644 --- a/utest/api_test.cpp +++ b/utest/api_test.cpp @@ -170,41 +170,38 @@ void *Ptr(const Offset offset) { return reinterpret_cast<void *>(offset); } } // namespace +class RobustMutex +{ +public: + RobustMutex() + { + pthread_mutexattr_t attr; + pthread_mutexattr_init(&attr); + pthread_mutexattr_setrobust(&attr, 1); + pthread_mutex_init(mtx(), &attr); + if (!valid()) { + throw("init mutex error."); + } + } + int TryLock() { return pthread_mutex_trylock(mtx()); } + int Lock() { return pthread_mutex_lock(mtx()); } + int Unlock() { return pthread_mutex_unlock(mtx()); } + bool valid() const { return false; } + +private: + pthread_mutex_t *mtx() { return &mutex_; } + pthread_mutex_t mutex_; +}; + BOOST_AUTO_TEST_CASE(MutexTest) { SharedMemory &shm = TestShm(); - MsgI::BindShm(shm); - - void *base_ptr = shm.get_address(); - auto PrintPtr = [&](void *p) { - printf("addr: %ld, ptr: %p, offset: %ld\n", Addr(p), p, Addr(p) - Addr(base_ptr)); - }; - - printf("base"); - PrintPtr(base_ptr); - - MsgI msg; - msg.Make("string data"); - for (int i = 0; i < 10; ++i) { - int n = msg.AddRef(); - printf("add %d ref: %d\n", i, n); - } - for (int i = 0; i < 10; ++i) { - int n = msg.Release(); - printf("release %d, ref : %d\n", i, n); - } - std::this_thread::sleep_for(1s); - msg.Release(); + GlobalInit(shm); const std::string mtx_name("test_mutex"); const std::string int_name("test_int"); auto mtx = shm.FindOrCreate<Mutex>(mtx_name); auto pi = shm.FindOrCreate<int>(int_name, 100); - - printf("mutetx "); - PrintPtr(mtx); - printf("int "); - PrintPtr(pi); typedef std::chrono::steady_clock Clock; auto Now = []() { return Clock::now().time_since_epoch(); }; @@ -310,7 +307,7 @@ const std::string topic_ = "topic_"; - { + { // Server Register Topics MsgTopicList topics; for (int i = 0; i < 10; ++i) { topics.add_topic_list(topic_ + std::to_string(i)); @@ -324,7 +321,7 @@ Sleep(1s); } - { + { // Subscribe MsgTopicList topics; for (int i = 0; i < 10; ++i) { topics.add_topic_list(topic_ + std::to_string(i * 2)); @@ -337,8 +334,57 @@ printf("subscribe topic : %s\n", r ? "ok" : "failed"); } - BHStartWorker(&ServerProc, &SubRecvProc, &ClientProc); + // BHStartWorker(&ServerProc, &SubRecvProc, &ClientProc); + auto ServerLoop = [&](std::atomic<bool> *run) { + while (*run) { + void *proc_id = 0; + int proc_id_len = 0; + DEFER1(BHFree(proc_id, proc_id_len);); + void *input = 0; + int input_len = 0; + DEFER1(BHFree(input, input_len)); + void *src = 0; + if (BHReadRequest(&proc_id, &proc_id_len, &input, &input_len, &src, 10) && src) { + MsgRequestTopic request; + if (request.ParseFromArray(input, input_len)) { + MsgRequestTopicReply reply; + reply.set_data(" reply: " + request.data()); + std::string s(reply.SerializeAsString()); + // printf("%s", reply.data().c_str()); + BHSendReply(src, s.data(), s.size()); + ++Status().nserved_; + } + src = 0; + } + } + }; + + auto SyncRequest = [&](int idx) { // SyncRequest + MsgRequestTopic req; + req.set_topic(topic_ + std::to_string(idx)); + req.set_data("request_data_" + std::to_string(idx)); + std::string s(req.SerializeAsString()); + // Sleep(10ms, false); + std::string dest(BHAddress().SerializeAsString()); + void *proc_id = 0; + int proc_id_len = 0; + DEFER1(BHFree(proc_id, proc_id_len);); + void *reply = 0; + int reply_len = 0; + DEFER1(BHFree(reply, reply_len)); + bool r = BHRequest(dest.data(), dest.size(), s.data(), s.size(), &proc_id, &proc_id_len, &reply, &reply_len, 100); + if (!r) { + int ec = 0; + std::string msg; + GetLastError(ec, msg); + printf("request error: %s\n", msg.c_str()); + } else { + MsgRequestTopicReply ret; + ret.ParseFromArray(reply, reply_len); + printf("request result: %s\n", ret.data().c_str()); + } + }; { for (int i = 0; i < 1; ++i) { MsgPublish pub; @@ -358,10 +404,10 @@ std::string s(req.SerializeAsString()); void *msg_id = 0; int len = 0; + DEFER1(BHFree(msg_id, len);); // Sleep(10ms, false); std::string dest(BHAddress().SerializeAsString()); bool r = BHAsyncRequest(dest.data(), dest.size(), s.data(), s.size(), 0, 0); - DEFER1(BHFree(msg_id, len);); if (r) { ++Status().nrequest_; } else { @@ -397,20 +443,30 @@ printf("heartbeat: %s\n", r ? "ok" : "failed"); } }; + std::atomic<bool> run(true); + ThreadManager threads; boost::timer::auto_cpu_timer timer; threads.Launch(hb, &run); + threads.Launch(ServerLoop, &run); threads.Launch(showStatus, &run); int ncli = 10; - const uint64_t nreq = 1000 * 100; + const uint64_t nreq = 1000 * 1; for (int i = 0; i < ncli; ++i) { - threads.Launch(asyncRequest, nreq); + // threads.Launch(asyncRequest, nreq); } + + for (int i = 0; i < 10; ++i) { + SyncRequest(i); + } + // run.store(false); + // server_thread.join(); + // return; int same = 0; int64_t last = 0; - while (last < nreq * ncli && same < 2) { + while (last < nreq * ncli && same < 1) { Sleep(1s, false); auto cur = Status().nreply_.load(); if (last == cur) { -- Gitblit v1.8.0