From 2c8a2337eff704cfab1ec9fa3126647d4cf1e714 Mon Sep 17 00:00:00 2001 From: Tamaki Nishino Date: Tue, 6 Oct 2026 22:51:26 +0900 Subject: [PATCH] Detach a running task instead of destroying its frame ~Task() destroyed the coroutine frame even while it was suspended in an awaiter. Everything that awaiter had handed the handle to -- a stream's waiter, an Event queue, a pending service response, a posted resumption, a TF pending request -- then pointed at freed memory, and each awaiter grew its own guard against that (is_done predicates, weak StopCb checks, the destructors from #56). A running frame is now never destroyed by its owner. ~Task() cancels it and lets go: the awaiter resumes it on the executor, CancelledException unwinds the body, and FinalAwaiter frees the frame. With a live frame behind every handed out handle, the per-awaiter destructors are no longer needed and are removed. create_task now posts the start instead of resuming on the spot, so a task starts after whatever was scheduled before it -- the unwinding of a task cancelled just before, say -- and a task released before its start never runs. This relies on the callable passed to create_task being kept alive (previous commit). Add tests for the ownership rule, the start order, and the remaining cases where a resumption was posted before the task was destroyed (Channel, TfBuffer, send_goal), which crashed on main. --- docs/guide.md | 20 ++- .../include/rclcpp_async/co_context.hpp | 19 ++- rclcpp_async/include/rclcpp_async/event.hpp | 10 -- .../include/rclcpp_async/executor.hpp | 5 +- .../include/rclcpp_async/goal_stream.hpp | 22 ---- rclcpp_async/include/rclcpp_async/mutex.hpp | 10 -- .../rclcpp_async/send_request_awaiter.hpp | 11 -- rclcpp_async/include/rclcpp_async/task.hpp | 76 +++++++---- .../include/rclcpp_async/tf_buffer.hpp | 11 -- .../include/rclcpp_async/topic_stream.hpp | 10 -- rclcpp_async/test/test_channel.cpp | 28 ++++ rclcpp_async/test/test_task_destroy.cpp | 123 ++++++++++++++++++ rclcpp_async/test/test_tf_buffer.cpp | 24 ++++ 13 files changed, 263 insertions(+), 106 deletions(-) diff --git a/docs/guide.md b/docs/guide.md index 90fe748..1e16144 100644 --- a/docs/guide.md +++ b/docs/guide.md @@ -4,7 +4,14 @@ ### create_task -`create_task` registers and starts a coroutine. It accepts either a `Task` directly or a callable that returns one. +`create_task` schedules a coroutine on the context. It accepts either a `Task` directly or a callable that returns one. The coroutine starts when the executor next runs, not inside `create_task` itself, so tasks start in creation order, after anything scheduled before them. + +Note that `task = ctx.create_task(...)` evaluates the right-hand side first: the new task is scheduled before the old one is released. If the old task's unwinding must come first, cancel it first: + +```cpp +task.cancel(); +task = ctx.create_task(next(ctx)); // next() starts after task has unwound +``` ```cpp // From a coroutine function @@ -459,6 +466,17 @@ Task run(CoContext & ctx) } ``` +### Destroying a running task + +Destroying a `Task` (or move-assigning over it) while it is still running cancels it, exactly like `task.cancel()`. The coroutine is not torn down on the spot: it keeps living until the executor resumes it, `CancelledException` unwinds its body, and the frame frees itself. This is what makes it safe to drop a task while it waits on a topic, a service response or an event -- whoever delivers the result still finds a live coroutine. + +Two consequences: + +- Code that runs during unwinding (destructors of locals, `catch (const CancelledException &)` handlers) runs *after* the owner of the `Task` has moved on. Do not touch anything by reference there that the owner may have destroyed. +- A task suspended on an awaitable that does not support cancellation (for example a `shield`ed task) lives until that awaitable completes. + +A task destroyed before the executor has started it never runs. A task that was never scheduled is simply freed. + ### shield `shield(task)` prevents a parent's cancellation from propagating to the inner task. The shielded task runs to completion even if the parent is cancelled. After the shielded task finishes, `CancelledException` is thrown at the next unshielded `co_await`. diff --git a/rclcpp_async/include/rclcpp_async/co_context.hpp b/rclcpp_async/include/rclcpp_async/co_context.hpp index 2792444..eaf49f3 100644 --- a/rclcpp_async/include/rclcpp_async/co_context.hpp +++ b/rclcpp_async/include/rclcpp_async/co_context.hpp @@ -139,9 +139,9 @@ class CoContext : public Executor // MutuallyExclusive: the waiters the context keeps are not thread-safe. // // Clients passed to send_request / send_goal resume in their own group, and - // Event::set, Mutex::unlock, TopicStream::close and create_task resume on - // the calling thread, so call or create those from the same group to keep - // a coroutine serialized with the rest of it. + // Event::set, Mutex::unlock and TopicStream::close resume on the calling + // thread, so call those from the same group to keep a coroutine serialized + // with the rest of it. explicit CoContext(rclcpp::Node & node, rclcpp::CallbackGroup::SharedPtr callback_group = nullptr) : node_(node), callback_group_(std::move(callback_group)) { @@ -166,11 +166,20 @@ class CoContext : public Executor rclcpp::CallbackGroup::SharedPtr callback_group() const { return callback_group_; } + // The task starts on the executor, not here. Whatever was posted before it + // -- the cancellation of the task it replaces, say -- runs first, and a + // task released before its start (see Task::release) never runs. template [[nodiscard]] Task create_task(Task task) { task.started_ = true; - task.handle.resume(); + post([h = task.handle]() { + if (h.promise().detached) { + h.destroy(); + return; + } + h.resume(); + }); return task; } @@ -454,7 +463,6 @@ 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; }, @@ -484,7 +492,6 @@ 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/event.hpp b/rclcpp_async/include/rclcpp_async/event.hpp index feff88d..11f1135 100644 --- a/rclcpp_async/include/rclcpp_async/event.hpp +++ b/rclcpp_async/include/rclcpp_async/event.hpp @@ -56,16 +56,6 @@ 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/executor.hpp b/rclcpp_async/include/rclcpp_async/executor.hpp index 24b724a..2afacdf 100644 --- a/rclcpp_async/include/rclcpp_async/executor.hpp +++ b/rclcpp_async/include/rclcpp_async/executor.hpp @@ -46,9 +46,8 @@ 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. + // The awaiter owns `out`, and is_done / action may refer to it. It is + // gone when the operation completed before this ran. if (!weak.lock()) { return; } diff --git a/rclcpp_async/include/rclcpp_async/goal_stream.hpp b/rclcpp_async/include/rclcpp_async/goal_stream.hpp index 3882352..02ce3be 100644 --- a/rclcpp_async/include/rclcpp_async/goal_stream.hpp +++ b/rclcpp_async/include/rclcpp_async/goal_stream.hpp @@ -114,19 +114,6 @@ 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. 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(); - } - } void set_token(std::stop_token t) { token = std::move(t); } @@ -204,15 +191,6 @@ 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 07fcbd4..d3e8779 100644 --- a/rclcpp_async/include/rclcpp_async/mutex.hpp +++ b/rclcpp_async/include/rclcpp_async/mutex.hpp @@ -52,16 +52,6 @@ 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 1935730..8703aa8 100644 --- a/rclcpp_async/include/rclcpp_async/send_request_awaiter.hpp +++ b/rclcpp_async/include/rclcpp_async/send_request_awaiter.hpp @@ -53,17 +53,6 @@ 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/task.hpp b/rclcpp_async/include/rclcpp_async/task.hpp index 6504630..5a40f10 100644 --- a/rclcpp_async/include/rclcpp_async/task.hpp +++ b/rclcpp_async/include/rclcpp_async/task.hpp @@ -41,6 +41,9 @@ struct Task std::exception_ptr exception; std::coroutine_handle<> continuation; std::stop_source stop_source; + // Set when the Task that owned this frame was destroyed while the frame + // was still running: the frame frees itself when it completes. + bool detached = false; Task get_return_object() { @@ -54,6 +57,10 @@ struct Task bool await_ready() noexcept { return false; } std::coroutine_handle<> await_suspend(std::coroutine_handle h) noexcept { + if (h.promise().detached) { + h.destroy(); + return std::noop_coroutine(); + } if (h.promise().continuation) { return h.promise().continuation; } @@ -129,13 +136,7 @@ struct Task explicit operator bool() const noexcept { return handle != nullptr; } bool done() const noexcept { return handle && handle.done(); } - ~Task() - { - if (handle) { - handle.promise().stop_source.request_stop(); - handle.destroy(); - } - } + ~Task() { release(); } Task(Task && o) noexcept : handle(o.handle), started_(o.started_), parent_cancel_cb_(std::move(o.parent_cancel_cb_)) { @@ -144,10 +145,7 @@ struct Task Task & operator=(Task && o) noexcept { if (this != &o) { - if (handle) { - handle.promise().stop_source.request_stop(); - handle.destroy(); - } + release(); handle = o.handle; started_ = o.started_; parent_cancel_cb_ = std::move(o.parent_cancel_cb_); @@ -160,6 +158,24 @@ struct Task private: explicit Task(std::coroutine_handle h) : handle(h) {} + + // A running frame is suspended inside an awaiter that has handed its handle + // out (to a stream, a queue, a pending response, a posted resumption...), so + // it cannot be destroyed here. Cancel it and let it go: the awaiter resumes + // it, CancelledException unwinds the body, and FinalAwaiter frees the frame. + void release() + { + if (!handle) { + return; + } + bool running = started_ && !handle.done(); + handle.promise().detached = running; + handle.promise().stop_source.request_stop(); + if (!running) { + handle.destroy(); + } + handle = nullptr; + } }; // void specialization @@ -171,6 +187,9 @@ struct Task std::exception_ptr exception; std::coroutine_handle<> continuation; std::stop_source stop_source; + // Set when the Task that owned this frame was destroyed while the frame + // was still running: the frame frees itself when it completes. + bool detached = false; Task get_return_object() { @@ -184,6 +203,10 @@ struct Task bool await_ready() noexcept { return false; } std::coroutine_handle<> await_suspend(std::coroutine_handle h) noexcept { + if (h.promise().detached) { + h.destroy(); + return std::noop_coroutine(); + } if (h.promise().continuation) { return h.promise().continuation; } @@ -249,13 +272,7 @@ struct Task explicit operator bool() const noexcept { return handle != nullptr; } bool done() const noexcept { return handle && handle.done(); } - ~Task() - { - if (handle) { - handle.promise().stop_source.request_stop(); - handle.destroy(); - } - } + ~Task() { release(); } Task(Task && o) noexcept : handle(o.handle), started_(o.started_), parent_cancel_cb_(std::move(o.parent_cancel_cb_)) { @@ -264,10 +281,7 @@ struct Task Task & operator=(Task && o) noexcept { if (this != &o) { - if (handle) { - handle.promise().stop_source.request_stop(); - handle.destroy(); - } + release(); handle = o.handle; started_ = o.started_; parent_cancel_cb_ = std::move(o.parent_cancel_cb_); @@ -280,6 +294,24 @@ struct Task private: explicit Task(std::coroutine_handle h) : handle(h) {} + + // A running frame is suspended inside an awaiter that has handed its handle + // out (to a stream, a queue, a pending response, a posted resumption...), so + // it cannot be destroyed here. Cancel it and let it go: the awaiter resumes + // it, CancelledException unwinds the body, and FinalAwaiter frees the frame. + void release() + { + if (!handle) { + return; + } + bool running = started_ && !handle.done(); + handle.promise().detached = running; + handle.promise().stop_source.request_stop(); + if (!running) { + handle.destroy(); + } + handle = nullptr; + } }; // Fire-and-forget coroutine that starts immediately and self-destructs on diff --git a/rclcpp_async/include/rclcpp_async/tf_buffer.hpp b/rclcpp_async/include/rclcpp_async/tf_buffer.hpp index 5c81bc8..4f4b491 100644 --- a/rclcpp_async/include/rclcpp_async/tf_buffer.hpp +++ b/rclcpp_async/include/rclcpp_async/tf_buffer.hpp @@ -111,17 +111,6 @@ 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/include/rclcpp_async/topic_stream.hpp b/rclcpp_async/include/rclcpp_async/topic_stream.hpp index bef0e2b..9707ce0 100644 --- a/rclcpp_async/include/rclcpp_async/topic_stream.hpp +++ b/rclcpp_async/include/rclcpp_async/topic_stream.hpp @@ -59,16 +59,6 @@ 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_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..b916bad 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,125 @@ TEST_F(TaskDestroyTest, ResponseAfterRequesterDestroyed) service->send_response(*pending_header, response); spin_for(200ms); } + +// Destroying a running task does not destroy its frame: the task is cancelled +// and unwinds on the executor, so whatever holds its handle may still resume it. + +TEST_F(TaskDestroyTest, DestroyedTaskUnwindsOnTheExecutor) +{ + auto guard = std::make_shared(0); + auto hold = [this](std::shared_ptr g) -> Task { + co_await ctx_->sleep(10s); + (void)g; + }; + { + auto task = ctx_->create_task(hold(guard)); + spin_for(50ms); + EXPECT_EQ(guard.use_count(), 2); + } + EXPECT_EQ(guard.use_count(), 2); // still suspended, not destroyed + spin_for(50ms); + EXPECT_EQ(guard.use_count(), 1); // unwound and freed +} + +TEST_F(TaskDestroyTest, DestroyedTaskWithoutCancellableAwaiterOutlivesItsOwner) +{ + // Nothing cancels this awaiter, so the frame lives until set() resumes it. + Event event(*ctx_); + auto guard = std::make_shared(0); + auto hold = [](Event & e, std::shared_ptr g) -> Task { + co_await shield([](Event & e2) -> Task { co_await e2.wait(); }(e)); + (void)g; + }; + { + auto task = ctx_->create_task(hold(event, guard)); + spin_for(50ms); + } + spin_for(50ms); + EXPECT_EQ(guard.use_count(), 2); + event.set(); + spin_for(50ms); + EXPECT_EQ(guard.use_count(), 1); +} + +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); +} + +static Task set_flag(bool & flag) +{ + flag = true; + co_return; +} + +TEST_F(TaskDestroyTest, TaskDestroyedBeforeItStartsNeverRuns) +{ + bool ran = false; + { + auto task = ctx_->create_task(set_flag(ran)); + } + spin_for(50ms); + EXPECT_FALSE(ran); +} + +static Task record_cancel(CoContext & ctx, std::vector & order) +{ + try { + co_await ctx.sleep(10s); + } catch (const CancelledException &) { + order.push_back(1); + throw; + } +} + +static Task record_start(std::vector & order) +{ + order.push_back(2); + co_return; +} + +TEST_F(TaskDestroyTest, ReplacedTaskUnwindsBeforeItsReplacementStarts) +{ + std::vector order; + auto task = ctx_->create_task(record_cancel(*ctx_, order)); + spin_for(50ms); + task.cancel(); + task = ctx_->create_task(record_start(order)); + spin_for(50ms); + EXPECT_EQ(order, (std::vector{1, 2})); +} 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;