From cab831748a2a9cc18b7f18f3b5e14a4374b7ab68 Mon Sep 17 00:00:00 2001
From: lichao <lichao@aiotlink.com>
Date: 星期一, 17 五月 2021 18:34:26 +0800
Subject: [PATCH] socket send using abs addr, avoid shm find by id.
---
src/sendq.cpp | 33 ++++++++++++++++++++++++++++++++-
1 files changed, 32 insertions(+), 1 deletions(-)
diff --git a/src/sendq.cpp b/src/sendq.cpp
index 1eaefe6..f1e5918 100644
--- a/src/sendq.cpp
+++ b/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.
\ No newline at end of file
--
Gitblit v1.8.0