From bcd780993c176b93f7393607f8003adf66e6676a Mon Sep 17 00:00:00 2001 From: lichao <lichao@aiotlink.com> Date: 星期一, 26 四月 2021 10:27:14 +0800 Subject: [PATCH] fix node default ignore msg. --- src/topic_node.cpp | 205 +++++++++++++++++++++++++++++---------------------- 1 files changed, 116 insertions(+), 89 deletions(-) diff --git a/src/topic_node.cpp b/src/topic_node.cpp index 8bf9cf8..1131816 100644 --- a/src/topic_node.cpp +++ b/src/topic_node.cpp @@ -25,7 +25,7 @@ namespace { -inline void AddRoute(BHMsgHead &head, const MQId &id) { head.add_route()->set_mq_id(&id, sizeof(id)); } +inline void AddRoute(BHMsgHead &head, const MQId id) { head.add_route()->set_mq_id(id); } struct SrcInfo { std::vector<BHAddress> route; @@ -44,9 +44,10 @@ } // 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); - } + SockNode().Start(default_ignore_msg); + // for (auto &p : sockets_) { + // p->Start(default_ignore_msg); + // } } TopicNode::~TopicNode() @@ -82,14 +83,14 @@ auto &sock = SockNode(); MsgRegister body; body.mutable_proc()->Swap(&proc); - auto AddId = [&](const MQId &id) { body.add_addrs()->set_mq_id(&id, sizeof(id)); }; + 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())); + auto head(InitMsgHead(GetType(body), body.proc().proc_id(), ssn())); AddRoute(head, sock.id()); auto CheckResult = [this](MsgI &msg, BHMsgHead &head, MsgCommonReply &rbody) { @@ -108,16 +109,49 @@ MsgCommonReply body; CheckResult(imsg, head, body); }; - return sock.Send(&BHTopicCenterAddress(), head, body, onResult); + return sock.Send(BHTopicCenterAddress(), head, body, onResult); } else { MsgI reply; - DEFER1(reply.Release(shm_);); + DEFER1(reply.Release();); BHMsgHead reply_head; - bool r = sock.SendAndRecv(&BHTopicCenterAddress(), head, body, reply, reply_head, timeout_ms); + bool r = sock.SendAndRecv(BHTopicCenterAddress(), head, body, reply, reply_head, timeout_ms); if (r) { CheckResult(reply, reply_head, reply_body); } return IsOnline(); + } +} +bool TopicNode::Unregister(ProcInfo &proc, MsgCommonReply &reply_body, const int timeout_ms) +{ + info_.Clear(); + state_cas(eStateOnline, eStateOffline); + + auto &sock = SockNode(); + MsgUnregister body; + body.mutable_proc()->Swap(&proc); + + auto head(InitMsgHead(GetType(body), body.proc().proc_id(), ssn())); + AddRoute(head, sock.id()); + + auto CheckResult = [this](MsgI &msg, BHMsgHead &head, MsgCommonReply &rbody) { + bool r = head.type() == kMsgTypeCommonReply && + msg.ParseBody(rbody) && + IsSuccess(rbody.errmsg().errcode()); + return r; + }; + + if (timeout_ms == 0) { + auto onResult = [this, CheckResult](ShmSocket &socket, MsgI &imsg, BHMsgHead &head) { + MsgCommonReply body; + CheckResult(imsg, head, body); + }; + return sock.Send(BHTopicCenterAddress(), head, body, onResult); + } else { + MsgI reply; + DEFER1(reply.Release();); + BHMsgHead reply_head; + bool r = sock.SendAndRecv(BHTopicCenterAddress(), head, body, reply, reply_head, timeout_ms); + return r && CheckResult(reply, reply_head, reply_body); } } @@ -132,16 +166,16 @@ MsgHeartbeat body; body.mutable_proc()->Swap(&proc); - auto head(InitMsgHead(GetType(body), body.proc().proc_id())); + auto head(InitMsgHead(GetType(body), body.proc().proc_id(), ssn())); AddRoute(head, sock.id()); if (timeout_ms == 0) { - return sock.Send(&BHTopicCenterAddress(), head, body); + return sock.Send(BHTopicCenterAddress(), head, body); } else { MsgI reply; - DEFER1(reply.Release(shm_);); + DEFER1(reply.Release();); BHMsgHead reply_head; - bool r = sock.SendAndRecv(&BHTopicCenterAddress(), head, body, reply, reply_head, timeout_ms); + bool r = sock.SendAndRecv(BHTopicCenterAddress(), head, body, reply, reply_head, timeout_ms); r = r && reply_head.type() == kMsgTypeCommonReply && reply.ParseBody(reply_body); return (r && IsSuccess(reply_body.errmsg().errcode())); } @@ -152,6 +186,25 @@ proc.set_proc_id(proc_id()); MsgCommonReply reply_body; return Heartbeat(proc, reply_body, timeout_ms); +} + +bool TopicNode::QueryTopicAddress(BHAddress &dest, MsgQueryTopic &query, MsgQueryTopicReply &reply_body, const int timeout_ms) +{ + if (!IsOnline()) { + SetLastError(eNotRegistered, "Not Registered."); + return false; + } + auto &sock = SockNode(); + + BHMsgHead head(InitMsgHead(GetType(query), proc_id(), ssn())); + AddRoute(head, sock.id()); + + MsgI reply; + DEFER1(reply.Release()); + BHMsgHead reply_head; + return (sock.SendAndRecv(BHTopicCenterAddress(), head, query, reply, reply_head, timeout_ms) && + reply_head.type() == kMsgTypeQueryTopicReply && + reply.ParseBody(reply_body)); } bool TopicNode::ServerRegisterRPC(MsgTopicList &topics, MsgCommonReply &reply_body, const int timeout_ms) @@ -165,16 +218,16 @@ MsgRegisterRPC body; body.mutable_topics()->Swap(&topics); - auto head(InitMsgHead(GetType(body), proc_id())); + auto head(InitMsgHead(GetType(body), proc_id(), ssn())); AddRoute(head, sock.id()); if (timeout_ms == 0) { - return sock.Send(&BHTopicCenterAddress(), head, body); + return sock.Send(BHTopicCenterAddress(), head, body); } else { MsgI reply; - DEFER1(reply.Release(shm_);); + DEFER1(reply.Release();); BHMsgHead reply_head; - bool r = sock.SendAndRecv(&BHTopicCenterAddress(), head, body, reply, reply_head, timeout_ms); + bool r = sock.SendAndRecv(BHTopicCenterAddress(), head, body, reply, reply_head, timeout_ms); r = r && reply_head.type() == kMsgTypeCommonReply; r = r && reply.ParseBody(reply_body); return r; @@ -190,13 +243,13 @@ MsgRequestTopicReply reply_body; if (rcb(head.proc_id(), req, reply_body)) { - BHMsgHead reply_head(InitMsgHead(GetType(reply_body), proc_id(), head.msg_id())); + BHMsgHead reply_head(InitMsgHead(GetType(reply_body), proc_id(), ssn(), head.msg_id())); 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(); - sock.Send(remote.data(), reply_head, reply_body); + auto remote = head.route().rbegin()->mq_id(); + sock.Send(remote, reply_head, reply_body); } }; @@ -259,11 +312,11 @@ if (!p || p->route.empty()) { return false; } - BHMsgHead head(InitMsgHead(GetType(body), proc_id(), p->msg_id)); + BHMsgHead head(InitMsgHead(GetType(body), proc_id(), ssn(), p->msg_id)); 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().data(), head, body); + return sock.Send(p->route.back().mq_id(), head, body); } bool TopicNode::ClientStartWorker(RequestResultCB const &cb, const int nworker) @@ -283,7 +336,7 @@ return SockClient().Start(onData, nworker); } -bool TopicNode::ClientAsyncRequest(const MsgRequestTopic &req, std::string &out_msg_id, const RequestResultCB &cb) +bool TopicNode::ClientAsyncRequest(const BHAddress &remote_addr, const MsgRequestTopic &req, std::string &out_msg_id, const RequestResultCB &cb) { if (!IsOnline()) { SetLastError(eNotRegistered, "Not Registered."); @@ -296,7 +349,7 @@ auto SendTo = [this, msg_id](const BHAddress &addr, const MsgRequestTopic &req, const RequestResultCB &cb) { auto &sock = SockClient(); - BHMsgHead head(InitMsgHead(GetType(req), proc_id(), msg_id)); + BHMsgHead head(InitMsgHead(GetType(req), proc_id(), ssn(), msg_id)); AddRoute(head, sock.id()); head.set_topic(req.topic()); @@ -309,47 +362,22 @@ } } }; - return sock.Send(addr.mq_id().data(), head, req, onRecv); + return sock.Send(addr.mq_id(), head, req, onRecv); } else { - return sock.Send(addr.mq_id().data(), head, req); + return sock.Send(addr.mq_id(), head, req); } }; try { BHAddress addr; -#if 1 return (ClientQueryRPCTopic(req.topic(), addr, 3000)) && SendTo(addr, req, cb); -#else - if (topic_query_cache_.Pick(req.topic(), addr)) { - return SendTo(addr, req, cb); - } - - auto &sock = SockClient(); - MsgQueryTopic query; - query.set_topic(req.topic()); - BHMsgHead head(InitMsgHead(GetType(query), proc_id())); - AddRoute(head, sock.id()); - - auto onQueryResult = [this, SendTo, req, cb](ShmSocket &sock, MsgI &imsg, BHMsgHead &head) { - MsgQueryTopicReply rep; - if (head.type() == kMsgTypeQueryTopicReply && imsg.ParseBody(rep)) { - auto &addr = rep.address(); - if (!addr.mq_id().empty()) { - topic_query_cache_.Store(req.topic(), addr); - SendTo(addr, req, cb); - } - } - }; - return sock.Send(&BHTopicCenterAddress(), head, query, std::move(onQueryResult)); -#endif - } catch (...) { SetLastError(eError, "internal error."); return false; } } -bool TopicNode::ClientSyncRequest(const MsgRequestTopic &request, std::string &out_proc_id, MsgRequestTopicReply &out_reply, const int timeout_ms) +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."); @@ -361,15 +389,16 @@ BHAddress addr; if (ClientQueryRPCTopic(request.topic(), addr, timeout_ms)) { - BHMsgHead head(InitMsgHead(GetType(request), proc_id())); + printf("node: %ld, topic dest: %ld\n", SockNode().id(), addr.mq_id()); + BHMsgHead head(InitMsgHead(GetType(request), proc_id(), ssn())); AddRoute(head, sock.id()); head.set_topic(request.topic()); MsgI reply_msg; - DEFER1(reply_msg.Release(shm_);); + DEFER1(reply_msg.Release();); BHMsgHead reply_head; - if (sock.SendAndRecv(addr.mq_id().data(), head, request, reply_msg, reply_head, timeout_ms) && + if (sock.SendAndRecv(addr.mq_id(), 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); @@ -384,6 +413,23 @@ return false; } +int TopicNode::QueryRPCTopics(const Topic &topic, std::vector<NodeAddress> &addr, const int timeout_ms) +{ + int n = 0; + MsgQueryTopic query; + query.set_topic(topic); + MsgQueryTopicReply rep; + BHAddress dest; // empty means local. + if (QueryTopicAddress(dest, query, rep, timeout_ms)) { + auto &ls = rep.node_address(); + n = ls.size(); + for (auto &na : ls) { + addr.push_back(na); + } + } + return n; +} + bool TopicNode::ClientQueryRPCTopic(const Topic &topic, BHAddress &addr, const int timeout_ms) { if (!IsOnline()) { @@ -391,35 +437,16 @@ return false; } - auto &sock = SockClient(); - if (topic_query_cache_.Find(topic, addr)) { return true; } - - MsgQueryTopic query; - query.set_topic(topic); - BHMsgHead head(InitMsgHead(GetType(query), proc_id())); - AddRoute(head, sock.id()); - - MsgI reply; - DEFER1(reply.Release(shm_)); - BHMsgHead reply_head; - - if (sock.SendAndRecv(&BHTopicCenterAddress(), head, query, reply, reply_head, timeout_ms)) { - if (reply_head.type() == kMsgTypeQueryTopicReply) { - MsgQueryTopicReply rep; - if (reply.ParseBody(rep)) { - addr = rep.address(); - if (addr.mq_id().empty()) { - return false; - } else { - topic_query_cache_.Store(topic, addr); - return true; - } - } + std::vector<NodeAddress> lst; + if (QueryRPCTopics(topic, lst, timeout_ms)) { + addr = lst.front().addr(); + if (addr.mq_id() != 0) { + topic_query_cache_.Store(topic, addr); + return true; } - } else { } return false; } @@ -435,17 +462,17 @@ try { auto &sock = SockPub(); - BHMsgHead head(InitMsgHead(GetType(pub), proc_id())); + BHMsgHead head(InitMsgHead(GetType(pub), proc_id(), ssn())); AddRoute(head, sock.id()); if (timeout_ms == 0) { - return sock.Send(&BHTopicBusAddress(), head, pub); + return sock.Send(BHTopicBusAddress(), head, pub); } else { MsgI reply; - DEFER1(reply.Release(shm());); + DEFER1(reply.Release();); BHMsgHead reply_head; MsgCommonReply reply_body; - return sock.SendAndRecv(&BHTopicBusAddress(), head, pub, reply, reply_head, timeout_ms) && + return sock.SendAndRecv(BHTopicBusAddress(), head, pub, reply, reply_head, timeout_ms) && reply_head.type() == kMsgTypeCommonReply && reply.ParseBody(reply_body) && IsSuccess(reply_body.errmsg().errcode()); @@ -469,15 +496,15 @@ MsgSubscribe sub; sub.mutable_topics()->Swap(&topics); - BHMsgHead head(InitMsgHead(GetType(sub), proc_id())); + BHMsgHead head(InitMsgHead(GetType(sub), proc_id(), ssn())); AddRoute(head, sock.id()); if (timeout_ms == 0) { - return sock.Send(&BHTopicBusAddress(), head, sub); + return sock.Send(BHTopicBusAddress(), head, sub); } else { MsgI reply; - DEFER1(reply.Release(shm());); + DEFER1(reply.Release();); BHMsgHead reply_head; - return sock.SendAndRecv(&BHTopicBusAddress(), head, sub, reply, reply_head, timeout_ms) && + return sock.SendAndRecv(BHTopicBusAddress(), head, sub, reply, reply_head, timeout_ms) && reply_head.type() == kMsgTypeCommonReply && reply.ParseBody(reply_body) && IsSuccess(reply_body.errmsg().errcode()); @@ -515,7 +542,7 @@ auto &sock = SockSub(); MsgI msg; - DEFER1(msg.Release(shm());); + DEFER1(msg.Release();); BHMsgHead head; //TODO error msg. if (sock.SyncRecv(msg, head, timeout_ms) && head.type() == kMsgTypePublish) { -- Gitblit v1.8.0