From 95fdc078c4c07c3b799496b284177e220e09ce9d Mon Sep 17 00:00:00 2001 From: Tamaki Nishino Date: Sun, 18 Oct 2026 13:18:00 +0900 Subject: [PATCH 1/2] Resume a posted waiter only while its awaiter is alive Channel::send/close, TfBuffer's transform arrival and send_goal's goal response post the waiting coroutine's resumption without checking that it is still waiting. A task destroyed after the post but before the executor runs it is then resumed from freed memory. Add Waiting, owned by the awaiter, and post_resume, which resumes it through a weak_ptr when the posted callback runs. A late resumption finds the weak_ptr expired with the awaiter. Add tests for the three cases, which hit a heap-use-after-free on main under ASan, and one that destroys a node-like owner of the context and its task while the task waits on an Event, a sleep or a topic. --- rclcpp_async/CMakeLists.txt | 3 +- rclcpp_async/include/rclcpp_async/channel.hpp | 44 +++++----- .../include/rclcpp_async/co_context.hpp | 11 ++- .../include/rclcpp_async/executor.hpp | 18 +++++ .../include/rclcpp_async/goal_stream.hpp | 2 + .../include/rclcpp_async/tf_buffer.hpp | 14 ++-- rclcpp_async/test/test_channel.cpp | 28 +++++++ rclcpp_async/test/test_task_destroy.cpp | 80 +++++++++++++++++++ rclcpp_async/test/test_tf_buffer.cpp | 24 ++++++ 9 files changed, 187 insertions(+), 37 deletions(-) diff --git a/rclcpp_async/CMakeLists.txt b/rclcpp_async/CMakeLists.txt index 31a71e4..0cd5729 100644 --- a/rclcpp_async/CMakeLists.txt +++ b/rclcpp_async/CMakeLists.txt @@ -149,7 +149,8 @@ if(BUILD_TESTING) target_link_libraries(test_tf_buffer ${PROJECT_NAME}) ament_add_ros_isolated_gtest(test_task_destroy test/test_task_destroy.cpp) - target_link_libraries(test_task_destroy ${PROJECT_NAME} ${example_interfaces_TARGETS}) + target_link_libraries( + test_task_destroy ${PROJECT_NAME} ${example_interfaces_TARGETS} ${std_msgs_TARGETS}) add_subdirectory(test/benchmark) endif() diff --git a/rclcpp_async/include/rclcpp_async/channel.hpp b/rclcpp_async/include/rclcpp_async/channel.hpp index c9e5ac4..06e9198 100644 --- a/rclcpp_async/include/rclcpp_async/channel.hpp +++ b/rclcpp_async/include/rclcpp_async/channel.hpp @@ -36,7 +36,7 @@ class Channel Executor & ctx_; std::mutex mutex_; std::queue queue_; - std::coroutine_handle<> waiter_; + std::weak_ptr waiter_; size_t max_depth_; bool closed_ = false; @@ -55,6 +55,7 @@ class Channel std::stop_token token; std::shared_ptr cancel_cb_; bool cancelled = false; + std::shared_ptr waiting_; void set_token(std::stop_token t) { token = std::move(t); } @@ -86,7 +87,7 @@ class Channel } }; - NextAwaiter next() { return NextAwaiter{*this, {}, {}, false}; } + NextAwaiter next() { return NextAwaiter{*this, {}, {}, false, nullptr}; } }; template @@ -97,26 +98,19 @@ bool Channel::NextAwaiter::await_suspend(std::coroutine_handle<> h) if (!ch.queue_.empty() || ch.closed_) { return false; // don't suspend, data already available } - ch.waiter_ = h; + waiting_ = std::make_shared(h); + ch.waiter_ = waiting_; } // release ch.mutex_ before stop_callback to avoid deadlock - cancel_cb_ = std::make_shared(token, [this, h, &cb = cancel_cb_]() { - std::coroutine_handle<> w; + cancel_cb_ = std::make_shared(token, [this]() { { std::lock_guard lock(ch.mutex_); - if (ch.waiter_ != h) { + if (ch.waiter_.lock() != waiting_) { return; } - w = ch.waiter_; - ch.waiter_ = nullptr; - } - if (w) { - cancelled = true; - ch.ctx_.post([w, weak = std::weak_ptr(cb)]() { - if (weak.lock()) { - w.resume(); - } - }); + ch.waiter_.reset(); } + cancelled = true; + post_resume(ch.ctx_, waiting_); }); return true; } @@ -124,7 +118,7 @@ bool Channel::NextAwaiter::await_suspend(std::coroutine_handle<> h) template void Channel::send(T value) { - std::coroutine_handle<> w; + std::weak_ptr w; { std::lock_guard lock(mutex_); if (closed_) { @@ -134,26 +128,24 @@ void Channel::send(T value) while (queue_.size() > max_depth_) { queue_.pop(); } - w = waiter_; - waiter_ = nullptr; + w = std::exchange(waiter_, {}); } - if (w) { - ctx_.post([w]() { w.resume(); }); + if (!w.expired()) { + post_resume(ctx_, std::move(w)); } } template void Channel::close() { - std::coroutine_handle<> w; + std::weak_ptr w; { std::lock_guard lock(mutex_); closed_ = true; - w = waiter_; - waiter_ = nullptr; + w = std::exchange(waiter_, {}); } - if (w) { - ctx_.post([w]() { w.resume(); }); + if (!w.expired()) { + post_resume(ctx_, std::move(w)); } } diff --git a/rclcpp_async/include/rclcpp_async/co_context.hpp b/rclcpp_async/include/rclcpp_async/co_context.hpp index 2792444..845ef66 100644 --- a/rclcpp_async/include/rclcpp_async/co_context.hpp +++ b/rclcpp_async/include/rclcpp_async/co_context.hpp @@ -259,7 +259,8 @@ class CoContext : public Executor max_depth, {}, {}, - std::make_shared::State>()}; + std::make_shared::State>(), + nullptr}; } template @@ -532,8 +533,10 @@ void SendGoalAwaiter::await_suspend(std::coroutine_handle<> h) // server's accept/reject). When `state->done` is already set by the // cancel path, the callback bails out before touching the (dangling) // coroutine handle. - options.goal_response_callback = [state = state_, client = client, &ctx = ctx, - h](const auto & goal_handle) { + waiting_ = std::make_shared(h); + state_->waiting = waiting_; + options.goal_response_callback = [state = state_, client = client, + &ctx = ctx](const auto & goal_handle) { if (state->done) { return; } @@ -550,7 +553,7 @@ void SendGoalAwaiter::await_suspend(std::coroutine_handle<> h) // (Jazzy bug: https://github.com/ros2/rclcpp/issues/2796). // Resuming synchronously here would deadlock if the coroutine // immediately calls async_send_goal again. - ctx.post([&ctx, h]() { ctx.resume(h); }); + post_resume(ctx, state->waiting); }; client->async_send_goal(goal, options); diff --git a/rclcpp_async/include/rclcpp_async/executor.hpp b/rclcpp_async/include/rclcpp_async/executor.hpp index 24b724a..42344bc 100644 --- a/rclcpp_async/include/rclcpp_async/executor.hpp +++ b/rclcpp_async/include/rclcpp_async/executor.hpp @@ -19,6 +19,7 @@ #include #include #include +#include namespace rclcpp_async { @@ -39,6 +40,23 @@ class Executor virtual void resume(std::coroutine_handle<> h) { h.resume(); } }; +// A suspended coroutine, owned by its awaiter. Whoever resumes it later holds +// a weak_ptr, which expires with the awaiter. +struct Waiting +{ + std::coroutine_handle<> h; +}; + +// Resume `w` on the executor unless its awaiter is gone by then. +inline void post_resume(Executor & ctx, std::weak_ptr w) +{ + ctx.post([w = std::move(w)]() { + if (auto s = w.lock()) { + s->h.resume(); + } + }); +} + template inline void register_cancel( std::shared_ptr & out, std::stop_token token, Executor & ctx, std::coroutine_handle<> h, diff --git a/rclcpp_async/include/rclcpp_async/goal_stream.hpp b/rclcpp_async/include/rclcpp_async/goal_stream.hpp index 3882352..9d6be47 100644 --- a/rclcpp_async/include/rclcpp_async/goal_stream.hpp +++ b/rclcpp_async/include/rclcpp_async/goal_stream.hpp @@ -194,6 +194,7 @@ struct SendGoalAwaiter bool cancelled = false; std::shared_ptr> stream; Result>> result; + std::weak_ptr waiting; }; CoContext & ctx; @@ -203,6 +204,7 @@ struct SendGoalAwaiter std::stop_token token; std::shared_ptr cancel_cb_; std::shared_ptr state_; + std::shared_ptr waiting_; // Same as SendRequestAwaiter: the goal response must not resume a frame // destroyed while suspended here. diff --git a/rclcpp_async/include/rclcpp_async/tf_buffer.hpp b/rclcpp_async/include/rclcpp_async/tf_buffer.hpp index 5c81bc8..6e0e76f 100644 --- a/rclcpp_async/include/rclcpp_async/tf_buffer.hpp +++ b/rclcpp_async/include/rclcpp_async/tf_buffer.hpp @@ -45,7 +45,7 @@ class TfBuffer std::string target_frame; std::string source_frame; tf2::TimePoint time; - std::coroutine_handle<> handle; + std::weak_ptr waiting; std::shared_ptr active; geometry_msgs::msg::TransformStamped result; }; @@ -110,6 +110,7 @@ class TfBuffer std::shared_ptr cancel_cb_; bool cancelled = false; std::shared_ptr request; + std::shared_ptr waiting; // A frame destroyed while suspended here must not stay pending, or the // transform's arrival would resume it. The deferred cancellation no longer @@ -144,7 +145,8 @@ class TfBuffer request->target_frame = target_frame; request->source_frame = source_frame; request->time = time; - request->handle = h; + waiting = std::make_shared(h); + request->waiting = waiting; request->active = std::make_shared(true); tf.add_pending(request); @@ -170,8 +172,9 @@ class TfBuffer LookupTransformAwaiter lookup_transform( const std::string & target_frame, const std::string & source_frame, const rclcpp::Time & time) { - return LookupTransformAwaiter{*this, target_frame, source_frame, tf2_ros::fromRclcpp(time), - {}, {}, false, nullptr}; + return LookupTransformAwaiter{*this, target_frame, source_frame, tf2_ros::fromRclcpp(time), + {}, {}, false, nullptr, + nullptr}; } void add_pending(std::shared_ptr req) @@ -213,8 +216,7 @@ class TfBuffer return false; } *req->active = false; - auto h = req->handle; - ctx_.post([h]() { h.resume(); }); + post_resume(ctx_, req->waiting); return true; } return false; diff --git a/rclcpp_async/test/test_channel.cpp b/rclcpp_async/test/test_channel.cpp index 8676165..fe64644 100644 --- a/rclcpp_async/test/test_channel.cpp +++ b/rclcpp_async/test/test_channel.cpp @@ -263,3 +263,31 @@ TEST_F(ChannelTest, CancelFromThread) ASSERT_TRUE(running.handle.done()); EXPECT_TRUE(was_cancelled); } + +TEST_F(ChannelTest, SendThenWaiterDestroyed) +{ + Channel ch(*ctx_); + auto coro = [&]() -> Task { co_await ch.next(); }; + { + auto task = ctx_->create_task(coro()); + ch.send(1); // posts the resumption, destroyed before it runs + } + for (int i = 0; i < 5; i++) { + executor_.spin_some(); + std::this_thread::sleep_for(1ms); + } +} + +TEST_F(ChannelTest, CloseThenWaiterDestroyed) +{ + Channel ch(*ctx_); + auto coro = [&]() -> Task { co_await ch.next(); }; + { + auto task = ctx_->create_task(coro()); + ch.close(); // posts the resumption, destroyed before it runs + } + for (int i = 0; i < 5; i++) { + executor_.spin_some(); + std::this_thread::sleep_for(1ms); + } +} diff --git a/rclcpp_async/test/test_task_destroy.cpp b/rclcpp_async/test/test_task_destroy.cpp index 211de02..b8da11e 100644 --- a/rclcpp_async/test/test_task_destroy.cpp +++ b/rclcpp_async/test/test_task_destroy.cpp @@ -21,6 +21,7 @@ #include #include #include +#include #include "rclcpp_async/rclcpp_async.hpp" @@ -314,3 +315,82 @@ TEST_F(TaskDestroyTest, ResponseAfterRequesterDestroyed) service->send_response(*pending_header, response); spin_for(200ms); } + +TEST_F(TaskDestroyTest, GoalResponseThenRequesterDestroyed) +{ + // The client answers in its own group, so its goal response can post the + // resumption without executor_ running it. + auto group = node_->create_callback_group(rclcpp::CallbackGroupType::MutuallyExclusive, false); + rclcpp::executors::SingleThreadedExecutor client_executor; + client_executor.add_callback_group(group, node_->get_node_base_interface()); + auto client = rclcpp_action::create_client(node_, "test_destroy_action", group); + bool goal_received = false; + action_server_ = rclcpp_action::create_server( + node_, "test_destroy_action", + [&goal_received](const rclcpp_action::GoalUUID &, std::shared_ptr) { + goal_received = true; + return rclcpp_action::GoalResponse::ACCEPT_AND_EXECUTE; + }, + [](const std::shared_ptr) { return rclcpp_action::CancelResponse::ACCEPT; }, + [](const std::shared_ptr) {}); + ASSERT_TRUE(client->wait_for_action_server(5s)); + + auto send = [this, &client]() -> Task { + co_await ctx_->send_goal(client, Fibonacci::Goal()); + }; + { + auto task = ctx_->create_task(send()); + auto deadline = std::chrono::steady_clock::now() + 5s; + while (!goal_received && std::chrono::steady_clock::now() < deadline) { + spin_for(10ms); + } + ASSERT_TRUE(goal_received); + // Let the goal response post the resumption, then destroy the task + // before executor_ runs it. + for (int i = 0; i < 20; i++) { + client_executor.spin_some(); + std::this_thread::sleep_for(10ms); + } + } + spin_for(50ms); +} + +// A node-like owner holding the context, what its task waits on, and the task. +struct TaskOwner +{ + TaskOwner(rclcpp::Node & node, std::shared_ptr g) + : ctx(node), event(ctx), guard(std::move(g)) + { + stream = ctx.subscribe("test_destroy_owner_topic", 10); + } + CoContext ctx; + Event event; + std::shared_ptr> stream; + std::shared_ptr guard; + Task task; + + template + Task run(F wait) + { + auto g = guard; + co_await wait(*this); + } +}; + +TEST_F(TaskDestroyTest, OwnerDestroyedWhileItsTaskWaits) +{ + auto pub = node_->create_publisher("test_destroy_owner_topic", 10); + auto check = [&](auto wait) { + auto guard = std::make_shared(0); + auto owner = std::make_unique(*node_, guard); + owner->task = owner->ctx.create_task(owner->run(wait)); + spin_for(50ms); + owner.reset(); + pub->publish(std_msgs::msg::String()); + spin_for(50ms); + EXPECT_EQ(guard.use_count(), 1); // the frame went with its owner + }; + check([](TaskOwner & o) { return as_task(o.event.wait()); }); + check([](TaskOwner & o) { return as_task(o.ctx.sleep(10s)); }); + check([](TaskOwner & o) -> Task { co_await o.stream->next(); }); +} diff --git a/rclcpp_async/test/test_tf_buffer.cpp b/rclcpp_async/test/test_tf_buffer.cpp index c711b32..e08f6bf 100644 --- a/rclcpp_async/test/test_tf_buffer.cpp +++ b/rclcpp_async/test/test_tf_buffer.cpp @@ -244,6 +244,30 @@ TEST_F(TfBufferTest, TransformAfterWaiterDestroyed) } } +TEST_F(TfBufferTest, TransformThenWaiterDestroyed) +{ + auto coro = [&]() -> Task { + co_await tf_buffer_->lookup_transform("r", "s", rclcpp::Time(0)); + }; + auto broadcaster = make_static_broadcaster(node_); + { + auto task = ctx_->create_task(coro()); + // The transform arrives on TfBuffer's own thread, which posts the + // resumption; the task is destroyed before executor_ runs it. + broadcaster.sendTransform(make_transform("r", "s")); + auto deadline = std::chrono::steady_clock::now() + 5s; + while (!tf_buffer_->lookup_transform("r", "s") && std::chrono::steady_clock::now() < deadline) { + std::this_thread::sleep_for(10ms); + } + ASSERT_TRUE(tf_buffer_->lookup_transform("r", "s")); + std::this_thread::sleep_for(50ms); + } + for (int i = 0; i < 5; i++) { + executor_.spin_some(); + std::this_thread::sleep_for(10ms); + } +} + TEST_F(TfBufferTest, MultipleConcurrentLookups) { geometry_msgs::msg::TransformStamped r1, r2; From 016ff12a8c42049e530b66fb7008ecf7ef56b6af Mon Sep 17 00:00:00 2001 From: Tamaki Nishino Date: Sun, 18 Oct 2026 13:50:52 +0900 Subject: [PATCH 2/2] Resume a held handle only while its awaiter is alive Two awaiters hand a bare handle to something that outlives them: - GoalStream::cancel_goal() gives `this` and the handle to the action client's cancel callback. Destroying the waiting task before the cancel response arrives makes the callback write to the freed awaiter and resume the freed frame. - co_await on a Task stores the awaiting handle as the task's continuation. When the awaited Task is owned elsewhere and the awaiting task is destroyed first, the awaited task resumes the freed frame when it finishes. Both now go through Waiting, as the posted resumptions do. Task gets an Awaiter that lives in the awaiting frame and owns the Waiting, and the continuation becomes a weak_ptr to it; shield() does the same. Add a test for each, which hit a heap-use-after-free under ASan without this change. --- .../include/rclcpp_async/co_context.hpp | 11 +- .../include/rclcpp_async/goal_stream.hpp | 3 +- rclcpp_async/include/rclcpp_async/shield.hpp | 7 +- rclcpp_async/include/rclcpp_async/task.hpp | 113 +++++++++++------- rclcpp_async/test/test_task_destroy.cpp | 62 ++++++++++ 5 files changed, 144 insertions(+), 52 deletions(-) diff --git a/rclcpp_async/include/rclcpp_async/co_context.hpp b/rclcpp_async/include/rclcpp_async/co_context.hpp index 845ef66..b9f888e 100644 --- a/rclcpp_async/include/rclcpp_async/co_context.hpp +++ b/rclcpp_async/include/rclcpp_async/co_context.hpp @@ -499,10 +499,15 @@ void GoalStream::NextAwaiter::await_suspend(std::coroutine_handle<> h) template void GoalStream::CancelAwaiter::await_suspend(std::coroutine_handle<> h) { + waiting_ = std::make_shared(h); + // `this` is alive exactly while waiting_ is. stream.client_->async_cancel_goal( - stream.goal_handle_, [this, h](typename CancelResponse::SharedPtr resp) { - response = std::move(resp); - stream.ctx_.resume(h); + stream.goal_handle_, + [this, w = std::weak_ptr(waiting_)](typename CancelResponse::SharedPtr resp) { + if (auto s = w.lock()) { + response = std::move(resp); + stream.ctx_.resume(s->h); + } }); } diff --git a/rclcpp_async/include/rclcpp_async/goal_stream.hpp b/rclcpp_async/include/rclcpp_async/goal_stream.hpp index 9d6be47..0545409 100644 --- a/rclcpp_async/include/rclcpp_async/goal_stream.hpp +++ b/rclcpp_async/include/rclcpp_async/goal_stream.hpp @@ -165,6 +165,7 @@ class GoalStream { GoalStream & stream; typename CancelResponse::SharedPtr response; + std::shared_ptr waiting_; bool await_ready() { return false; } @@ -173,7 +174,7 @@ class GoalStream typename CancelResponse::SharedPtr await_resume() { return std::move(response); } }; - CancelAwaiter cancel_goal() { return CancelAwaiter{*this, nullptr}; } + CancelAwaiter cancel_goal() { return CancelAwaiter{*this, nullptr, nullptr}; } void set_auto_cancel_on_stop(bool v) { auto_cancel_on_stop_ = v; } diff --git a/rclcpp_async/include/rclcpp_async/shield.hpp b/rclcpp_async/include/rclcpp_async/shield.hpp index 922336e..84ab4b7 100644 --- a/rclcpp_async/include/rclcpp_async/shield.hpp +++ b/rclcpp_async/include/rclcpp_async/shield.hpp @@ -16,6 +16,7 @@ #include #include +#include #include #include @@ -35,12 +36,14 @@ template struct ShieldAwaiter { Task task_; + std::shared_ptr waiting_; bool await_ready() { return task_.handle.done(); } std::coroutine_handle<> await_suspend(std::coroutine_handle<> cont) { - task_.handle.promise().continuation = cont; + waiting_ = std::make_shared(cont); + task_.handle.promise().continuation = waiting_; if (task_.started_) { return std::noop_coroutine(); } @@ -71,7 +74,7 @@ struct ShieldAwaiter template detail::ShieldAwaiter shield(Task task) { - return {std::move(task)}; + return {std::move(task), nullptr}; } } // namespace rclcpp_async diff --git a/rclcpp_async/include/rclcpp_async/task.hpp b/rclcpp_async/include/rclcpp_async/task.hpp index 6504630..fa07f03 100644 --- a/rclcpp_async/include/rclcpp_async/task.hpp +++ b/rclcpp_async/include/rclcpp_async/task.hpp @@ -24,6 +24,7 @@ #include #include "rclcpp_async/cancelled_exception.hpp" +#include "rclcpp_async/executor.hpp" namespace rclcpp_async { @@ -39,7 +40,7 @@ struct Task { std::optional value; std::exception_ptr exception; - std::coroutine_handle<> continuation; + std::weak_ptr continuation; std::stop_source stop_source; Task get_return_object() @@ -54,8 +55,8 @@ struct Task bool await_ready() noexcept { return false; } std::coroutine_handle<> await_suspend(std::coroutine_handle h) noexcept { - if (h.promise().continuation) { - return h.promise().continuation; + if (auto c = h.promise().continuation.lock()) { + return c->h; } return std::noop_coroutine(); } @@ -91,30 +92,40 @@ struct Task void set_token(std::stop_token t) { parent_token_ = std::move(t); } - // co_await Task (coroutine chaining) - bool await_ready() { return handle.done(); } - std::coroutine_handle<> await_suspend(std::coroutine_handle<> h) + // co_await Task (coroutine chaining). The awaiter lives in the awaiting + // frame, so the continuation expires with it even if this Task outlives it. + struct Awaiter { - handle.promise().continuation = h; - if (parent_token_.stop_possible()) { - auto child = handle; - parent_cancel_cb_ = std::unique_ptr( - new StopCb(parent_token_, [child]() { child.promise().stop_source.request_stop(); })); - } - if (started_) { - return std::noop_coroutine(); + Task & task; + std::shared_ptr waiting_; + + bool await_ready() { return task.handle.done(); } + std::coroutine_handle<> await_suspend(std::coroutine_handle<> h) + { + waiting_ = std::make_shared(h); + task.handle.promise().continuation = waiting_; + if (task.parent_token_.stop_possible()) { + auto child = task.handle; + task.parent_cancel_cb_ = std::unique_ptr(new StopCb( + task.parent_token_, [child]() { child.promise().stop_source.request_stop(); })); + } + if (task.started_) { + return std::noop_coroutine(); + } + task.started_ = true; + return task.handle; } - started_ = true; - return handle; - } - T await_resume() - { - parent_cancel_cb_.reset(); - if (handle.promise().exception) { - std::rethrow_exception(handle.promise().exception); + T await_resume() + { + task.parent_cancel_cb_.reset(); + if (task.handle.promise().exception) { + std::rethrow_exception(task.handle.promise().exception); + } + return std::move(*task.handle.promise().value); } - return std::move(*handle.promise().value); - } + }; + Awaiter operator co_await() & { return Awaiter{*this, nullptr}; } + Awaiter operator co_await() && { return Awaiter{*this, nullptr}; } T & result() { @@ -169,7 +180,7 @@ struct Task struct promise_type { std::exception_ptr exception; - std::coroutine_handle<> continuation; + std::weak_ptr continuation; std::stop_source stop_source; Task get_return_object() @@ -184,8 +195,8 @@ struct Task bool await_ready() noexcept { return false; } std::coroutine_handle<> await_suspend(std::coroutine_handle h) noexcept { - if (h.promise().continuation) { - return h.promise().continuation; + if (auto c = h.promise().continuation.lock()) { + return c->h; } return std::noop_coroutine(); } @@ -221,28 +232,38 @@ struct Task void set_token(std::stop_token t) { parent_token_ = std::move(t); } - bool await_ready() { return handle.done(); } - std::coroutine_handle<> await_suspend(std::coroutine_handle<> h) + // See Task::Awaiter. + struct Awaiter { - handle.promise().continuation = h; - if (parent_token_.stop_possible()) { - auto child = handle; - parent_cancel_cb_ = std::unique_ptr( - new StopCb(parent_token_, [child]() { child.promise().stop_source.request_stop(); })); - } - if (started_) { - return std::noop_coroutine(); + Task & task; + std::shared_ptr waiting_; + + bool await_ready() { return task.handle.done(); } + std::coroutine_handle<> await_suspend(std::coroutine_handle<> h) + { + waiting_ = std::make_shared(h); + task.handle.promise().continuation = waiting_; + if (task.parent_token_.stop_possible()) { + auto child = task.handle; + task.parent_cancel_cb_ = std::unique_ptr(new StopCb( + task.parent_token_, [child]() { child.promise().stop_source.request_stop(); })); + } + if (task.started_) { + return std::noop_coroutine(); + } + task.started_ = true; + return task.handle; } - started_ = true; - return handle; - } - void await_resume() - { - parent_cancel_cb_.reset(); - if (handle.promise().exception) { - std::rethrow_exception(handle.promise().exception); + void await_resume() + { + task.parent_cancel_cb_.reset(); + if (task.handle.promise().exception) { + std::rethrow_exception(task.handle.promise().exception); + } } - } + }; + Awaiter operator co_await() & { return Awaiter{*this, nullptr}; } + Awaiter operator co_await() && { return Awaiter{*this, nullptr}; } void cancel() { handle.promise().stop_source.request_stop(); } diff --git a/rclcpp_async/test/test_task_destroy.cpp b/rclcpp_async/test/test_task_destroy.cpp index b8da11e..479243e 100644 --- a/rclcpp_async/test/test_task_destroy.cpp +++ b/rclcpp_async/test/test_task_destroy.cpp @@ -394,3 +394,65 @@ TEST_F(TaskDestroyTest, OwnerDestroyedWhileItsTaskWaits) check([](TaskOwner & o) { return as_task(o.ctx.sleep(10s)); }); check([](TaskOwner & o) -> Task { co_await o.stream->next(); }); } + +TEST_F(TaskDestroyTest, CancelGoalResponseAfterRequesterDestroyed) +{ + // As in GoalResponseThenRequesterDestroyed, the client answers in its own + // group, so the cancel response arrives only when client_executor spins. + auto group = node_->create_callback_group(rclcpp::CallbackGroupType::MutuallyExclusive, false); + rclcpp::executors::SingleThreadedExecutor client_executor; + client_executor.add_callback_group(group, node_->get_node_base_interface()); + auto client = rclcpp_action::create_client(node_, "test_destroy_action", group); + std::shared_ptr goal_handle; // keeps the goal active + action_server_ = rclcpp_action::create_server( + node_, "test_destroy_action", + [](const rclcpp_action::GoalUUID &, std::shared_ptr) { + return rclcpp_action::GoalResponse::ACCEPT_AND_EXECUTE; + }, + [this](const std::shared_ptr) { + cancel_requested_ = true; + return rclcpp_action::CancelResponse::ACCEPT; + }, + [&goal_handle](const std::shared_ptr gh) { goal_handle = gh; }); + ASSERT_TRUE(client->wait_for_action_server(5s)); + + std::shared_ptr> stream; + auto send = [this, &client, &stream]() -> Task { + auto r = co_await ctx_->send_goal(client, Fibonacci::Goal()); + stream = *r.value; + }; + auto sender = ctx_->create_task(send()); + auto deadline = std::chrono::steady_clock::now() + 5s; + while (!stream && std::chrono::steady_clock::now() < deadline) { + executor_.spin_some(); + client_executor.spin_some(); + std::this_thread::sleep_for(1ms); + } + ASSERT_TRUE(stream); + + auto cancel = [&stream]() -> Task { co_await stream->cancel_goal(); }; + { + auto task = ctx_->create_task(cancel()); + deadline = std::chrono::steady_clock::now() + 5s; + while (!cancel_requested_ && std::chrono::steady_clock::now() < deadline) { + spin_for(10ms); + } + ASSERT_TRUE(cancel_requested_); + } + for (int i = 0; i < 20; i++) { + client_executor.spin_some(); // the cancel response finds the task gone + std::this_thread::sleep_for(10ms); + } + spin_for(50ms); +} + +TEST_F(TaskDestroyTest, AwaitedTaskFinishesAfterAwaiterDestroyed) +{ + // The awaited task is owned outside the awaiting one, so it outlives it. + auto child = ctx_->create_task(as_task(ctx_->sleep(50ms))); + { + auto task = ctx_->create_task([&child]() -> Task { co_await child; }()); + spin_for(10ms); + } + spin_for(100ms); +}