Detach a running task instead of destroying its frame

docs/guide.md

@@ -4,7 +4,14 @@
 
 ### create_task
 
-`create_task` registers and starts a coroutine. It accepts either a `Task<T>` directly or a callable that returns one.
+`create_task` schedules a coroutine on the context. It accepts either a `Task<T>` 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<void> 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`.

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 <typename T>
   [[nodiscard]] Task<T> create_task(Task<T> 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 <typename MsgT>
 void TopicStream<MsgT>::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<ActionT>::resume_waiter(std::coroutine_handle<> h)
 template <typename ActionT>
 void GoalStream<ActionT>::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; },

rclcpp_async/include/rclcpp_async/event.hpp

@@ -56,16 +56,6 @@ class Event
     std::shared_ptr<StopCb> cancel_cb_;
     std::shared_ptr<bool> 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()

rclcpp_async/include/rclcpp_async/executor.hpp

@@ -46,9 +46,8 @@ inline void register_cancel(
 {
   out = std::make_shared<StopCb>(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;
       }

rclcpp_async/include/rclcpp_async/goal_stream.hpp

@@ -114,19 +114,6 @@ class GoalStream
     std::stop_token token;
     std::shared_ptr<StopCb> 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<StopCb> cancel_cb_;
   std::shared_ptr<State> 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()

rclcpp_async/include/rclcpp_async/mutex.hpp

@@ -52,16 +52,6 @@ class Mutex
     std::shared_ptr<StopCb> cancel_cb_;
     std::shared_ptr<bool> 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()

rclcpp_async/include/rclcpp_async/send_request_awaiter.hpp

@@ -53,17 +53,6 @@ struct SendRequestAwaiter
   std::shared_ptr<StopCb> cancel_cb_;
   std::shared_ptr<State> 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()

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<promise_type> 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<promise_type> 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<void>
     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<void>
       bool await_ready() noexcept { return false; }
       std::coroutine_handle<> await_suspend(std::coroutine_handle<promise_type> 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<void>
   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<void>
   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<void>
 
 private:
   explicit Task(std::coroutine_handle<promise_type> 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

rclcpp_async/include/rclcpp_async/tf_buffer.hpp

@@ -111,17 +111,6 @@ class TfBuffer
     bool cancelled = false;
     std::shared_ptr<PendingRequest> 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()

rclcpp_async/include/rclcpp_async/topic_stream.hpp

@@ -59,16 +59,6 @@ class TopicStream
     std::stop_token token;
     std::shared_ptr<StopCb> 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); }
 

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<int> ch(*ctx_);
+  auto coro = [&]() -> Task<void> { 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<int> ch(*ctx_);
+  auto coro = [&]() -> Task<void> { 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);
+  }
+}