test lock code.

lichao

2021-05-10 77a6c3512a44dfe6540dde71946e6484fe4f173f

 src/topic_node.cpp

@@ -23,6 +23,9 @@
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); }
@@ -63,24 +66,33 @@
   MsgI msg;
   msg.OffsetRef() = ssn_id_;
   if (ShmMsgQueue::TrySend(shm(), BHInitAddress(), msg)) {
      sockets_.resize(eSockEnd);
      for (int i = eSockStart; i < eSockEnd; ++i) {
         sockets_[i].reset(new ShmSocket(shm_, ssn_id_ + i, kMqLen));

      auto end_time = steady_clock::now() + 3s;
      do {
         try {
            for (int i = eSockStart; i < eSockEnd; ++i) {
               sockets_.emplace_back(new ShmSocket(shm_, false, ssn_id_ + i, kMqLen));
            }
            break;
         } catch (...) {
            sockets_.clear();
            std::this_thread::sleep_for(100ms);
         }
      } while (steady_clock::now() < end_time);
      if (!sockets_.empty()) {
         // recv msgs to avoid memory leak.
         auto default_ignore_msg = [](ShmSocket &sock, MsgI &imsg, BHMsgHead &head) { return true; };
         SockNode().Start(default_ignore_msg);
         return true;
      }
      // recv msgs to avoid memory leak.
      auto default_ignore_msg = [](ShmSocket &sock, MsgI &imsg, BHMsgHead &head) { return true; };
      SockNode().Start(default_ignore_msg);
      return true;
   }
   return false;
}

void TopicNode::Start(ServerAsyncCB const &server_cb, SubDataCB const &sub_cb, RequestResultCB &client_cb, int nworker)
{
   std::lock_guard<std::mutex> lk(mutex_);

   if (!Init()) {
      SetLastError(eError, "BHome Node Not Inited.");
      SetLastError(eError, kErrMsgNotInit);
      return;
   }
   if (nworker < 1) {
@@ -103,7 +115,7 @@
bool TopicNode::Register(ProcInfo &proc, MsgCommonReply &reply_body, const int timeout_ms)
{
   if (!Init()) {
      SetLastError(eError, "BHome Node Not Inited.");
      SetLastError(eError, kErrMsgNotInit);
      return false;
   }

@@ -153,7 +165,7 @@
bool TopicNode::Unregister(ProcInfo &proc, MsgCommonReply &reply_body, const int timeout_ms)
{
   if (!IsOnline()) {
      SetLastError(eNotRegistered, "Not Registered.");
      SetLastError(eNotRegistered, kErrMsgNotRegistered);
      return false;
   }

@@ -192,7 +204,7 @@
bool TopicNode::Heartbeat(ProcInfo &proc, MsgCommonReply &reply_body, const int timeout_ms)
{
   if (!IsOnline()) {
      SetLastError(eNotRegistered, "Not Registered.");
      SetLastError(eNotRegistered, kErrMsgNotRegistered);
      return false;
   }

@@ -225,7 +237,7 @@
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();
@@ -244,7 +256,7 @@
bool TopicNode::ServerRegisterRPC(MsgTopicList &topics, MsgCommonReply &reply_body, const int timeout_ms)
{
   if (!IsOnline()) {
      SetLastError(eNotRegistered, "Not Registered.");
      SetLastError(eNotRegistered, kErrMsgNotRegistered);
      return false;
   }

@@ -311,7 +323,7 @@
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;
   }

@@ -335,7 +347,7 @@
bool TopicNode::ServerSendReply(void *src_info, const MsgRequestTopicReply &body)
{
   if (!IsOnline()) {
      SetLastError(eNotRegistered, "Not Registered.");
      SetLastError(eNotRegistered, kErrMsgNotRegistered);
      return false;
   }

@@ -373,7 +385,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;
   }

@@ -414,7 +426,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;
   }

@@ -465,7 +477,7 @@
bool TopicNode::ClientQueryRPCTopic(const Topic &topic, BHAddress &addr, const int timeout_ms)
{
   if (!IsOnline()) {
      SetLastError(eNotRegistered, "Not Registered.");
      SetLastError(eNotRegistered, kErrMsgNotRegistered);
      return false;
   }

@@ -489,7 +501,7 @@
bool TopicNode::Publish(const MsgPublish &pub, const int timeout_ms)
{
   if (!IsOnline()) {
      SetLastError(eNotRegistered, "Not Registered.");
      SetLastError(eNotRegistered, kErrMsgNotRegistered);
      return false;
   }

@@ -520,7 +532,7 @@
bool TopicNode::Subscribe(MsgTopicList &topics, MsgCommonReply &reply_body, const int timeout_ms)
{
   if (!IsOnline()) {
      SetLastError(eNotRegistered, "Not Registered.");
      SetLastError(eNotRegistered, kErrMsgNotRegistered);
      return false;
   }

@@ -569,7 +581,7 @@
bool TopicNode::RecvSub(std::string &proc_id, MsgPublish &pub, const int timeout_ms)
{
   if (!IsOnline()) {
      SetLastError(eNotRegistered, "Not Registered.");
      SetLastError(eNotRegistered, kErrMsgNotRegistered);
      return false;
   }