From 34cd75f77d0ca94dbdba4e6cc9451fe4d33e78b3 Mon Sep 17 00:00:00 2001 From: lichao <lichao@aiotlink.com> Date: 星期三, 19 五月 2021 19:14:13 +0800 Subject: [PATCH] add api BHQueryProcs. --- src/topic_node.cpp | 327 +++++++++++++++++++++++++++++++++++++++--------------- 1 files changed, 235 insertions(+), 92 deletions(-) diff --git a/src/topic_node.cpp b/src/topic_node.cpp index 13bb8ee..8bbb929 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,60 +45,133 @@ } // namespace TopicNode::TopicNode(SharedMemory &shm) : - shm_(shm), sockets_(eSockEnd), state_(eStateUnregistered) + shm_(shm), state_(eStateUninited) { - 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 = [&]() { + int64_t init_request = ssn_id_ << 4 | EncodeCmd(eCmdNodeInit); + int64_t reply = 0; + if (BHNodeInit(init_request, reply) && DecodeCmd(reply) == eCmdNodeInitReply) { + int64_t abs_addr = reply >> 4; + sockets_.emplace_back(new ShmSocket(abs_addr, shm_, ssn_id_)); + LOG_DEBUG() << "node init ok"; + } else { + LOG_ERROR() << "Node Init Error"; + } + }; + if (sockets_.empty()) { + NodeInit(); + } + if (!sockets_.empty()) { + auto onMsg = [this](ShmSocket &socket, MsgI &imsg, BHMsgHead &head) { + LOG_DEBUG() << "node recv type: " << head.type(); + switch (head.type()) { + case kMsgTypeProcInit: { + // reuse msg to send proc init. + MsgProcInit body; + body.set_extra_mq_num(eSockEnd - eSockStart - 1); + auto head = InitMsgHead(GetType(body), info_.proc_id(), ssn_id_); + AddRoute(head, socket); + if (imsg.Fill(head, body)) { + socket.Send(BHTopicCenterAddress(), imsg); + } + } break; + case kMsgTypeProcInitReply: { + LOG_DEBUG() << "got proc init reply"; + MsgProcInitReply reply; + if (imsg.ParseBody(reply) && IsSuccess(reply.errmsg().errcode())) { + for (auto &addr : reply.extra_mqs()) { + LOG_DEBUG() << "add socket " << addr.abs_addr() << ", id:" << addr.mq_id(); + sockets_.emplace_back(new ShmSocket(addr.abs_addr(), shm(), addr.mq_id())); + } + SetProcIndex(reply.proc_index()); + this->state_ = eStateUnregistered; + + ServerStart(ServerAsyncCB(), 1); + SubscribeStartWorker(SubDataCB(), 1); + } + } break; + default: break; + } + return true; + }; + SockNode().Start(1, onMsg); + 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 end_time = steady_clock::now() + milliseconds(timeout_ms); + + while (!Valid() && steady_clock::now() < end_time) { + std::this_thread::yield(); + } + if (!Valid()) { + 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 && @@ -122,6 +203,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); @@ -130,7 +216,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 && @@ -157,7 +243,7 @@ bool TopicNode::Heartbeat(ProcInfo &proc, MsgCommonReply &reply_body, const int timeout_ms) { if (!IsOnline()) { - SetLastError(eNotRegistered, "Not Registered."); + SetLastError(eNotRegistered, kErrMsgNotRegistered); return false; } @@ -166,7 +252,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); @@ -190,13 +276,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()); @@ -206,10 +292,29 @@ reply.ParseBody(reply_body)); } +bool TopicNode::QueryProcs(BHAddress &dest, MsgQueryProc &query, MsgQueryProcReply &reply_body, const int timeout_ms) +{ + if (!IsOnline()) { + SetLastError(eNotRegistered, kErrMsgNotRegistered); + return false; + } + auto &sock = SockNode(); + + BHMsgHead head(InitMsgHead(GetType(query), proc_id(), ssn())); + AddRoute(head, sock); + + MsgI reply; + DEFER1(reply.Release()); + BHMsgHead reply_head; + return (sock.SendAndRecv(BHTopicCenterAddress(), head, query, reply, reply_head, timeout_ms) && + reply_head.type() == kMsgTypeQueryProcReply && + reply.ParseBody(reply_body)); +} + bool TopicNode::ServerRegisterRPC(MsgTopicList &topics, MsgCommonReply &reply_body, const int timeout_ms) { if (!IsOnline()) { - SetLastError(eNotRegistered, "Not Registered."); + SetLastError(eNotRegistered, kErrMsgNotRegistered); return false; } @@ -218,7 +323,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); @@ -247,7 +352,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); } }; @@ -258,40 +363,67 @@ bool TopicNode::ServerStart(const ServerAsyncCB &acb, int nworker) { - auto onRecv = [this, acb](ShmSocket &sock, MsgI &imsg, BHMsgHead &head) { - if (head.type() != kMsgTypeRequestTopic || head.route_size() == 0) { return; } - MsgRequestTopic req; - if (!imsg.ParseBody(req)) { return; } + if (acb) { + auto onRecv = [this, acb](ShmSocket &sock, MsgI &imsg, BHMsgHead &head) { + if (head.type() != kMsgTypeRequestTopic || head.route_size() == 0) { return; } + 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(); - return acb && sock.Start(onRecv, nworker); + return SockServer().Start(onRecv, nworker); + } else { + auto onRequest = [this](ShmSocket &socket, MsgI &msg, BHMsgHead &head) { + server_buffer_->Write(std::move(head), msg.body()); + }; + return SockServer().Start(onRequest, nworker); + } } 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; } - - auto &sock = SockServer(); - - MsgI imsg; BHMsgHead head; - if (sock.SyncRecv(imsg, head, timeout_ms) && head.type() == kMsgTypeRequestTopic) { - if (imsg.ParseBody(request)) { + std::string body; + auto end_time = steady_clock::now() + milliseconds(timeout_ms); + while (!server_buffer_->Read(head, body)) { + if (steady_clock::now() < end_time) { + robust::QuickSleep(); + } else { + return false; + } + } + + if (head.type() == kMsgTypeRequestTopic) { + if (request.ParseFromString(body)) { 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; @@ -300,7 +432,7 @@ bool TopicNode::ServerSendReply(void *src_info, const MsgRequestTopicReply &body) { if (!IsOnline()) { - SetLastError(eNotRegistered, "Not Registered."); + SetLastError(eNotRegistered, kErrMsgNotRegistered); return false; } @@ -315,7 +447,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) @@ -338,7 +471,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; } @@ -346,10 +479,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) { @@ -361,15 +494,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; @@ -379,7 +512,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,22 +521,21 @@ BHAddress addr; 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); return true; } - } else { - SetLastError(eNotFound, "remote not found."); } } catch (...) { SetLastError(eError, __func__ + std::string(" internal errer.")); @@ -411,7 +543,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; @@ -431,7 +563,7 @@ bool TopicNode::ClientQueryRPCTopic(const Topic &topic, BHAddress &addr, const int timeout_ms) { if (!IsOnline()) { - SetLastError(eNotRegistered, "Not Registered."); + SetLastError(eNotRegistered, kErrMsgNotRegistered); return false; } @@ -439,13 +571,14 @@ 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); return true; } } + SetLastError(eNotFound, "remote not found."); return false; } @@ -454,14 +587,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); @@ -485,7 +618,7 @@ bool TopicNode::Subscribe(MsgTopicList &topics, MsgCommonReply &reply_body, const int timeout_ms) { if (!IsOnline()) { - SetLastError(eNotRegistered, "Not Registered."); + SetLastError(eNotRegistered, kErrMsgNotRegistered); return false; } @@ -495,7 +628,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 { @@ -515,36 +648,46 @@ bool TopicNode::SubscribeStartWorker(const SubDataCB &tdcb, int nworker) { - auto &sock = SockSub(); - - auto AsyncRecvProc = [this, tdcb](ShmSocket &, MsgI &imsg, BHMsgHead &head) { - if (head.type() == kMsgTypePublish) { - MsgPublish pub; - if (imsg.ParseBody(pub)) { - tdcb(head.proc_id(), pub); + if (tdcb) { + auto AsyncRecvProc = [this, tdcb](ShmSocket &, MsgI &imsg, BHMsgHead &head) { + if (head.type() == kMsgTypePublish) { + MsgPublish pub; + if (imsg.ParseBody(pub)) { + tdcb(head.proc_id(), pub); + } + } else { + // ignored, or dropped } - } else { - // ignored, or dropped - } - }; - - return tdcb && sock.Start(AsyncRecvProc, nworker); + }; + return SockSub().Start(AsyncRecvProc, nworker); + } else { + auto onSub = [this](ShmSocket &socket, MsgI &msg, BHMsgHead &head) { + sub_buffer_->Write(std::move(head), msg.body()); + }; + return SockSub().Start(onSub, nworker); + } } bool TopicNode::RecvSub(std::string &proc_id, MsgPublish &pub, const int timeout_ms) { if (!IsOnline()) { - SetLastError(eNotRegistered, "Not Registered."); + SetLastError(eNotRegistered, kErrMsgNotRegistered); return false; } - auto &sock = SockSub(); - MsgI msg; - DEFER1(msg.Release();); BHMsgHead head; + std::string body; + auto end_time = steady_clock::now() + milliseconds(timeout_ms); + while (!sub_buffer_->Read(head, body)) { + if (steady_clock::now() < end_time) { + robust::QuickSleep(); + } else { + return false; + } + } //TODO error msg. - if (sock.SyncRecv(msg, head, timeout_ms) && head.type() == kMsgTypePublish) { - if (msg.ParseBody(pub)) { + if (head.type() == kMsgTypePublish) { + if (pub.ParseFromString(body)) { head.mutable_proc_id()->swap(proc_id); return true; } -- Gitblit v1.8.0