From c6964d5af25d4ec7ed9dbe7674dc4e3896b36ead Mon Sep 17 00:00:00 2001
From: lichao <lichao@aiotlink.com>
Date: 星期五, 16 四月 2021 16:10:02 +0800
Subject: [PATCH] node remove mq if never registered; refactor.
---
utest/speed_test.cpp | 30 ++++++++++++++++--------------
1 files changed, 16 insertions(+), 14 deletions(-)
diff --git a/utest/speed_test.cpp b/utest/speed_test.cpp
index d777f91..52fe824 100644
--- a/utest/speed_test.cpp
+++ b/utest/speed_test.cpp
@@ -26,7 +26,7 @@
ShmRemover auto_remove(shm_name);
const int mem_size = 1024 * 1024 * 50;
MQId id = boost::uuids::random_generator()();
- const int timeout = 100;
+ const int timeout = 1000;
const uint32_t data_size = 4000;
const std::string proc_id = "demo_proc";
@@ -40,10 +40,10 @@
body.set_data(str);
auto head(InitMsgHead(GetType(body), proc_id));
msg.MakeRC(shm, head, body);
+ assert(msg.IsCounted());
DEFER1(msg.Release(shm););
for (uint64_t i = 0; i < n; ++i) {
- // mq.Send(id, str.data(), str.size(), timeout);
mq.Send(id, msg, timeout);
}
};
@@ -90,6 +90,7 @@
www.Launch(Writer, i, nmsg);
}
www.WaitAll();
+ printf("writer finished\n");
run.store(false);
rrr.WaitAll();
printf("Write %ld msg R(%3d) W(%3d), : ", total_msg, nreader, nwriter);
@@ -127,22 +128,26 @@
SharedMemory shm(shm_name, 1024 * 1024 * 50);
auto Avail = [&]() { return shm.get_free_memory(); };
auto init_avail = Avail();
- ShmMsgQueue srv(shm, qlen);
- ShmMsgQueue cli(shm, qlen);
+ ShmSocket srv(shm, qlen);
+ ShmSocket cli(shm, qlen);
MsgI request_rc;
MsgRequestTopic req_body;
req_body.set_topic("topic");
req_body.set_data(msg_content);
auto req_head(InitMsgHead(GetType(req_body), client_proc_id));
+ req_head.add_route()->set_mq_id(&cli.id(), cli.id().size());
request_rc.MakeRC(shm, req_head, req_body);
+ DEFER1(request_rc.Release(shm));
MsgRequestTopic reply_body;
reply_body.set_topic("topic");
reply_body.set_data(msg_content);
auto reply_head(InitMsgHead(GetType(reply_body), server_proc_id));
+ reply_head.add_route()->set_mq_id(&srv.id(), srv.id().size());
MsgI reply_rc;
reply_rc.MakeRC(shm, reply_head, reply_body);
+ DEFER1(reply_rc.Release(shm));
std::atomic<uint64_t> count(0);
@@ -156,9 +161,9 @@
req_body.set_topic("topic");
req_body.set_data(msg_content);
auto req_head(InitMsgHead(GetType(req_body), client_proc_id));
- return cli.Send(srv.Id(), req_head, req_body, 100);
+ return cli.Send(&srv.id(), req_head, req_body);
};
- auto ReqRC = [&]() { return cli.Send(srv.Id(), request_rc, 1000); };
+ auto ReqRC = [&]() { return cli.Send(&srv.id(), request_rc); };
if (!ReqRC()) {
printf("********** client send error.\n");
@@ -166,7 +171,7 @@
}
MsgI msg;
BHMsgHead head;
- if (!cli.Recv(msg, 1000)) {
+ if (!cli.SyncRecv(msg, head, 1000)) {
printf("********** client recv error.\n");
} else {
DEFER1(msg.Release(shm));
@@ -187,8 +192,9 @@
BHMsgHead req_head;
while (!stop) {
- if (srv.Recv(req, 100)) {
+ if (srv.SyncRecv(req, req_head, 10)) {
DEFER1(req.Release(shm));
+
if (req.ParseHead(req_head) && req_head.type() == kMsgTypeRequestTopic) {
auto &mqid = req_head.route()[0].mq_id();
MQId src_id;
@@ -198,9 +204,9 @@
reply_body.set_topic("topic");
reply_body.set_data(msg_content);
auto reply_head(InitMsgHead(GetType(reply_body), server_proc_id, req_head.msg_id()));
- return srv.Send(src_id, reply_head, reply_body, 100);
+ return srv.Send(&src_id, reply_head, reply_body);
};
- auto ReplyRC = [&]() { return srv.Send(src_id, reply_rc, 100); };
+ auto ReplyRC = [&]() { return srv.Send(&src_id, reply_rc); };
if (ReplyRC()) {
}
@@ -222,9 +228,5 @@
printf("request ok: %ld\n", count.load());
stop = true;
servers.WaitAll();
- BOOST_CHECK(request_rc.IsCounted());
- BOOST_CHECK_EQUAL(request_rc.Count(), 1);
- request_rc.Release(shm);
- BOOST_CHECK(!request_rc.IsCounted());
// BOOST_CHECK_THROW(reply.Count(), int);
}
--
Gitblit v1.8.0