From f17a57f9f9fa353b52ef12b32520476c8369634c Mon Sep 17 00:00:00 2001 From: yukiendo-pfr Date: Fri, 2 Oct 2026 14:49:36 +0900 Subject: [PATCH 1/3] Leave a destroyed awaiter alone when its cancellation runs late --- .../include/rclcpp_async/co_context.hpp | 2 + .../include/rclcpp_async/executor.hpp | 10 +++-- .../include/rclcpp_async/goal_stream.hpp | 10 +++++ .../include/rclcpp_async/topic_stream.hpp | 10 +++++ rclcpp_async/test/test_topic_stream.cpp | 40 +++++++++++++++++++ 5 files changed, 69 insertions(+), 3 deletions(-) diff --git a/rclcpp_async/include/rclcpp_async/co_context.hpp b/rclcpp_async/include/rclcpp_async/co_context.hpp index cffd59b..1382c89 100644 --- a/rclcpp_async/include/rclcpp_async/co_context.hpp +++ b/rclcpp_async/include/rclcpp_async/co_context.hpp @@ -445,6 +445,7 @@ inline void SleepAwaiter::await_suspend(std::coroutine_handle<> h) template void TopicStream::NextAwaiter::await_suspend(std::coroutine_handle<> h) { + waiting_ = h; stream.waiter_ = h; register_cancel( cancel_cb_, token, stream.ctx_, h, [this, h]() { return stream.waiter_ != h; }, @@ -474,6 +475,7 @@ void GoalStream::resume_waiter(std::coroutine_handle<> h) template void GoalStream::NextAwaiter::await_suspend(std::coroutine_handle<> h) { + waiting_ = h; stream.waiter_ = h; register_cancel( cancel_cb_, token, stream.ctx_, h, [this, h]() { return stream.waiter_ != h; }, diff --git a/rclcpp_async/include/rclcpp_async/executor.hpp b/rclcpp_async/include/rclcpp_async/executor.hpp index 5fcbe2b..24b724a 100644 --- a/rclcpp_async/include/rclcpp_async/executor.hpp +++ b/rclcpp_async/include/rclcpp_async/executor.hpp @@ -46,13 +46,17 @@ inline void register_cancel( { out = std::make_shared(token, [&ctx, h, is_done, action, &out]() { ctx.post([&ctx, h, is_done, action, weak = std::weak_ptr(out)]() { + // The awaiter owns `out`, and is_done / action may refer to it. Once it + // is gone (resumed, or its frame destroyed after the stop was + // requested) neither may be called. + if (!weak.lock()) { + return; + } if (is_done()) { return; } action(); - if (weak.lock()) { - ctx.resume(h); - } + ctx.resume(h); }); }); } diff --git a/rclcpp_async/include/rclcpp_async/goal_stream.hpp b/rclcpp_async/include/rclcpp_async/goal_stream.hpp index 1f84534..361c27c 100644 --- a/rclcpp_async/include/rclcpp_async/goal_stream.hpp +++ b/rclcpp_async/include/rclcpp_async/goal_stream.hpp @@ -106,6 +106,16 @@ class GoalStream std::stop_token token; std::shared_ptr cancel_cb_; bool cancelled = false; + std::coroutine_handle<> waiting_{}; + + // A frame destroyed while suspended here must not stay registered as the + // stream's waiter, or the next message would resume it. + ~NextAwaiter() + { + if (waiting_ && stream.waiter_ == waiting_) { + stream.waiter_ = nullptr; + } + } void set_token(std::stop_token t) { token = std::move(t); } diff --git a/rclcpp_async/include/rclcpp_async/topic_stream.hpp b/rclcpp_async/include/rclcpp_async/topic_stream.hpp index 9707ce0..bef0e2b 100644 --- a/rclcpp_async/include/rclcpp_async/topic_stream.hpp +++ b/rclcpp_async/include/rclcpp_async/topic_stream.hpp @@ -59,6 +59,16 @@ class TopicStream std::stop_token token; std::shared_ptr cancel_cb_; bool cancelled = false; + std::coroutine_handle<> waiting_{}; + + // A frame destroyed while suspended here must not stay registered as the + // stream's waiter, or the next message would resume it. + ~NextAwaiter() + { + if (waiting_ && stream.waiter_ == waiting_) { + stream.waiter_ = nullptr; + } + } void set_token(std::stop_token t) { token = std::move(t); } diff --git a/rclcpp_async/test/test_topic_stream.cpp b/rclcpp_async/test/test_topic_stream.cpp index f9d0b32..3d8f22a 100644 --- a/rclcpp_async/test/test_topic_stream.cpp +++ b/rclcpp_async/test/test_topic_stream.cpp @@ -173,3 +173,43 @@ TEST_F(TopicStreamTest, CancelDuringNext) ASSERT_TRUE(running.handle.done()); EXPECT_TRUE(was_cancelled); } + +// The stream outlives the coroutine waiting on it. Destroying the waiting task +// right after cancel() leaves the queued cancellation, and later messages, to +// meet a frame that no longer exists. +Task WaitForNext(std::shared_ptr> stream) { co_await stream->next(); } + +TEST_F(TopicStreamTest, DestroyingACancelledWaiterBeforeItUnwindsIsSafe) +{ + auto stream = ctx_->subscribe("test_topic", 10); + { + auto task = ctx_->create_task(WaitForNext(stream)); + task.cancel(); + } + for (int i = 0; i < 10; ++i) { + executor_.spin_some(); + std::this_thread::sleep_for(1ms); + } + publish("after"); + for (int i = 0; i < 10; ++i) { + executor_.spin_some(); + std::this_thread::sleep_for(1ms); + } + + auto next = ctx_->create_task(WaitForNext(stream)); + spin_until_done(next); + EXPECT_TRUE(next.handle.done()); +} + +TEST_F(TopicStreamTest, DestroyingASuspendedWaiterDetachesIt) +{ + auto stream = ctx_->subscribe("test_topic", 10); + { + auto task = ctx_->create_task(WaitForNext(stream)); + } + publish("after"); + + auto next = ctx_->create_task(WaitForNext(stream)); + spin_until_done(next); + EXPECT_TRUE(next.handle.done()); +} From 5fb74570052be9583d3df1b3690a3a50e2d57bc5 Mon Sep 17 00:00:00 2001 From: yukiendo-pfr Date: Fri, 2 Oct 2026 15:37:23 +0900 Subject: [PATCH 2/3] Cancel the goal of a stream whose waiting task is destroyed --- .../include/rclcpp_async/co_context.hpp | 6 +-- .../include/rclcpp_async/goal_stream.hpp | 13 +++++- rclcpp_async/test/test_task_destroy.cpp | 40 ++++++++++++++++++- 3 files changed, 52 insertions(+), 7 deletions(-) diff --git a/rclcpp_async/include/rclcpp_async/co_context.hpp b/rclcpp_async/include/rclcpp_async/co_context.hpp index 1382c89..6ca2d46 100644 --- a/rclcpp_async/include/rclcpp_async/co_context.hpp +++ b/rclcpp_async/include/rclcpp_async/co_context.hpp @@ -482,11 +482,7 @@ void GoalStream::NextAwaiter::await_suspend(std::coroutine_handle<> h) [this]() { stream.waiter_ = nullptr; cancelled = true; - if ( - stream.auto_cancel_on_stop_ && !stream.completed_ && stream.goal_handle_ && - stream.client_) { - stream.client_->async_cancel_goal(stream.goal_handle_); - } + stream.cancel_goal_on_stop(); }); } diff --git a/rclcpp_async/include/rclcpp_async/goal_stream.hpp b/rclcpp_async/include/rclcpp_async/goal_stream.hpp index 361c27c..ffa8e58 100644 --- a/rclcpp_async/include/rclcpp_async/goal_stream.hpp +++ b/rclcpp_async/include/rclcpp_async/goal_stream.hpp @@ -92,6 +92,14 @@ class GoalStream void resume_waiter(std::coroutine_handle<> h); + // The waiting coroutine was stopped: cancel the goal it was following. + void cancel_goal_on_stop() + { + if (auto_cancel_on_stop_ && !completed_ && goal_handle_ && client_) { + client_->async_cancel_goal(goal_handle_); + } + } + public: explicit GoalStream(CoContext & ctx, size_t max_depth = kDefaultStreamDepth) : ctx_(ctx), max_depth_(max_depth) @@ -109,11 +117,14 @@ class GoalStream std::coroutine_handle<> waiting_{}; // A frame destroyed while suspended here must not stay registered as the - // stream's waiter, or the next message would resume it. + // stream's waiter, or the next message would resume it. Destroying the + // task stops it first, and the deferred cancellation no longer runs for a + // destroyed awaiter, so the goal is cancelled here instead. ~NextAwaiter() { if (waiting_ && stream.waiter_ == waiting_) { stream.waiter_ = nullptr; + stream.cancel_goal_on_stop(); } } diff --git a/rclcpp_async/test/test_task_destroy.cpp b/rclcpp_async/test/test_task_destroy.cpp index 46e5dd3..f0f6bd1 100644 --- a/rclcpp_async/test/test_task_destroy.cpp +++ b/rclcpp_async/test/test_task_destroy.cpp @@ -14,6 +14,7 @@ #include +#include #include #include #include @@ -65,7 +66,10 @@ class TaskDestroyTest : public ::testing::Test [](const rclcpp_action::GoalUUID &, std::shared_ptr) { return rclcpp_action::GoalResponse::ACCEPT_AND_EXECUTE; }, - [](const std::shared_ptr) { return rclcpp_action::CancelResponse::ACCEPT; }, + [this](const std::shared_ptr) { + cancel_requested_ = true; + return rclcpp_action::CancelResponse::ACCEPT; + }, [count, interval_ms](const std::shared_ptr goal_handle) { std::thread([goal_handle, count, interval_ms]() { std::this_thread::sleep_for(200ms); @@ -107,6 +111,7 @@ class TaskDestroyTest : public ::testing::Test rclcpp::executors::SingleThreadedExecutor executor_; rclcpp_action::Client::SharedPtr action_client_; rclcpp_action::Server::SharedPtr action_server_; + std::atomic cancel_requested_{false}; }; TEST_F(TaskDestroyTest, DestroyWhileSuspendedOnTimer) @@ -191,3 +196,36 @@ TEST_F(TaskDestroyTest, DestroyGoalTaskWhileAwaitingFeedback) spin_until_done(task, 15s); EXPECT_TRUE(outer_done); } + +TEST_F(TaskDestroyTest, DestroyingATaskAwaitingFeedbackCancelsTheGoal) +{ + create_feedback_server(30, 100); + wait_for_server(); + + auto follow = [this]() -> Task { + Fibonacci::Goal goal; + goal.order = 30; + auto goal_result = co_await ctx_->send_goal(action_client_, goal); + if (!goal_result.ok()) { + co_return; + } + auto stream = *goal_result.value; + while (co_await stream->next()) { + } + }; + { + auto task = ctx_->create_task(follow()); + auto deadline = std::chrono::steady_clock::now() + 2s; + while (std::chrono::steady_clock::now() < deadline) { + executor_.spin_some(); + std::this_thread::sleep_for(1ms); + } + } + + auto deadline = std::chrono::steady_clock::now() + 2s; + while (!cancel_requested_ && std::chrono::steady_clock::now() < deadline) { + executor_.spin_some(); + std::this_thread::sleep_for(1ms); + } + EXPECT_TRUE(cancel_requested_); +} From 6565ecb83869bd2ac9438de0cc670d9466e5903f Mon Sep 17 00:00:00 2001 From: yukiendo-pfr Date: Tue, 6 Oct 2026 09:23:02 +0900 Subject: [PATCH 3/3] Withdraw a waiter in its awaiter's destructor when its frame is destroyed --- rclcpp_async/include/rclcpp_async/event.hpp | 10 +++ .../include/rclcpp_async/goal_stream.hpp | 9 ++ rclcpp_async/include/rclcpp_async/mutex.hpp | 10 +++ .../rclcpp_async/send_request_awaiter.hpp | 11 +++ .../include/rclcpp_async/tf_buffer.hpp | 11 +++ rclcpp_async/test/test_task_destroy.cpp | 85 +++++++++++++++++++ rclcpp_async/test/test_tf_buffer.cpp | 25 ++++++ 7 files changed, 161 insertions(+) diff --git a/rclcpp_async/include/rclcpp_async/event.hpp b/rclcpp_async/include/rclcpp_async/event.hpp index 11f1135..feff88d 100644 --- a/rclcpp_async/include/rclcpp_async/event.hpp +++ b/rclcpp_async/include/rclcpp_async/event.hpp @@ -56,6 +56,16 @@ class Event std::shared_ptr cancel_cb_; std::shared_ptr active; + // A frame destroyed while suspended here must not stay registered, or a + // later set() would resume it. The deferred cancellation no longer runs + // for a destroyed awaiter, so the waiter is withdrawn here. + ~WaitAwaiter() + { + if (active) { + *active = false; + } + } + void set_token(std::stop_token t) { token = std::move(t); } bool await_ready() diff --git a/rclcpp_async/include/rclcpp_async/goal_stream.hpp b/rclcpp_async/include/rclcpp_async/goal_stream.hpp index ffa8e58..3882352 100644 --- a/rclcpp_async/include/rclcpp_async/goal_stream.hpp +++ b/rclcpp_async/include/rclcpp_async/goal_stream.hpp @@ -204,6 +204,15 @@ struct SendGoalAwaiter std::shared_ptr cancel_cb_; std::shared_ptr state_; + // Same as SendRequestAwaiter: the goal response must not resume a frame + // destroyed while suspended here. + ~SendGoalAwaiter() + { + if (cancel_cb_) { + state_->done = true; + } + } + void set_token(std::stop_token t) { token = std::move(t); } bool await_ready() diff --git a/rclcpp_async/include/rclcpp_async/mutex.hpp b/rclcpp_async/include/rclcpp_async/mutex.hpp index d3e8779..07fcbd4 100644 --- a/rclcpp_async/include/rclcpp_async/mutex.hpp +++ b/rclcpp_async/include/rclcpp_async/mutex.hpp @@ -52,6 +52,16 @@ class Mutex std::shared_ptr cancel_cb_; std::shared_ptr active; + // A frame destroyed while suspended here must not stay registered, or a + // later unlock() would resume it. The deferred cancellation no longer runs + // for a destroyed awaiter, so the waiter is withdrawn here. + ~LockAwaiter() + { + if (active) { + *active = false; + } + } + void set_token(std::stop_token t) { token = std::move(t); } bool await_ready() diff --git a/rclcpp_async/include/rclcpp_async/send_request_awaiter.hpp b/rclcpp_async/include/rclcpp_async/send_request_awaiter.hpp index 8703aa8..1935730 100644 --- a/rclcpp_async/include/rclcpp_async/send_request_awaiter.hpp +++ b/rclcpp_async/include/rclcpp_async/send_request_awaiter.hpp @@ -53,6 +53,17 @@ struct SendRequestAwaiter std::shared_ptr cancel_cb_; std::shared_ptr state_; + // A frame destroyed while suspended here must not be resumed by the + // response. The deferred cancellation no longer runs for a destroyed + // awaiter, so the request is marked done here. cancel_cb_ is set only while + // suspended, which leaves copies made before the co_await alone. + ~SendRequestAwaiter() + { + if (cancel_cb_) { + state_->done = true; + } + } + void set_token(std::stop_token t) { token = std::move(t); } bool await_ready() diff --git a/rclcpp_async/include/rclcpp_async/tf_buffer.hpp b/rclcpp_async/include/rclcpp_async/tf_buffer.hpp index 4f4b491..5c81bc8 100644 --- a/rclcpp_async/include/rclcpp_async/tf_buffer.hpp +++ b/rclcpp_async/include/rclcpp_async/tf_buffer.hpp @@ -111,6 +111,17 @@ class TfBuffer bool cancelled = false; std::shared_ptr request; + // A frame destroyed while suspended here must not stay pending, or the + // transform's arrival would resume it. The deferred cancellation no longer + // runs for a destroyed awaiter, so the request is withdrawn here. + ~LookupTransformAwaiter() + { + if (request && request->active && *request->active) { + tf.remove_pending(request); + *request->active = false; + } + } + void set_token(std::stop_token t) { token = std::move(t); } bool await_ready() diff --git a/rclcpp_async/test/test_task_destroy.cpp b/rclcpp_async/test/test_task_destroy.cpp index f0f6bd1..211de02 100644 --- a/rclcpp_async/test/test_task_destroy.cpp +++ b/rclcpp_async/test/test_task_destroy.cpp @@ -17,6 +17,7 @@ #include #include #include +#include #include #include #include @@ -27,6 +28,7 @@ using namespace rclcpp_async; // NOLINT(build/namespaces) using namespace std::chrono_literals; // NOLINT(build/namespaces) using Fibonacci = example_interfaces::action::Fibonacci; using GoalHandle = rclcpp_action::ServerGoalHandle; +using AddTwoInts = example_interfaces::srv::AddTwoInts; class TaskDestroyTest : public ::testing::Test { @@ -59,6 +61,15 @@ class TaskDestroyTest : public ::testing::Test } } + void spin_for(std::chrono::milliseconds duration) + { + auto deadline = std::chrono::steady_clock::now() + duration; + while (std::chrono::steady_clock::now() < deadline) { + executor_.spin_some(); + std::this_thread::sleep_for(1ms); + } + } + void create_feedback_server(int count = 30, int interval_ms = 100) { action_server_ = rclcpp_action::create_server( @@ -229,3 +240,77 @@ TEST_F(TaskDestroyTest, DestroyingATaskAwaitingFeedbackCancelsTheGoal) } EXPECT_TRUE(cancel_requested_); } + +static Task wait_event(Event & event) { co_await event.wait(); } +static Task lock_mutex(Mutex & mutex) { co_await mutex.lock(); } + +TEST_F(TaskDestroyTest, EventSetAfterWaiterDestroyed) +{ + Event event(*ctx_); + { + auto task = ctx_->create_task(wait_event(event)); + spin_for(50ms); + } + spin_for(50ms); // let the deferred cancellation run + event.set(); + spin_for(50ms); +} + +TEST_F(TaskDestroyTest, EventSetBeforeDeferredCancellationRuns) +{ + Event event(*ctx_); + { + auto task = ctx_->create_task(wait_event(event)); + spin_for(50ms); + } + event.set(); // the deferred cancellation has not run yet + spin_for(50ms); +} + +TEST_F(TaskDestroyTest, MutexUnlockAfterWaiterDestroyed) +{ + Mutex mutex(*ctx_); + auto holder = ctx_->create_task(lock_mutex(mutex)); + spin_for(50ms); + ASSERT_TRUE(mutex.is_locked()); + { + auto task = ctx_->create_task(lock_mutex(mutex)); + spin_for(50ms); + } + spin_for(50ms); + mutex.unlock(); + spin_for(50ms); + EXPECT_FALSE(mutex.is_locked()); +} + +TEST_F(TaskDestroyTest, ResponseAfterRequesterDestroyed) +{ + std::shared_ptr pending_header; + auto service = node_->create_service( + "test_destroy_service", + [&pending_header]( + std::shared_ptr>, std::shared_ptr header, + std::shared_ptr) { pending_header = header; }); + auto client = node_->create_client("test_destroy_service"); + auto deadline = std::chrono::steady_clock::now() + 5s; + while (!client->service_is_ready() && std::chrono::steady_clock::now() < deadline) { + spin_for(10ms); + } + ASSERT_TRUE(client->service_is_ready()); + + auto request = [this, &client]() -> Task { + co_await ctx_->send_request(client, std::make_shared()); + }; + { + auto task = ctx_->create_task(request()); + deadline = std::chrono::steady_clock::now() + 5s; + while (!pending_header && std::chrono::steady_clock::now() < deadline) { + spin_for(10ms); + } + ASSERT_TRUE(pending_header); + } + spin_for(50ms); // let the deferred cancellation run + AddTwoInts::Response response; + service->send_response(*pending_header, response); + spin_for(200ms); +} diff --git a/rclcpp_async/test/test_tf_buffer.cpp b/rclcpp_async/test/test_tf_buffer.cpp index a306dba..c711b32 100644 --- a/rclcpp_async/test/test_tf_buffer.cpp +++ b/rclcpp_async/test/test_tf_buffer.cpp @@ -219,6 +219,31 @@ TEST_F(TfBufferTest, CancellationRemovesPendingRequest) EXPECT_TRUE(was_cancelled); } +TEST_F(TfBufferTest, TransformAfterWaiterDestroyed) +{ + auto coro = [&]() -> Task { + co_await tf_buffer_->lookup_transform("p", "q", rclcpp::Time(0)); + }; + { + auto task = ctx_->create_task(coro()); + for (int i = 0; i < 5; i++) { + executor_.spin_some(); + std::this_thread::sleep_for(10ms); + } + } + // Let the deferred cancellation run, then deliver the awaited transform. + for (int i = 0; i < 5; i++) { + executor_.spin_some(); + std::this_thread::sleep_for(10ms); + } + auto broadcaster = make_static_broadcaster(node_); + broadcaster.sendTransform(make_transform("p", "q")); + for (int i = 0; i < 30; i++) { + executor_.spin_some(); + std::this_thread::sleep_for(10ms); + } +} + TEST_F(TfBufferTest, MultipleConcurrentLookups) { geometry_msgs::msg::TransformStamped r1, r2;