From cedb2172ba45cb20e9c8ab10372e09750d39f2d1 Mon Sep 17 00:00:00 2001 From: zhaoxi <535394140@qq.com> Date: Sun, 26 Jul 2026 01:07:43 +0800 Subject: [PATCH] =?UTF-8?q?feat(lpss):=20=E5=A2=9E=E5=8A=A0=E6=96=B0?= =?UTF-8?q?=E7=9A=84=20Client=20=E6=8E=A5=E5=8F=A3=E5=92=8C=E8=8A=82?= =?UTF-8?q?=E7=82=B9=E6=9C=BA=E5=88=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 1. DataReaderBase 与 DataWriterBase 增加匹配判断函数 2. Client 增加 wait 接口 3. 异步节点下线的信号监听增加 SIGTERM --- cmake/RMVLCompilerImages.cmake | 2 +- .../include/rmvl/lpss/details/node_impl.hpp | 15 ++++++ .../include/rmvl/lpss/details/node_rmtp.hpp | 49 +++++++++++-------- modules/lpss/include/rmvl/lpss/node.hpp | 15 ++++-- modules/lpss/src/node_async.cpp | 18 +++---- modules/lpss/src/node_rmtp.cpp | 8 ++- modules/lpss/test/test_node.cpp | 11 ++++- 7 files changed, 81 insertions(+), 37 deletions(-) diff --git a/cmake/RMVLCompilerImages.cmake b/cmake/RMVLCompilerImages.cmake index 41ae020d..8887b53e 100644 --- a/cmake/RMVLCompilerImages.cmake +++ b/cmake/RMVLCompilerImages.cmake @@ -43,7 +43,7 @@ foreach(line ${parse_result}) math(EXPR img_num "${img_num} + 1") add_custom_command( - TARGET run_in_list + TARGET run_in_list POST_BUILD COMMAND ${CMAKE_COMMAND} -E cmake_echo_color --blue "${img_num}. run_in_${name}: ${description}" COMMAND ${CMAKE_COMMAND} -E echo " image: ${image}" COMMAND ${CMAKE_COMMAND} -E echo " options: ${options}" diff --git a/modules/lpss/include/rmvl/lpss/details/node_impl.hpp b/modules/lpss/include/rmvl/lpss/details/node_impl.hpp index 00006d21..09c2af69 100644 --- a/modules/lpss/include/rmvl/lpss/details/node_impl.hpp +++ b/modules/lpss/include/rmvl/lpss/details/node_impl.hpp @@ -233,6 +233,21 @@ rm::async::Task<> Client::receive() { } } +template +template +rm::async::Task Client::wait(std::chrono::duration timeout) { + if (invalid()) + co_return false; + + rm::async::Timer timer(_ctx); + const auto deadline = std::chrono::steady_clock::now() + timeout; + while ((!_request_writer->matched() || !_response_reader->matched()) && + std::chrono::steady_clock::now() < deadline) + co_await timer.sleep_for(std::chrono::milliseconds(1)); + + co_return _request_writer->matched() && _response_reader->matched(); +} + template template auto Client::call(const Request &request, std::chrono::duration timeout) -> rm::async::Task> { diff --git a/modules/lpss/include/rmvl/lpss/details/node_rmtp.hpp b/modules/lpss/include/rmvl/lpss/details/node_rmtp.hpp index 2490cc0b..c050dea5 100644 --- a/modules/lpss/include/rmvl/lpss/details/node_rmtp.hpp +++ b/modules/lpss/include/rmvl/lpss/details/node_rmtp.hpp @@ -37,9 +37,9 @@ struct MTPWriterTarget { //! MTP 共享内存写入目标 struct MTPShmTarget { - std::string name{}; //!< 共享内存通道名称 - Locator locator{}; //!< 目标定位器，用于发送唤醒通知 - std::shared_ptr shm{}; //!< 最新字节流共享内存 + std::string name{}; //!< 共享内存通道名称 + Locator locator{}; //!< 目标定位器，用于发送唤醒通知 + std::shared_ptr shm{}; //!< 最新字节流共享内存 }; //! MTP 共享内存读取源 @@ -193,14 +193,14 @@ class DataReaderBase { void remove(const Guid &guid) noexcept; protected: - uint16_t _port{}; //!< 监听端口 - Guid _guid; //!< 读取器所属实体 GUID - DgramSocket _udpv4; //!< UDPv4 通道 - std::string_view _type{}; //!< 消息类型 - std::string _topic{}; //!< 监听话题 - std::unordered_map _asms{}; //!< MTP 重组缓存 - std::size_t _asm_bytes{}; //!< 待重组载荷占用字节数 - std::shared_mutex _shm_mtx{}; //!< 保护共享内存读取源 + uint16_t _port{}; //!< 监听端口 + Guid _guid; //!< 读取器所属实体 GUID + DgramSocket _udpv4; //!< UDPv4 通道 + std::string_view _type{}; //!< 消息类型 + std::string _topic{}; //!< 监听话题 + std::unordered_map _asms{}; //!< MTP 重组缓存 + std::size_t _asm_bytes{}; //!< 待重组载荷占用字节数 + std::shared_mutex _shm_mtx{}; //!< 保护共享内存读取源 std::unordered_map _shm_sources{}; //!< 共享内存读取源缓存集合 }; @@ -282,6 +282,9 @@ class DataWriterBase { //! 获取写入话题的消息类型 inline std::string_view msgtype() const noexcept { return _type; } + //! 是否已匹配数据接收端点 + inline bool matched() const noexcept { return !_udpv4_targets.empty() || !_shm_targets.empty(); } + /** * @brief 添加数据接收端点 * @@ -314,9 +317,9 @@ class DataWriterBase { std::unordered_map _udpv4_targets; //! 目标共享内存通道缓存集合 std::unordered_map _shm_targets; - std::atomic_uint16_t _sequence{}; //!< MTP 发送序列号 - std::optional _pending{}; //!< 发送中收到的最新待发送消息 - bool _sending{}; //!< 是否已有发送协程正在运行 + std::atomic_uint16_t _sequence{}; //!< MTP 发送序列号 + std::optional _pending{}; //!< 发送中收到的最新待发送消息 + bool _sending{}; //!< 是否已有发送协程正在运行 }; /** @@ -358,6 +361,9 @@ class DataReaderBase { //! 获取监听的端口 inline uint16_t port() const noexcept { return _port; } + //! 是否已匹配数据写入端点 + inline bool matched() const noexcept { return !_matched_writers.empty(); } + /** * @brief 添加数据写入端点 * @@ -373,13 +379,14 @@ class DataReaderBase { void remove(const Guid &guid) noexcept; protected: - uint16_t _port{}; //!< 监听端口 - Guid _guid; //!< 读取器所属实体 GUID - rm::async::DgramSocket _udpv4; //!< UDPv4 通道 - std::string_view _type{}; //!< 消息类型 - std::string _topic{}; //!< 监听话题 - std::unordered_map _asms{}; //!< MTP 重组缓存 - std::size_t _asm_bytes{}; //!< 待重组载荷占用字节数 + uint16_t _port{}; //!< 监听端口 + Guid _guid; //!< 读取器所属实体 GUID + rm::async::DgramSocket _udpv4; //!< UDPv4 通道 + std::string_view _type{}; //!< 消息类型 + std::string _topic{}; //!< 监听话题 + std::unordered_map _asms{}; //!< MTP 重组缓存 + std::size_t _asm_bytes{}; //!< 待重组载荷占用字节数 + std::unordered_set _matched_writers{}; //!< 已匹配数据写入端点 std::unordered_map _shm_sources{}; //!< 共享内存读取源缓存集合 }; diff --git a/modules/lpss/include/rmvl/lpss/node.hpp b/modules/lpss/include/rmvl/lpss/node.hpp index 5a354f6f..e80b4aac 100644 --- a/modules/lpss/include/rmvl/lpss/node.hpp +++ b/modules/lpss/include/rmvl/lpss/node.hpp @@ -480,6 +480,15 @@ class Client final : public std::enable_shared_from_this> { //! 判断客户端是否无效 bool invalid() const noexcept { return !_request_writer || !_response_reader; } + /** + * @brief 等待服务端上线 + * + * @param[in] timeout 最大等待时间 + * @return 服务端在超时前上线返回 `true`，否则返回 `false` + */ + template + rm::async::Task wait(std::chrono::duration timeout); + /** * @brief 调用服务并异步等待响应 * @@ -646,14 +655,14 @@ class Node { //! 心跳检测 rm::async::Task<> heartbeat_detect(); - //! 处理 SIGINT 信号 - rm::async::Task<> on_sigint(); + //! 处理要求节点正常退出的信号 + rm::async::Task<> on_shutdown_signal(int signum); protected: rm::async::IOContext _ctx{}; //!< 异步 IO 上下文 private: - bool _running{true}; //!< 运行状态 + std::atomic_bool _running{true}; //!< 运行状态 uint16_t _next_eid{1}; //!< 用于生成实体 ID 的计数器 uint16_t _rndp_port{}; //!< RNDP 广播端口号 diff --git a/modules/lpss/src/node_async.cpp b/modules/lpss/src/node_async.cpp index fc6d8ffd..01d1a8d5 100644 --- a/modules/lpss/src/node_async.cpp +++ b/modules/lpss/src/node_async.cpp @@ -215,14 +215,11 @@ static void sendStopMessage(const std::unordered_map Node::on_sigint() { - rm::async::Signal sig(_ctx, SIGINT); +rm::async::Task<> Node::on_shutdown_signal(int signum) { + rm::async::Signal sig(_ctx, signum); co_await sig.wait(); - printf("\nReceived interrupt signal, stopping node...\n"); - sendStopMessage(_discovered_nodes, _local_writers, _local_readers); - _local_readers.clear(); - _local_writers.clear(); - _ctx.stop(); + printf("\nReceived %s, stopping node...\n", signum == SIGINT ? "SIGINT" : "SIGTERM"); + shutdown(); } Node::Node(std::string_view name, uint8_t domain_id) : _rndp_port(7500 + domain_id), _rndp_writer(rm::async::Sender(_ctx, ip::udp::v4()).create()) { @@ -252,11 +249,14 @@ Node::Node(std::string_view name, uint8_t domain_id) : _rndp_port(7500 + domain_ // 启动心跳检测协程任务 co_spawn(_ctx, &Node::heartbeat_detect, this); - // 启动 SIGINT 信号处理协程任务 - co_spawn(_ctx, &Node::on_sigint, this); + // 启动正常退出信号处理协程任务 + co_spawn(_ctx, &Node::on_shutdown_signal, this, SIGINT); + co_spawn(_ctx, &Node::on_shutdown_signal, this, SIGTERM); } void Node::shutdown() noexcept { + if (!_running.exchange(false, std::memory_order_acq_rel)) + return; sendStopMessage(_discovered_nodes, _local_writers, _local_readers); _local_readers.clear(); _local_writers.clear(); diff --git a/modules/lpss/src/node_rmtp.cpp b/modules/lpss/src/node_rmtp.cpp index 1cfcecab..ac9cf4d6 100644 --- a/modules/lpss/src/node_rmtp.cpp +++ b/modules/lpss/src/node_rmtp.cpp @@ -601,13 +601,19 @@ DataReaderBase::DataReaderBase(rm::async::IOContext &io_context, const Guid &gui } void DataReaderBase::add(const Guid &guid) noexcept { + _matched_writers.insert(guid); if (!same_host(_guid, guid)) return; auto name = shm_channel_name(guid, _guid); _shm_sources[guid] = {name, create_shm_channel(name), 0}; } -void DataReaderBase::remove(const Guid &guid) noexcept { erase_endpoint_or_node(_shm_sources, guid); } +void DataReaderBase::remove(const Guid &guid) noexcept { + if (_matched_writers.erase(guid) == 0) + for (auto it = _matched_writers.begin(); it != _matched_writers.end();) + same_node(*it, guid) ? it = _matched_writers.erase(it) : ++it; + erase_endpoint_or_node(_shm_sources, guid); +} rm::async::Task DataReaderBase::read() noexcept { while (true) { diff --git a/modules/lpss/test/test_node.cpp b/modules/lpss/test/test_node.cpp index da3ab3f2..d7f9a286 100644 --- a/modules/lpss/test/test_node.cpp +++ b/modules/lpss/test/test_node.cpp @@ -379,15 +379,22 @@ TEST(LPSS_node, async_service_client_local_call) { srv::SetBool::Request request{}; request.data = true; + bool service_ready{}; std::optional result{}; - auto call = [](lpss::async::Client::ptr client, srv::SetBool::Request request, + auto call = [](lpss::async::Client::ptr client, srv::SetBool::Request request, bool *service_ready, std::optional *result, rm::async::IOContext *ctx) -> rm::async::Task<> { + *service_ready = co_await client->wait(100ms); + if (!*service_ready) { + ctx->stop(); + co_return; + } *result = co_await client->call(request, 500ms); ctx->stop(); }; - co_spawn(io_context, call, client, request, &result, &io_context); + co_spawn(io_context, call, client, request, &service_ready, &result, &io_context); io_context.run(); + EXPECT_TRUE(service_ready); ASSERT_TRUE(result.has_value()); EXPECT_TRUE(result->success); EXPECT_EQ(result->message, "enabled");