change node socket to vector; try lock free queue.

lichao

2021-04-21 3931f83205f153f2bc7fc36d1a894cdc3f14b4db

 src/topic_node.cpp

@@ -37,14 +37,16 @@
} // namespace

TopicNode::TopicNode(SharedMemory &shm) :
    shm_(shm), sock_node_(shm), sock_client_(shm, kMqLen), sock_server_(shm, kMqLen), sock_sub_(shm, kMqLen), state_(eStateUnregistered)
    shm_(shm), sockets_(eSockEnd), state_(eStateUnregistered)
{
   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; };
   SockNode().Start(default_ignore_msg);
   SockClient().Start(default_ignore_msg);
   SockServer().Start(default_ignore_msg);
   SockSub().Start(default_ignore_msg);
   for (auto &p : sockets_) {
      p->Start(default_ignore_msg);
   }
}

TopicNode::~TopicNode()
@@ -52,10 +54,7 @@
   Stop();
   SockNode().Stop();
   if (state() == eStateUnregistered) {
      SockNode().Remove();
      SockClient().Remove();
      SockServer().Remove();
      SockSub().Remove();
      for (auto &p : sockets_) { p->Remove(); }
   }
}

@@ -66,16 +65,14 @@
   } else if (nworker > 16) {
      nworker = 16;
   }

   SockNode().Start();
   ServerStart(server_cb, nworker);
   SubscribeStartWorker(sub_cb, nworker);
   ClientStartWorker(client_cb, nworker);
}
void TopicNode::Stop()
{
   SockSub().Stop();
   SockServer().Stop();
   SockClient().Stop();
   for (auto &p : sockets_) { p->Stop(); }
}

bool TopicNode::Register(ProcInfo &proc, MsgCommonReply &reply_body, const int timeout_ms)