cv-rmvl · TooPretty0108 · Jul 25, 2026 · Jul 25, 2026
diff --git 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
@@ -233,6 +233,21 @@ rm::async::Task<> Client<SrvType>::receive() {
     }
 }
 
+template <typename SrvType>
+template <typename Rep, typename Period>
+rm::async::Task<bool> Client<SrvType>::wait(std::chrono::duration<Rep, Period> 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 <typename SrvType>
 template <typename Rep, typename Period>
 auto Client<SrvType>::call(const Request &request, std::chrono::duration<Rep, Period> timeout) -> rm::async::Task<std::optional<Response>> {

diff --git 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<LatestBytesSHM> shm{};  //!< 最新字节流共享内存
+    std::string name{};                    //!< 共享内存通道名称
+    Locator locator{};                     //!< 目标定位器，用于发送唤醒通知
+    std::shared_ptr<LatestBytesSHM> 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<MTPAsmKey, MTPAsm, MTPAsmKeyHash> _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<MTPAsmKey, MTPAsm, MTPAsmKeyHash> _asms{};    //!< MTP 重组缓存
+    std::size_t _asm_bytes{};                                        //!< 待重组载荷占用字节数
+    std::shared_mutex _shm_mtx{};                                    //!< 保护共享内存读取源
     std::unordered_map<Guid, MTPShmSource, GuidHash> _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<Guid, MTPWriterTarget, GuidHash> _udpv4_targets;
     //! 目标共享内存通道缓存集合
     std::unordered_map<Guid, MTPShmTarget, GuidHash> _shm_targets;
-    std::atomic_uint16_t _sequence{};       //!< MTP 发送序列号
-    std::optional<std::string> _pending{};  //!< 发送中收到的最新待发送消息
-    bool _sending{};                        //!< 是否已有发送协程正在运行
+    std::atomic_uint16_t _sequence{};      //!< MTP 发送序列号
+    std::optional<std::string> _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<MTPAsmKey, MTPAsm, MTPAsmKeyHash> _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<MTPAsmKey, MTPAsm, MTPAsmKeyHash> _asms{};    //!< MTP 重组缓存
+    std::size_t _asm_bytes{};                                        //!< 待重组载荷占用字节数
+    std::unordered_set<Guid, GuidHash> _matched_writers{};           //!< 已匹配数据写入端点
     std::unordered_map<Guid, MTPShmSource, GuidHash> _shm_sources{}; //!< 共享内存读取源缓存集合
 };
 

diff --git 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<Client<SrvType>> {
     //! 判断客户端是否无效
     bool invalid() const noexcept { return !_request_writer || !_response_reader; }
 
+    /**
+     * @brief 等待服务端上线
+     *
+     * @param[in] timeout 最大等待时间
+     * @return 服务端在超时前上线返回 `true`，否则返回 `false`
+    */
+    template <typename Rep, typename Period>
+    rm::async::Task<bool> wait(std::chrono::duration<Rep, Period> 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
@@ -215,14 +215,11 @@ static void sendStopMessage(const std::unordered_map<Guid, NodeStorageInfo, Guid
             sendREDPMessage(node_info.ctrl_loc, msg);
 }
 
-rm::async::Task<> 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
@@ -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<std::string> DataReaderBase::read() noexcept {
     while (true) {

diff --git 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<srv::SetBool::Response> result{};
-    auto call = [](lpss::async::Client<srv::SetBool>::ptr client, srv::SetBool::Request request,
+    auto call = [](lpss::async::Client<srv::SetBool>::ptr client, srv::SetBool::Request request, bool *service_ready,
                    std::optional<srv::SetBool::Response> *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");