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/socket.h | 44 +++++++++++---------------------------------
1 files changed, 11 insertions(+), 33 deletions(-)
diff --git a/src/socket.h b/src/socket.h
index 8e9db69..dea106c 100644
--- a/src/socket.h
+++ b/src/socket.h
@@ -59,7 +59,6 @@
{
node_proc_index_ = proc_index;
socket_index_ = socket_index;
- LOG_DEBUG() << "Set Node Proc " << node_proc_index_ << ", " << socket_index_;
}
// start recv.
bool Start(int nworker = 1, const RecvCB &onMsg = RecvCB(), const RawRecvCB &onRaw = RawRecvCB(), const IdleCB &onIdle = IdleCB());
@@ -68,7 +67,7 @@
bool Stop();
template <class Body>
- bool CenterSend(const MQId remote, BHMsgHead &head, Body &body)
+ bool CenterSend(const MQInfo &remote, BHMsgHead &head, Body &body)
{
try {
//TODO alloc outsiez and use send.
@@ -86,39 +85,17 @@
bool RequestAlloc(const int64_t size, std::function<void(MsgI &msg)> const &onResult);
template <class Body>
- bool Send(const MQId remote, BHMsgHead &head, Body &body, RecvCB &&cb = RecvCB())
+ bool Send(const MQInfo &remote, BHMsgHead &head, Body &body, RecvCB &&cb = RecvCB())
{
- std::string msg_id(head.msg_id());
- std::string content(MsgI::Serialize(head, body));
- size_t size = content.size();
- auto OnResult = [content = std::move(content), msg_id, remote, cb = std::move(cb), this](MsgI &msg) mutable {
- if (!msg.Fill(content)) { return; }
-
- try {
- if (!cb) {
- Send(remote, msg);
- } else {
- per_msg_cbs_->Store(msg_id, std::move(cb));
- auto onExpireRemoveCB = [this, msg_id](SendQ::Data const &msg) {
- RecvCB cb_no_use;
- per_msg_cbs_->Pick(msg_id, cb_no_use);
- };
- Send(remote, msg, onExpireRemoveCB);
- }
- } catch (...) {
- SetLastError(eError, "Send internal error.");
- }
- };
-
- return RequestAlloc(size, OnResult);
+ return Send(remote, MsgI::Serialize(head, body), head.msg_id(), std::move(cb));
}
template <class... T>
- bool Send(const MQId remote, const MsgI &imsg, T &&...t)
+ bool Send(const MQInfo &remote, const MsgI &imsg, T &&...t)
{
return SendImpl(remote, imsg, std::forward<decltype(t)>(t)...);
}
template <class... T>
- bool Send(const MQId remote, const int64_t cmd, T &&...t)
+ bool Send(const MQInfo &remote, const int64_t cmd, T &&...t)
{
return SendImpl(remote, cmd, std::forward<decltype(t)>(t)...);
}
@@ -126,7 +103,7 @@
bool SyncRecv(MsgI &msg, bhome_msg::BHMsgHead &head, const int timeout_ms);
template <class Body>
- bool SendAndRecv(const MQId remote, BHMsgHead &head, Body &body, MsgI &reply, BHMsgHead &reply_head, const int timeout_ms)
+ bool SendAndRecv(const MQInfo &remote, BHMsgHead &head, Body &body, MsgI &reply, BHMsgHead &reply_head, const int timeout_ms)
{
struct State {
std::mutex mutex;
@@ -136,6 +113,7 @@
try {
std::shared_ptr<State> st(new State);
+
auto endtime = std::chrono::steady_clock::now() + std::chrono::milliseconds(timeout_ms);
auto OnRecv = [st, &reply, &reply_head](ShmSocket &sock, MsgI &msg, BHMsgHead &head) {
@@ -176,12 +154,12 @@
bool StopNoLock();
bool RunningNoLock() { return !workers_.empty(); }
+ bool Send(const MQInfo &remote, std::string &&content, const std::string &msg_id, RecvCB &&cb = RecvCB());
+
template <class... Rest>
- bool SendImpl(const MQId remote, Rest &&...rest)
+ bool SendImpl(const MQInfo &remote, Rest &&...rest)
{
- // TODO send alloc request, and pack later, higher bit means alloc?
- send_buffer_.Append(remote, std::forward<decltype(rest)>(rest)...);
- return true;
+ return send_buffer_.Append(remote, std::forward<decltype(rest)>(rest)...);
}
std::vector<std::thread> workers_;
--
Gitblit v1.8.0