From 4353f73ea0c30c776a3957dc674d750e51519ca3 Mon Sep 17 00:00:00 2001
From: lichao <lichao@aiotlink.com>
Date: 星期三, 31 三月 2021 15:02:19 +0800
Subject: [PATCH] add queue length status.
---
src/socket.h | 1 +
src/shm_queue.h | 1 +
src/socket.cpp | 1 -
src/pubsub.cpp | 11 ++++++++++-
utest/utest.cpp | 16 ++++++++++------
5 files changed, 22 insertions(+), 8 deletions(-)
diff --git a/src/pubsub.cpp b/src/pubsub.cpp
index 52285b1..eff54ab 100644
--- a/src/pubsub.cpp
+++ b/src/pubsub.cpp
@@ -34,6 +34,15 @@
bool BusManager::Start(const int nworker)
{
auto onRecv = [&](MsgI &imsg) {
+#ifndef NDEBUG
+ static std::atomic<time_t> last(0);
+ time_t now = 0;
+ time(&now);
+ if (last.exchange(now) < now) {
+ printf("bus queue size: %ld\n", socket_.Pending());
+ }
+#endif
+
BHMsg msg;
if (!imsg.Unpack(msg)) {
return;
@@ -130,7 +139,7 @@
}
};
- return socket_.StartRaw(onRecv, std::min(nworker, kMaxWorker));
+ return socket_.StartRaw(onRecv, std::min((nworker > 0 ? nworker : 2), kMaxWorker));
}
} // namespace bhome_shm
diff --git a/src/shm_queue.h b/src/shm_queue.h
index 9064f55..f1e67f3 100644
--- a/src/shm_queue.h
+++ b/src/shm_queue.h
@@ -138,6 +138,7 @@
{
return Send(shm(), remote_id, msg, timeout_ms);
}
+ size_t Pending() const { return data()->size(); }
};
} // namespace bhome_shm
diff --git a/src/socket.cpp b/src/socket.cpp
index 5eb6756..4d9fcc9 100644
--- a/src/socket.cpp
+++ b/src/socket.cpp
@@ -63,7 +63,6 @@
}
DEFER1(imsg.Release(shm_));
return Queue::Send(shm_, kBHBusQueueId, imsg, timeout_ms);
-
} catch (...) {
return false;
}
diff --git a/src/socket.h b/src/socket.h
index b94eca2..c468dd3 100644
--- a/src/socket.h
+++ b/src/socket.h
@@ -61,6 +61,7 @@
bool Start(const RecvCB &onData, int nworker = 1);
bool StartRaw(const RecvRawCB &onData, int nworker = 1);
bool Stop();
+ size_t Pending() const { return mq_ ? mq_->Pending() : 0; }
private:
bool StopNoLock();
diff --git a/utest/utest.cpp b/utest/utest.cpp
index e24d34a..fbe4d51 100644
--- a/utest/utest.cpp
+++ b/utest/utest.cpp
@@ -66,20 +66,24 @@
BOOST_AUTO_TEST_CASE(PubSubTest)
{
const std::string shm_name("ShmPubSub");
- ShmRemover auto_remove(shm_name); //remove twice? in case of killed?
+ // ShmRemover auto_remove(shm_name); //remove twice? in case of killed?
SharedMemory shm(shm_name, 1024 * 1024 * 50);
+ DEFER1(shm.Remove());
auto Avail = [&]() { return shm.get_free_memory(); };
auto init_avail = Avail();
+ int *flag = shm.find_or_construct<int>("flag")(123);
+ printf("flag = %d\n", *flag);
+ ++*flag;
BusManager bus(shm);
- bus.Start(1);
+ bus.Start();
std::this_thread::sleep_for(100ms);
std::atomic<uint64_t> count(0);
std::atomic<ptime> last_time(Now() - seconds(1));
std::atomic<uint64_t> last_count(0);
- const uint64_t nmsg = 100;
+ const uint64_t nmsg = 100 * 2;
const int timeout = 1000;
auto Sub = [&](int id, const std::vector<std::string> &topics) {
ShmSocket client(ShmSocket::eSockSubscribe, shm);
@@ -87,7 +91,7 @@
std::mutex mutex;
std::condition_variable cv;
- int i = 0;
+ uint64_t i = 0;
auto OnRecv = [&](BHMsg &msg) {
if (msg.type() != kMsgTypePublish) {
BOOST_CHECK(false);
@@ -117,7 +121,7 @@
auto Pub = [&](const std::string &topic) {
ShmSocket provider(ShmSocket::eSockPublish, shm);
- for (int i = 0; i < nmsg; ++i) {
+ for (unsigned i = 0; i < nmsg; ++i) {
std::string data = topic + std::to_string(i) + std::string(1000, '-');
bool r = provider.Publish(topic, data.data(), data.size(), timeout);
@@ -134,7 +138,7 @@
topics.push_back("t" + std::to_string(i));
}
Topics part;
- for (int i = 0; i < topics.size(); ++i) {
+ for (size_t i = 0; i < topics.size(); ++i) {
part.push_back(topics[i]);
threads.Launch(Sub, i, topics);
}
--
Gitblit v1.8.0