socket send using abs addr, avoid shm find by id.

lichao

2021-05-17 cab831748a2a9cc18b7f18f3b5e14a4374b7ab68

 src/sendq.cpp

@@ -21,6 +21,24 @@

using namespace bhome_shm;

void SendQ::AppendData(const MQInfo &mq, const Data data, const TimePoint &expire, OnMsgEvent onExpire)
{
   TimedMsg tmp(expire, MsgInfo{mq, data, std::move(onExpire)});
   std::unique_lock<std::mutex> lock(mutex_in_);

   try {
      auto &al = in_[mq.id_];
      if (!al.empty()) {
         al.front().emplace_back(std::move(tmp));
      } else {
         al.insert(al.begin(), Array())->emplace_back(std::move(tmp));
      }
   } catch (std::exception &e) {
      LOG_ERROR() << "sendq error: " << e.what();
      throw e;
   }
}

int SendQ::DoSend1Remote(ShmMsgQueue &mq, const Remote remote, Array &arr)
{
   auto FirstNotExpired = [](Array &l) {
@@ -36,7 +54,7 @@
      }
   }

   while (pos != arr.end() && mq.TrySend(remote, pos->data().data_)) {
   while (pos != arr.end() && mq.TrySend(pos->data().mq_, pos->data().data_)) {
      ++pos;
   }

@@ -59,6 +77,8 @@
bool SendQ::TrySend(ShmMsgQueue &mq)
{
   std::unique_lock<std::mutex> lock(mutex_out_);
   // if (TooFast()) { return false; }

   size_t nsend = 0;
   if (!out_.empty()) {
      auto rec = out_.begin();
@@ -89,3 +109,14 @@

   return !out_.empty();
}

bool SendQ::TooFast()
{
   auto cur = NowSec();
   if (cur > last_time_) {
      last_time_ = cur;
      count_ = 0;
   }

   return ++count_ > 1000 * 100;
} // not accurate in multi-thread.