From 47affe737816c21e5836eb18fdf381e7c6573d61 Mon Sep 17 00:00:00 2001 From: acuadros95 Date: Tue, 25 Jul 2023 15:32:48 +0200 Subject: [PATCH 1/8] Run session if any session is present Signed-off-by: acuadros95 --- rmw_microxrcedds_c/src/rmw_wait.c | 43 +++---------------------------- 1 file changed, 3 insertions(+), 40 deletions(-) diff --git a/rmw_microxrcedds_c/src/rmw_wait.c b/rmw_microxrcedds_c/src/rmw_wait.c index e70c9d17..297259b7 100644 --- a/rmw_microxrcedds_c/src/rmw_wait.c +++ b/rmw_microxrcedds_c/src/rmw_wait.c @@ -60,42 +60,15 @@ rmw_wait( rmw_uxrce_clean_expired_static_input_buffer(); - // Clear run flag for all sessions - rmw_uxrce_mempool_item_t * item = session_memory.allocateditems; - while (item != NULL) { - rmw_context_impl_t * custom_context = (rmw_context_impl_t *)item->data; - custom_context->need_to_be_ran = false; - item = item->next; - } - - // TODO(pablogs9): What happens if there already data in one entity? - // Enable flag for every XRCE session available in the entities - for (size_t i = 0; services && i < services->service_count; ++i) { - rmw_uxrce_service_t * custom_service = (rmw_uxrce_service_t *)services->services[i]; - custom_service->owner_node->context->need_to_be_ran = true; - } - - for (size_t i = 0; clients && i < clients->client_count; ++i) { - rmw_uxrce_client_t * custom_client = (rmw_uxrce_client_t *)clients->clients[i]; - custom_client->owner_node->context->need_to_be_ran = true; - } - - for (size_t i = 0; subscriptions && i < subscriptions->subscriber_count; ++i) { - rmw_uxrce_subscription_t * custom_subscription = - (rmw_uxrce_subscription_t *)subscriptions->subscribers[i]; - custom_subscription->owner_node->context->need_to_be_ran = true; - } - // Count sessions to be ran uint8_t available_contexts = 0; item = session_memory.allocateditems; while (item != NULL) { - rmw_context_impl_t * custom_context = (rmw_context_impl_t *)item->data; - available_contexts += custom_context->need_to_be_ran ? 1 : 0; + available_contexts++; item = item->next; } - // There is no context that contais any of the wait set entities. Nothing to wait here. + // Spin all available contexts to atleast handle session metatraffic if (available_contexts != 0) { int32_t per_session_timeout = (timeout.i32 == UXR_TIMEOUT_INF) ? UXR_TIMEOUT_INF : @@ -104,17 +77,7 @@ rmw_wait( item = session_memory.allocateditems; while (item != NULL) { rmw_context_impl_t * custom_context = (rmw_context_impl_t *)item->data; - if (custom_context->need_to_be_ran) { - uxr_run_session_until_data(&custom_context->session, per_session_timeout); - } - item = item->next; - } - } else { - // Spin with no blocking to handle session metatraffic - item = session_memory.allocateditems; - while (item != NULL) { - rmw_context_impl_t * custom_context = (rmw_context_impl_t *)item->data; - uxr_run_session_timeout(&custom_context->session, 0); + uxr_run_session_until_data(&custom_context->session, per_session_timeout); item = item->next; } } From b92c82ac2030fc6db93724b9eaa8bf4f9af7d5a5 Mon Sep 17 00:00:00 2001 From: acuadros95 Date: Tue, 25 Jul 2023 15:49:16 +0200 Subject: [PATCH 2/8] Fix var type Signed-off-by: acuadros95 --- rmw_microxrcedds_c/src/rmw_wait.c | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/rmw_microxrcedds_c/src/rmw_wait.c b/rmw_microxrcedds_c/src/rmw_wait.c index 297259b7..290125da 100644 --- a/rmw_microxrcedds_c/src/rmw_wait.c +++ b/rmw_microxrcedds_c/src/rmw_wait.c @@ -62,7 +62,7 @@ rmw_wait( // Count sessions to be ran uint8_t available_contexts = 0; - item = session_memory.allocateditems; + rmw_uxrce_mempool_item_t * item = session_memory.allocateditems; while (item != NULL) { available_contexts++; item = item->next; From 19b7d8a161344e52d0d8b8ef180c15c8326c26b4 Mon Sep 17 00:00:00 2001 From: acuadros95 Date: Tue, 25 Jul 2023 16:28:05 +0200 Subject: [PATCH 3/8] Modify timeout approach Signed-off-by: acuadros95 --- rmw_microxrcedds_c/src/rmw_wait.c | 51 ++++++++++++++++++++++++++++--- 1 file changed, 47 insertions(+), 4 deletions(-) diff --git a/rmw_microxrcedds_c/src/rmw_wait.c b/rmw_microxrcedds_c/src/rmw_wait.c index 290125da..8e681d80 100644 --- a/rmw_microxrcedds_c/src/rmw_wait.c +++ b/rmw_microxrcedds_c/src/rmw_wait.c @@ -60,15 +60,44 @@ rmw_wait( rmw_uxrce_clean_expired_static_input_buffer(); + // Clear run flag for all sessions + rmw_uxrce_mempool_item_t * item = session_memory.allocateditems; + while (item != NULL) { + rmw_context_impl_t * custom_context = (rmw_context_impl_t *)item->data; + custom_context->need_to_be_ran = false; + item = item->next; + } + + // TODO(pablogs9): What happens if there already data in one entity? + // Enable flag for every XRCE session available in the entities + for (size_t i = 0; services && i < services->service_count; ++i) { + rmw_uxrce_service_t * custom_service = (rmw_uxrce_service_t *)services->services[i]; + custom_service->owner_node->context->need_to_be_ran = true; + } + + for (size_t i = 0; clients && i < clients->client_count; ++i) { + rmw_uxrce_client_t * custom_client = (rmw_uxrce_client_t *)clients->clients[i]; + custom_client->owner_node->context->need_to_be_ran = true; + } + + for (size_t i = 0; subscriptions && i < subscriptions->subscriber_count; ++i) { + rmw_uxrce_subscription_t * custom_subscription = + (rmw_uxrce_subscription_t *)subscriptions->subscribers[i]; + custom_subscription->owner_node->context->need_to_be_ran = true; + } + // Count sessions to be ran uint8_t available_contexts = 0; - rmw_uxrce_mempool_item_t * item = session_memory.allocateditems; + uint8_t available_sessions = 0; + item = session_memory.allocateditems; while (item != NULL) { - available_contexts++; + rmw_context_impl_t * custom_context = (rmw_context_impl_t *)item->data; + available_contexts += custom_context->need_to_be_ran ? 1 : 0; + available_sessions++; item = item->next; } - // Spin all available contexts to atleast handle session metatraffic + // There is no context that contais any of the wait set entities. Nothing to wait here. if (available_contexts != 0) { int32_t per_session_timeout = (timeout.i32 == UXR_TIMEOUT_INF) ? UXR_TIMEOUT_INF : @@ -77,7 +106,21 @@ rmw_wait( item = session_memory.allocateditems; while (item != NULL) { rmw_context_impl_t * custom_context = (rmw_context_impl_t *)item->data; - uxr_run_session_until_data(&custom_context->session, per_session_timeout); + if (custom_context->need_to_be_ran) { + uxr_run_session_until_data(&custom_context->session, per_session_timeout); + } + item = item->next; + } + } else if (available_sessions != 0) { + int32_t per_session_timeout = + (timeout.i32 == UXR_TIMEOUT_INF) ? UXR_TIMEOUT_INF : + (int32_t)((float)timeout.i32 / (float)available_sessions); + + // Spin with no blocking to handle session metatraffic + item = session_memory.allocateditems; + while (item != NULL) { + rmw_context_impl_t * custom_context = (rmw_context_impl_t *)item->data; + uxr_run_session_timeout(&custom_context->session, per_session_timeout); item = item->next; } } From 34fe6bd147d5ac7a64f9ddf3fdd7c1acf440f0a2 Mon Sep 17 00:00:00 2001 From: David Laseca Perez Date: Thu, 20 Aug 2026 14:08:48 +0200 Subject: [PATCH 4/8] Modify wait approach Signed-off-by: David Laseca Perez --- .../src/rmw_microros_internal/types.h | 4 +- rmw_microxrcedds_c/src/rmw_wait.c | 74 ++++++++++++++----- 2 files changed, 57 insertions(+), 21 deletions(-) diff --git a/rmw_microxrcedds_c/src/rmw_microros_internal/types.h b/rmw_microxrcedds_c/src/rmw_microros_internal/types.h index 45cadb25..b274c116 100644 --- a/rmw_microxrcedds_c/src/rmw_microros_internal/types.h +++ b/rmw_microxrcedds_c/src/rmw_microros_internal/types.h @@ -252,7 +252,9 @@ typedef struct rmw_uxrce_guard_condition_t { rmw_uxrce_mempool_item_t mem; - bool hasTriggered; + // Polled by rmw_wait and can be written from arbitrary threads + // by rmw_trigger_guard_condition, needs 'volatile' + volatile bool hasTriggered; rmw_guard_condition_t rmw_guard_condition; } rmw_uxrce_guard_condition_t; diff --git a/rmw_microxrcedds_c/src/rmw_wait.c b/rmw_microxrcedds_c/src/rmw_wait.c index 8e681d80..6b09daae 100644 --- a/rmw_microxrcedds_c/src/rmw_wait.c +++ b/rmw_microxrcedds_c/src/rmw_wait.c @@ -18,9 +18,25 @@ #include #include #include +#include #include "./rmw_microros_internal/utils.h" +#define RMW_UXRCE_MAX_SESSION_WAIT_SLICE_MS 10 + +static bool rmw_uxrce_any_guard_condition_triggered( + const rmw_guard_conditions_t * guard_conditions) +{ + for (size_t i = 0; guard_conditions && i < guard_conditions->guard_condition_count; ++i) { + rmw_uxrce_guard_condition_t * custom_guard_condition = + (rmw_uxrce_guard_condition_t *)guard_conditions->guard_conditions[i]; + if (custom_guard_condition->hasTriggered) { + return true; + } + } + return false; +} + rmw_ret_t rmw_wait( rmw_subscriptions_t * subscriptions, @@ -97,31 +113,49 @@ rmw_wait( item = item->next; } - // There is no context that contais any of the wait set entities. Nothing to wait here. - if (available_contexts != 0) { - int32_t per_session_timeout = - (timeout.i32 == UXR_TIMEOUT_INF) ? UXR_TIMEOUT_INF : - (int32_t)((float)timeout.i32 / (float)available_contexts); - + // Ensure spinning at least once, even if the timeout is 0 + bool data_available = false; + item = session_memory.allocateditems; + while (item != NULL) { + rmw_context_impl_t * custom_context = (rmw_context_impl_t *)item->data; + if (custom_context->need_to_be_ran) { + data_available |= uxr_run_session_until_data(&custom_context->session, 0); + } else { + uxr_run_session_timeout(&custom_context->session, 0); + } + item = item->next; + } + bool guard_condition_triggered = rmw_uxrce_any_guard_condition_triggered(guard_conditions); + + if (available_sessions != 0 && !data_available && !guard_condition_triggered && + timeout.i32 != 0) + { + // An infinite timeout is handled as a deadline that never expires, so that guard conditions + // are still polled between slices instead of blocking inside a single XRCE receive. + const int64_t deadline = (timeout.i32 == UXR_TIMEOUT_INF) ? + INT64_MAX : uxr_millis() + timeout.i32; + + // Sessions are serviced round-robin in bounded slices, so that no session monopolises the + // wait, guard conditions are polled between slices and the deadline is not overshot. item = session_memory.allocateditems; - while (item != NULL) { + int64_t now = uxr_millis(); + while (!data_available && !guard_condition_triggered && now < deadline) { rmw_context_impl_t * custom_context = (rmw_context_impl_t *)item->data; + const int64_t remaining = deadline - now; + const int32_t slice = (remaining < RMW_UXRCE_MAX_SESSION_WAIT_SLICE_MS) ? + (int32_t)remaining : RMW_UXRCE_MAX_SESSION_WAIT_SLICE_MS; + if (custom_context->need_to_be_ran) { - uxr_run_session_until_data(&custom_context->session, per_session_timeout); + // If at least one context needs to be ran, the other sessions are not serviced + data_available = uxr_run_session_until_data(&custom_context->session, slice); + } else if (available_contexts == 0) { + // No wait set entity belongs to any session, keep every session's metatraffic alive + uxr_run_session_timeout(&custom_context->session, slice); } - item = item->next; - } - } else if (available_sessions != 0) { - int32_t per_session_timeout = - (timeout.i32 == UXR_TIMEOUT_INF) ? UXR_TIMEOUT_INF : - (int32_t)((float)timeout.i32 / (float)available_sessions); - // Spin with no blocking to handle session metatraffic - item = session_memory.allocateditems; - while (item != NULL) { - rmw_context_impl_t * custom_context = (rmw_context_impl_t *)item->data; - uxr_run_session_timeout(&custom_context->session, per_session_timeout); - item = item->next; + guard_condition_triggered = rmw_uxrce_any_guard_condition_triggered(guard_conditions); + item = (item->next == NULL) ? session_memory.allocateditems : item->next; + now = uxr_millis(); } } From 4dfc5fb4a295f7ae5237bd20cf7a62d1907ce794 Mon Sep 17 00:00:00 2001 From: David Laseca Perez Date: Fri, 21 Aug 2026 13:58:40 +0200 Subject: [PATCH 5/8] Add tests for 'rmw_wait' Signed-off-by: David Laseca Perez --- rmw_microxrcedds_c/test/CMakeLists.txt | 3 +- rmw_microxrcedds_c/test/test_wait.cpp | 353 +++++++++++++++++++++++++ 2 files changed, 355 insertions(+), 1 deletion(-) create mode 100644 rmw_microxrcedds_c/test/test_wait.cpp diff --git a/rmw_microxrcedds_c/test/CMakeLists.txt b/rmw_microxrcedds_c/test/CMakeLists.txt index 0be24fc8..d7d08c6a 100644 --- a/rmw_microxrcedds_c/test/CMakeLists.txt +++ b/rmw_microxrcedds_c/test/CMakeLists.txt @@ -57,4 +57,5 @@ rmw_test(test-reqres test_reqres.cpp) rmw_test(test-topic test_topic.cpp) rmw_test(test-rmw test_rmw.cpp) rmw_test(test-sizes test_sizes.cpp) -rmw_test(test-guardcond test_guard_condition.cpp) \ No newline at end of file +rmw_test(test-guardcond test_guard_condition.cpp) +rmw_test(test-wait test_wait.cpp) diff --git a/rmw_microxrcedds_c/test/test_wait.cpp b/rmw_microxrcedds_c/test/test_wait.cpp new file mode 100644 index 00000000..fdeff574 --- /dev/null +++ b/rmw_microxrcedds_c/test/test_wait.cpp @@ -0,0 +1,353 @@ +// Copyright 2026 Proyectos y Sistemas de Mantenimiento SL (eProsima). +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +#include +#include +#include +#include + +#include +#include +#include +#include +#include + +#include "./rmw_base_test.hpp" +#include "./test_utils.hpp" + +using std::chrono::milliseconds; +using std::chrono::steady_clock; + +// Wall clock tolerances. The lower bound is what the test actually asserts; the upper bound is +// deliberately loose because CI runs these next to a real Agent under an unpredictable load. +static const int64_t TOLERANCE_MS = 100; + +class TestWait : public RMWBaseTest +{ +protected: + void SetUp() override + { + RMWBaseTest::SetUp(); + + node = rmw_create_node(&test_context, "wait_node", "/ns"); + ASSERT_NE(node, nullptr); + + guard_condition = rmw_create_guard_condition(&test_context); + ASSERT_NE(guard_condition, nullptr); + + guard_condition_handle = guard_condition->data; + guard_conditions.guard_conditions = &guard_condition_handle; + guard_conditions.guard_condition_count = 1; + } + + void TearDown() override + { + ASSERT_EQ(rmw_destroy_guard_condition(guard_condition), RMW_RET_OK); + ASSERT_EQ(rmw_destroy_node(node), RMW_RET_OK); + RMWBaseTest::TearDown(); + } + + // rmw_wait() empties the wait set arrays it is given, so they are rebuilt before every call. + void ResetGuardConditions() + { + guard_condition_handle = guard_condition->data; + guard_conditions.guard_conditions = &guard_condition_handle; + guard_conditions.guard_condition_count = 1; + } + + rmw_subscription_t * CreateSubscription() + { + ConfigureDummyTypeSupport( + topic_type, topic_type, message_namespace, id_gen++, &dummy_type_support); + + rmw_qos_profile_t dummy_qos_policies; + ConfigureDefaultQOSPolices(&dummy_qos_policies); + + rmw_subscription_options_t options = rmw_get_default_subscription_options(); + + return rmw_create_subscription( + node, &dummy_type_support.type_support, topic_name, &dummy_qos_policies, &options); + } + + // Runs rmw_wait() on a detached thread and gives up after `budget` to avoid hungs + bool WaitWithBudget( + rmw_subscriptions_t * subscriptions, + rmw_guard_conditions_t * gcs, + const rmw_time_t * timeout, + milliseconds budget, + rmw_ret_t * out_ret) + { + auto done = std::make_shared(false); + auto result = std::make_shared(RMW_RET_ERROR); + + std::thread( + [done, result, subscriptions, gcs, timeout]() { + rmw_ret_t rc = rmw_wait(subscriptions, gcs, NULL, NULL, NULL, NULL, timeout); + result->store(rc); + done->store(true); + }).detach(); + + const auto deadline = steady_clock::now() + budget; + while (!done->load() && steady_clock::now() < deadline) { + std::this_thread::sleep_for(milliseconds(1)); + } + + if (!done->load()) { + return false; + } + *out_ret = static_cast(result->load()); + return true; + } + + rmw_node_t * node; + rmw_guard_condition_t * guard_condition; + void * guard_condition_handle; + rmw_guard_conditions_t guard_conditions; + + dummy_type_support_t dummy_type_support; + const char * topic_type = "topic_type"; + const char * topic_name = "wait_topic"; + const char * message_namespace = "package_name"; + size_t id_gen = 0; +}; + +/* + * A wait set holding only guard conditions must block for the requested timeout instead of + * returning immediately. Returning immediately is what makes rclc spin at 100% CPU when an + * application only has timers. + */ +TEST_F(TestWait, guard_condition_only_blocks_for_timeout) +{ + const int64_t timeout_ms = 300; + rmw_time_t timeout = {0LL, timeout_ms * 1000000ULL}; + + ResetGuardConditions(); + const auto start = steady_clock::now(); + rmw_ret_t rc = rmw_wait(NULL, &guard_conditions, NULL, NULL, NULL, NULL, &timeout); + const auto elapsed = + std::chrono::duration_cast(steady_clock::now() - start).count(); + + EXPECT_EQ(rc, RMW_RET_TIMEOUT); + EXPECT_GE(elapsed, timeout_ms - TOLERANCE_MS); +} + +/* + * A wait shorter than one internal wait slice must cost about the timeout, not a whole slice. + * This is what catches a slice that is not clamped to the time remaining. Tested in a loop + * to ensure the aggregated difference between clamped and unclamped slices is significant + * enough to be measured reliably. + */ +TEST_F(TestWait, short_timeouts_are_not_rounded_up_to_a_full_slice) +{ + const int iterations = 20; + const int64_t timeout_ms = 1; + rmw_time_t timeout = {0LL, timeout_ms * 1000000ULL}; + + const auto start = steady_clock::now(); + for (int i = 0; i < iterations; ++i) { + ResetGuardConditions(); + EXPECT_EQ( + rmw_wait(NULL, &guard_conditions, NULL, NULL, NULL, NULL, &timeout), + RMW_RET_TIMEOUT) << "iteration " << i; + } + const auto elapsed = + std::chrono::duration_cast(steady_clock::now() - start).count(); + + // Around 20 ms when every slice is clamped to the time remaining, and around 200 ms when each + // wait costs a full RMW_UXRCE_MAX_SESSION_WAIT_SLICE_MS regardless of how little is left + EXPECT_LE(elapsed, 100) << iterations << " waits of " << timeout_ms << " ms took " + << elapsed << " ms"; +} + +/* + * A guard condition triggered before rmw_wait() is called must be reported without blocking. + */ +TEST_F(TestWait, already_triggered_guard_condition_returns_immediately) +{ + rmw_time_t timeout = {1LL, 0LL}; // 1 second + + ASSERT_EQ(rmw_trigger_guard_condition(guard_condition), RMW_RET_OK); + + ResetGuardConditions(); + const auto start = steady_clock::now(); + rmw_ret_t rc = rmw_wait(NULL, &guard_conditions, NULL, NULL, NULL, NULL, &timeout); + const auto elapsed = + std::chrono::duration_cast(steady_clock::now() - start).count(); + + EXPECT_EQ(rc, RMW_RET_OK); + EXPECT_LT(elapsed, 1000 - TOLERANCE_MS); + EXPECT_NE(guard_conditions.guard_conditions[0], nullptr); +} + +/* + * A guard condition triggered by another thread while rmw_wait() is blocked must wake the wait + * up early. rmw_trigger_guard_condition() cannot interrupt an XRCE receive, so the only way to + * pass this is to poll the flag while waiting. + */ +TEST_F(TestWait, guard_condition_triggered_while_waiting_wakes_up) +{ + const int64_t timeout_ms = 2000; + const int64_t trigger_after_ms = 200; + rmw_time_t timeout = {0LL, timeout_ms * 1000000ULL}; + + std::thread trigger_thread( + [this, trigger_after_ms]() { + std::this_thread::sleep_for(milliseconds(trigger_after_ms)); + EXPECT_EQ(rmw_trigger_guard_condition(this->guard_condition), RMW_RET_OK); + }); + + ResetGuardConditions(); + const auto start = steady_clock::now(); + rmw_ret_t rc = rmw_wait(NULL, &guard_conditions, NULL, NULL, NULL, NULL, &timeout); + const auto elapsed = + std::chrono::duration_cast(steady_clock::now() - start).count(); + + trigger_thread.join(); + + EXPECT_EQ(rc, RMW_RET_OK); + EXPECT_GE(elapsed, trigger_after_ms - TOLERANCE_MS); + EXPECT_LT(elapsed, timeout_ms - TOLERANCE_MS); +} + +/* + * The same as the previous test, but with a subscription in the wait set. + * This is the ordinary rclc executor shape and it takes a different branch + * of rmw_wait(): the one that blocks in uxr_run_session_until_data() for + * every session owning a wait set entity. + */ +TEST_F(TestWait, guard_condition_with_subscription_in_wait_set_wakes_up) +{ + rmw_subscription_t * subscription = CreateSubscription(); + ASSERT_NE(subscription, nullptr); + + const int64_t timeout_ms = 2000; + const int64_t trigger_after_ms = 200; + rmw_time_t timeout = {0LL, timeout_ms * 1000000ULL}; + + void * subscription_handle = subscription->data; + rmw_subscriptions_t subscriptions; + subscriptions.subscribers = &subscription_handle; + subscriptions.subscriber_count = 1; + + std::thread trigger_thread( + [this, trigger_after_ms]() { + std::this_thread::sleep_for(milliseconds(trigger_after_ms)); + EXPECT_EQ(rmw_trigger_guard_condition(this->guard_condition), RMW_RET_OK); + }); + + ResetGuardConditions(); + const auto start = steady_clock::now(); + rmw_ret_t rc = rmw_wait(&subscriptions, &guard_conditions, NULL, NULL, NULL, NULL, &timeout); + const auto elapsed = + std::chrono::duration_cast(steady_clock::now() - start).count(); + + trigger_thread.join(); + + EXPECT_EQ(rc, RMW_RET_OK); + // More tight timeout to prove the guard condition woke the wait + EXPECT_LT(elapsed, trigger_after_ms + 3 * TOLERANCE_MS) << "woke late: " << elapsed << " ms"; + + EXPECT_EQ(rmw_destroy_subscription(node, subscription), RMW_RET_OK); +} + +/* + * A sub-millisecond timeout truncates to 0 ms in rmw_wait(). The call must still terminate + * promptly and must not turn into a spin. + * With RMW_UXRCE_MAX_SESSIONS forced to 2 for the tests, this is the 1 ms / 2 sessions = 0 case. + */ +TEST_F(TestWait, sub_millisecond_timeout_terminates) +{ + rmw_time_t timeout = {0LL, 500000ULL}; // 500 us + + ResetGuardConditions(); + const auto start = steady_clock::now(); + rmw_ret_t rc = rmw_wait(NULL, &guard_conditions, NULL, NULL, NULL, NULL, &timeout); + const auto elapsed = + std::chrono::duration_cast(steady_clock::now() - start).count(); + + EXPECT_EQ(rc, RMW_RET_TIMEOUT); + EXPECT_LT(elapsed, TOLERANCE_MS); +} + +/* + * A zero timeout is a poll: it must return without blocking, whether or not anything is ready. + */ +TEST_F(TestWait, zero_timeout_returns_immediately) +{ + rmw_time_t timeout = {0LL, 0LL}; + + ResetGuardConditions(); + const auto start = steady_clock::now(); + rmw_ret_t rc = rmw_wait(NULL, &guard_conditions, NULL, NULL, NULL, NULL, &timeout); + const auto elapsed = + std::chrono::duration_cast(steady_clock::now() - start).count(); + + EXPECT_EQ(rc, RMW_RET_TIMEOUT); + EXPECT_LT(elapsed, TOLERANCE_MS); +} + +/* + * An infinite timeout must still be woken by a guard condition. Run on a detached thread + * with a hard budget, so that a regression is reported as a failed assertion rather than + * hanging. + */ +TEST_F(TestWait, infinite_timeout_wakes_on_guard_condition) +{ + const int64_t trigger_after_ms = 200; + + std::thread trigger_thread( + [this, trigger_after_ms]() { + std::this_thread::sleep_for(milliseconds(trigger_after_ms)); + EXPECT_EQ(rmw_trigger_guard_condition(this->guard_condition), RMW_RET_OK); + }); + + ResetGuardConditions(); + rmw_time_t timeout = (rmw_time_t)RMW_DURATION_INFINITE; + rmw_ret_t rc = RMW_RET_ERROR; + const bool returned = + WaitWithBudget(NULL, &guard_conditions, &timeout, milliseconds(5000), &rc); + + trigger_thread.join(); + + ASSERT_TRUE(returned) << "rmw_wait() with an infinite timeout was not woken by a guard " + "condition within 5 s"; + EXPECT_EQ(rc, RMW_RET_OK); +} + +/* + * Repeated waits that time out must not leave the guard condition latched, and must keep + * reporting RMW_RET_TIMEOUT. This is a cheap regression net around the flag bookkeeping at the + * end of rmw_wait(), which clears hasTriggered on the way out. + */ +TEST_F(TestWait, repeated_timeouts_are_stable) +{ + rmw_time_t timeout = {0LL, 50000000ULL}; // 50 ms + + for (int i = 0; i < 5; ++i) { + ResetGuardConditions(); + rmw_ret_t rc = rmw_wait(NULL, &guard_conditions, NULL, NULL, NULL, NULL, &timeout); + EXPECT_EQ(rc, RMW_RET_TIMEOUT) << "iteration " << i << " returned " << rc; + EXPECT_EQ(guard_conditions.guard_conditions[0], nullptr) << "iteration " << i; + } + + ASSERT_EQ(rmw_trigger_guard_condition(guard_condition), RMW_RET_OK); + ResetGuardConditions(); + EXPECT_EQ( + rmw_wait(NULL, &guard_conditions, NULL, NULL, NULL, NULL, &timeout), RMW_RET_OK); + + // The trigger must have been consumed by the successful wait above. + ResetGuardConditions(); + EXPECT_EQ( + rmw_wait(NULL, &guard_conditions, NULL, NULL, NULL, NULL, &timeout), RMW_RET_TIMEOUT); +} From 999c6f166c1a802f5b1c0db60f24a230280fafd7 Mon Sep 17 00:00:00 2001 From: David Laseca Perez Date: Thu, 27 Aug 2026 07:32:11 +0200 Subject: [PATCH 6/8] Fix possible segfault Signed-off-by: David Laseca Perez --- rmw_microxrcedds_c/src/rmw_wait.c | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/rmw_microxrcedds_c/src/rmw_wait.c b/rmw_microxrcedds_c/src/rmw_wait.c index 6b09daae..a5b49b01 100644 --- a/rmw_microxrcedds_c/src/rmw_wait.c +++ b/rmw_microxrcedds_c/src/rmw_wait.c @@ -30,7 +30,7 @@ static bool rmw_uxrce_any_guard_condition_triggered( for (size_t i = 0; guard_conditions && i < guard_conditions->guard_condition_count; ++i) { rmw_uxrce_guard_condition_t * custom_guard_condition = (rmw_uxrce_guard_condition_t *)guard_conditions->guard_conditions[i]; - if (custom_guard_condition->hasTriggered) { + if (NULL != custom_guard_condition && custom_guard_condition->hasTriggered) { return true; } } @@ -201,7 +201,7 @@ rmw_wait( for (size_t i = 0; guard_conditions && i < guard_conditions->guard_condition_count; ++i) { rmw_uxrce_guard_condition_t * custom_guard_condition = (rmw_uxrce_guard_condition_t *)guard_conditions->guard_conditions[i]; - if (custom_guard_condition->hasTriggered == false) { + if (NULL == custom_guard_condition || custom_guard_condition->hasTriggered == false) { guard_conditions->guard_conditions[i] = NULL; } else { custom_guard_condition->hasTriggered = false; From 8d859cbd8ea039354d264d602629a4efef75bf84 Mon Sep 17 00:00:00 2001 From: David Laseca Perez Date: Mon, 24 Aug 2026 09:43:49 +0200 Subject: [PATCH 7/8] Add CI workflow step Signed-off-by: David Laseca Perez --- .github/workflows/ubuntu-ci.yml | 22 ++++++++++++++++++++++ 1 file changed, 22 insertions(+) diff --git a/.github/workflows/ubuntu-ci.yml b/.github/workflows/ubuntu-ci.yml index 0b0bf0d6..0abd6a28 100644 --- a/.github/workflows/ubuntu-ci.yml +++ b/.github/workflows/ubuntu-ci.yml @@ -156,6 +156,7 @@ jobs: # The tests establish real XRCE sessions, so an Agent has to be running before colcon test starts # The Agent must not pick up the implementation under test as its own middleware RMW_IMPLEMENTATION=rmw_fastrtps_cpp ros2 run micro_ros_agent micro_ros_agent udp4 --port 8888 -d -v4 & + agent_pid=$! sleep 1 colcon test \ @@ -163,6 +164,27 @@ jobs: --return-code-on-test-failure \ --packages-select $TESTED_PACKAGES + # Kill Agent so the next step can start its own Agent + kill $agent_pid + + - name: Ensure wait set stability (rmw_wait) + # test_wait.cpp asserts on wall clock timings, so it is the part of the suite most + # exposed to flakiness, so it is repeated 100 times in a random order + run: | + . /opt/ros/$ROS_DISTRO/setup.bash + . install/local_setup.bash + + RMW_IMPLEMENTATION=rmw_fastrtps_cpp ros2 run micro_ros_agent micro_ros_agent udp4 --port 8888 -d -v4 & + agent_pid=$! + sleep 1 + + # --gtest_shuffle reseeds on every iteration and prints the seed it used + ./build/rmw_microxrcedds/test/test-wait \ + --gtest_repeat=100 \ + --gtest_shuffle + + kill $agent_pid + - name: Static memory report # Kept separate from the test step so it still runs when a test fails, # but needs the build to succeed From f11fab6c27aab99c8fca8099d124740c9b817f96 Mon Sep 17 00:00:00 2001 From: David Laseca Perez Date: Tue, 8 Sep 2026 16:19:30 +0200 Subject: [PATCH 8/8] Fix CI after new step Signed-off-by: David Laseca Perez --- .github/workflows/ubuntu-ci.yml | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/.github/workflows/ubuntu-ci.yml b/.github/workflows/ubuntu-ci.yml index 0abd6a28..bc3842f1 100644 --- a/.github/workflows/ubuntu-ci.yml +++ b/.github/workflows/ubuntu-ci.yml @@ -191,8 +191,16 @@ jobs: if: ${{ !cancelled() && steps.build.outcome == 'success' }} run: | . /opt/ros/$ROS_DISTRO/setup.bash + . install/local_setup.bash + + RMW_IMPLEMENTATION=rmw_fastrtps_cpp ros2 run micro_ros_agent micro_ros_agent udp4 --port 8888 -d -v4 & + agent_pid=$! + sleep 1 + ./build/rmw_microxrcedds/test/test-sizes 2> memanalisys_out + kill $agent_pid + - name: Test results if: ${{ !cancelled() }} run: |