diff --git a/rmw_unix_socket_cpp/src/rmw_wait.cpp b/rmw_unix_socket_cpp/src/rmw_wait.cpp index 6b0028b..5941c35 100644 --- a/rmw_unix_socket_cpp/src/rmw_wait.cpp +++ b/rmw_unix_socket_cpp/src/rmw_wait.cpp @@ -395,6 +395,15 @@ rmw_ret_t rmw_wait( } } + // Bound the wait so an idle executor loops and re-runs the top-of-wait + // TRANSIENT_LOCAL late-joiner replay: a joining subscriber only bumps the + // shm generation counter (no fd fires), so an unbounded wait would never + // replay to it. Only shortens the timeout; a non-blocking 0 stays 0. + constexpr int TL_REPLAY_POLL_MS = 200; + if (timeout_ms < 0 || timeout_ms > TL_REPLAY_POLL_MS) { + timeout_ms = TL_REPLAY_POLL_MS; + } + // Block, retrying on EINTR. A finite timeout uses a steady_clock deadline // so a signal interruption neither returns TIMEOUT early nor busy-loops. struct epoll_event ready_events[64]; diff --git a/rmw_unix_socket_cpp/test/test_rmw_qos.cpp b/rmw_unix_socket_cpp/test/test_rmw_qos.cpp index 8695960..02571d0 100644 --- a/rmw_unix_socket_cpp/test/test_rmw_qos.cpp +++ b/rmw_unix_socket_cpp/test/test_rmw_qos.cpp @@ -14,6 +14,7 @@ #include "test_base.hpp" +#include #include #include #include @@ -975,3 +976,92 @@ TEST_F(QosTest, MultipleClientsOneService) auto _r2 [[maybe_unused]] = rmw_destroy_client(node, cli1); auto _r3 [[maybe_unused]] = rmw_destroy_service(node, srv); } + +TEST_F(QosTest, TransientLocalReplayReachesLateJoinerWhileWaitBlocked) +{ + // The subscriber joins AFTER the publisher's executor is already blocked in + // rmw_wait. A joining subscriber only bumps the shm generation counter, which + // signals no fd, so an idle rmw_wait(infinite) would block in epoll forever + // and never re-run the top-of-wait replay. The latched message must still + // reach the late joiner within a bounded time. + auto qos = make_qos( + RMW_QOS_POLICY_RELIABILITY_RELIABLE, + RMW_QOS_POLICY_DURABILITY_TRANSIENT_LOCAL, 5); + + // A service anchors ctx resolution inside rmw_wait, mirroring an idle node + // whose wait set holds only its services. + auto srv_ts = rosidl_typesupport_cpp::get_service_type_support_handle< + test_msgs::srv::BasicTypes>(); + auto svc_qos = make_qos( + RMW_QOS_POLICY_RELIABILITY_RELIABLE, RMW_QOS_POLICY_DURABILITY_VOLATILE); + auto * srv = rmw_create_service(node, srv_ts, "/idle_anchor", &svc_qos); + ASSERT_NE(nullptr, srv); + + // Guard condition only unblocks the executor thread on teardown so the test + // never hangs when the message never arrives (the failing case). + auto * gc = rmw_create_guard_condition(&context); + ASSERT_NE(nullptr, gc); + + auto pub_opts = rmw_get_default_publisher_options(); + auto * pub = rmw_create_publisher(node, ts, "/idle_replay", &qos, &pub_opts); + ASSERT_NE(nullptr, pub); + + // Publish before any subscriber exists; then the node goes idle. + test_msgs::msg::BasicTypes m; + m.int32_value = 7; + EXPECT_EQ(RMW_RET_OK, rmw_publish(pub, &m, nullptr)); + + auto * ws = rmw_create_wait_set(&context, 4); + ASSERT_NE(nullptr, ws); + + // Executor thread: spin rmw_wait with an INFINITE timeout, like an idle node. + std::atomic stop{false}; + std::thread executor( + [&] { + while (!stop.load()) { + void * srv_array[1] = {srv->data}; + rmw_services_t services; + services.services = srv_array; + services.service_count = 1; + void * gc_array[1] = {gc->data}; + rmw_guard_conditions_t gcs; + gcs.guard_conditions = gc_array; + gcs.guard_condition_count = 1; + auto _r [[maybe_unused]] = rmw_wait( + nullptr, &gcs, &services, nullptr, nullptr, ws, nullptr); + } + }); + + // Let the executor reach epoll and block before the subscriber joins. + std::this_thread::sleep_for(std::chrono::milliseconds(150)); + + // Late joiner — created after the executor is already blocked in rmw_wait. + auto sub_opts = rmw_get_default_subscription_options(); + auto * sub = rmw_create_subscription(node, ts, "/idle_replay", &qos, &sub_opts); + ASSERT_NE(nullptr, sub); + + bool got = false; + for (int i = 0; i < 300 && !got; ++i) { + test_msgs::msg::BasicTypes recv; + bool taken = false; + if (rmw_take(sub, &recv, &taken, nullptr) == RMW_RET_OK && taken && + recv.int32_value == 7) + { + got = true; + } + std::this_thread::sleep_for(std::chrono::milliseconds(10)); + } + + stop.store(true); + auto _t [[maybe_unused]] = rmw_trigger_guard_condition(gc); + executor.join(); + + EXPECT_TRUE(got) << "late joiner never received the latched message while the " + "publisher's executor was blocked in rmw_wait"; + + auto _r1 [[maybe_unused]] = rmw_destroy_subscription(node, sub); + auto _r2 [[maybe_unused]] = rmw_destroy_wait_set(ws); + auto _r3 [[maybe_unused]] = rmw_destroy_publisher(node, pub); + auto _r4 [[maybe_unused]] = rmw_destroy_guard_condition(gc); + auto _r5 [[maybe_unused]] = rmw_destroy_service(node, srv); +}