From d81640e992f2763eec0b46ec045453c51a6e7876 Mon Sep 17 00:00:00 2001 From: Tamaki Nishino Date: Sun, 18 Oct 2026 14:01:12 +0900 Subject: [PATCH 1/2] Track every waiting coroutine with Waiting Each awaiter guarded its handed-out handle in its own way: Event and Mutex an active flag, send_request and send_goal a shared State with a done flag and a destructor, the streams a destructor clearing waiter_, Sleep a done flag, and register_cancel an is_done predicate plus a weak_ptr to the stop callback. All of them now hold a Waiting, and whoever resumes them a weak_ptr to it. Waiting's handle is taken once, which decides between completion and cancellation, and the weak_ptr expiring with the awaiter keeps a destroyed one from being resumed. resume_now, post_resume and register_cancel(waiting, action) are the three ways to resume it. The per-awaiter flags, State structs, is_done predicates and destructors go away, except GoalStream::NextAwaiter's, which still cancels the goal. TfBuffer drops a destroyed awaiter's request on the next check instead of in the destructor. No behavior change otherwise. --- rclcpp_async/include/rclcpp_async/channel.hpp | 10 +- .../include/rclcpp_async/co_context.hpp | 188 ++++++------------ rclcpp_async/include/rclcpp_async/event.hpp | 35 +--- .../include/rclcpp_async/executor.hpp | 69 +++++-- .../include/rclcpp_async/goal_stream.hpp | 57 ++---- rclcpp_async/include/rclcpp_async/mutex.hpp | 34 +--- .../rclcpp_async/send_request_awaiter.hpp | 32 +-- .../include/rclcpp_async/sleep_awaiter.hpp | 2 +- rclcpp_async/include/rclcpp_async/task.hpp | 8 +- .../include/rclcpp_async/tf_buffer.hpp | 29 +-- .../include/rclcpp_async/timer_stream.hpp | 7 +- .../include/rclcpp_async/topic_stream.hpp | 15 +- 12 files changed, 173 insertions(+), 313 deletions(-) diff --git a/rclcpp_async/include/rclcpp_async/channel.hpp b/rclcpp_async/include/rclcpp_async/channel.hpp index 06e9198..e102227 100644 --- a/rclcpp_async/include/rclcpp_async/channel.hpp +++ b/rclcpp_async/include/rclcpp_async/channel.hpp @@ -101,16 +101,12 @@ bool Channel::NextAwaiter::await_suspend(std::coroutine_handle<> 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]() { - { - std::lock_guard lock(ch.mutex_); - if (ch.waiter_.lock() != waiting_) { - return; - } + register_cancel(cancel_cb_, token, ch.ctx_, waiting_, [this]() { + std::lock_guard lock(ch.mutex_); + if (ch.waiter_.lock() == waiting_) { ch.waiter_.reset(); } cancelled = true; - post_resume(ch.ctx_, waiting_); }); return true; } diff --git a/rclcpp_async/include/rclcpp_async/co_context.hpp b/rclcpp_async/include/rclcpp_async/co_context.hpp index b9f888e..2b0cf92 100644 --- a/rclcpp_async/include/rclcpp_async/co_context.hpp +++ b/rclcpp_async/include/rclcpp_async/co_context.hpp @@ -199,17 +199,12 @@ class CoContext : public Executor typename ServiceT::Request::SharedPtr request) { return SendRequestAwaiter{ - *this, - std::move(client), - std::move(request), - {}, - {}, - std::make_shared::State>()}; + *this, std::move(client), std::move(request), {}, {}, false, nullptr, nullptr}; } SleepAwaiter sleep(std::chrono::nanoseconds duration) { - return SleepAwaiter{*this, duration, nullptr, {}, {}, false, false}; + return SleepAwaiter{*this, duration, nullptr, {}, {}, false, nullptr}; } template @@ -237,11 +232,7 @@ class CoContext : public Executor while (s->queue_.size() > s->max_depth_) { s->queue_.pop(); } - if (s->waiter_) { - auto w = s->waiter_; - s->waiter_ = nullptr; - s->ctx_.resume(w); - } + resume_now(s->ctx_, std::exchange(s->waiter_, {})); }, subscription_options()); return stream; @@ -253,14 +244,7 @@ class CoContext : public Executor size_t max_depth = kDefaultStreamDepth) { return SendGoalAwaiter{ - *this, - std::move(client), - std::move(goal), - max_depth, - {}, - {}, - std::make_shared::State>(), - nullptr}; + *this, std::move(client), std::move(goal), max_depth, {}, {}, false, nullptr, {}, nullptr}; } template @@ -297,11 +281,7 @@ class CoContext : public Executor if (!s || s->closed_) { return; } - if (s->waiter_) { - auto w = s->waiter_; - s->waiter_ = nullptr; - resume(w); - } else { + if (!resume_now(*this, std::exchange(s->waiter_, {}))) { s->pending_++; } }, @@ -401,53 +381,37 @@ class CoContext : public Executor template void SendRequestAwaiter::await_suspend(std::coroutine_handle<> h) { - // Capture `state` by value (shared_ptr) so the callback stays valid even - // if the awaiter is destroyed (e.g. cancellation) before the response - // arrives. When `state->done` is already set by the cancel path, the - // response callback bails out before touching the (dangling) handle. + // The callback may outlive the awaiter, so it uses `this` only while + // waiting_ is alive. + waiting_ = std::make_shared(h); client->async_send_request( request, - [state = state_, &ctx = ctx, h](typename rclcpp::Client::SharedFuture future) { - if (state->done) { - return; + [this, w = std::weak_ptr(waiting_)](typename rclcpp::Client::SharedFuture future) { + if (auto s = w.lock()) { + response = future.get(); + resume_now(ctx, w); } - state->done = true; - state->response = future.get(); - ctx.resume(h); }); - register_cancel( - cancel_cb_, token, ctx, h, [state = state_]() { return state->done; }, - [state = state_]() { - state->done = true; - state->cancelled = true; - }); + register_cancel(cancel_cb_, token, ctx, waiting_, [this]() { cancelled = true; }); } inline void SleepAwaiter::await_suspend(std::coroutine_handle<> h) { - auto finish = [this, h]() { - if (done) { - return; - } - done = true; - timer->cancel(); - ctx.resume(h); - }; - - timer = ctx.node().create_wall_timer(duration, [finish]() { finish(); }, ctx.callback_group()); - - cancel_cb_ = std::make_shared(token, [this, h, &cb = cancel_cb_]() { - if (done) { - return; - } - done = true; - timer->cancel(); - cancelled = true; - ctx.post([h, weak = std::weak_ptr(cb)]() { - if (weak.lock()) { - h.resume(); - } + waiting_ = std::make_shared(h); + timer = ctx.node().create_wall_timer( + duration, + [this]() { + timer->cancel(); + resume_now(ctx, waiting_); + }, + ctx.callback_group()); + + // Cancelled on the spot, so a tick after the stop request cannot win. + cancel_cb_ = std::make_shared(token, [this]() { + post_resume(ctx, waiting_, [this]() { + timer->cancel(); + cancelled = true; }); }); } @@ -455,45 +419,37 @@ 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; }, - [this]() { - stream.waiter_ = nullptr; - cancelled = true; - }); + waiting_ = std::make_shared(h); + stream.waiter_ = waiting_; + register_cancel(cancel_cb_, token, stream.ctx_, waiting_, [this]() { + stream.waiter_.reset(); + cancelled = true; + }); } template void TopicStream::close() { closed_ = true; - if (waiter_) { - auto w = waiter_; - waiter_ = nullptr; - ctx_.resume(w); - } + resume_now(ctx_, std::exchange(waiter_, {})); } template -void GoalStream::resume_waiter(std::coroutine_handle<> h) +void GoalStream::resume_waiter(const std::weak_ptr & w) { - ctx_.resume(h); + resume_now(ctx_, w); } 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; }, - [this]() { - stream.waiter_ = nullptr; - cancelled = true; - stream.cancel_goal_on_stop(); - }); + waiting_ = std::make_shared(h); + stream.waiter_ = waiting_; + register_cancel(cancel_cb_, token, stream.ctx_, waiting_, [this]() { + stream.waiter_.reset(); + cancelled = true; + stream.cancel_goal_on_stop(); + }); } template @@ -506,7 +462,7 @@ void GoalStream::CancelAwaiter::await_suspend(std::coroutine_handle<> h [this, w = std::weak_ptr(waiting_)](typename CancelResponse::SharedPtr resp) { if (auto s = w.lock()) { response = std::move(resp); - stream.ctx_.resume(s->h); + resume_now(stream.ctx_, w); } }); } @@ -514,76 +470,60 @@ void GoalStream::CancelAwaiter::await_suspend(std::coroutine_handle<> h template void SendGoalAwaiter::await_suspend(std::coroutine_handle<> h) { - state_->stream = std::make_shared>(ctx, max_depth); + stream = std::make_shared>(ctx, max_depth); typename rclcpp_action::Client::SendGoalOptions options; - options.feedback_callback = [s = state_->stream](auto, const auto & feedback) { + options.feedback_callback = [s = stream](auto, const auto & feedback) { GoalEvent event; event.type = GoalEvent::Type::kFeedback; event.feedback = feedback; s->push(std::move(event)); }; - options.result_callback = [s = state_->stream](const auto & wrapped_result) { + options.result_callback = [s = stream](const auto & wrapped_result) { GoalEvent event; event.type = GoalEvent::Type::kComplete; event.result = wrapped_result; s->push(std::move(event)); }; - // Capture `state` by value (shared_ptr) so the callback stays valid even - // if the awaiter is destroyed before goal_response arrives (e.g. the - // awaiting coroutine was cancelled between async_send_goal() and the - // server's accept/reject). When `state->done` is already set by the - // cancel path, the callback bails out before touching the (dangling) - // coroutine handle. + // As in send_request, the callback uses `this` only while waiting_ is alive. 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) { + options.goal_response_callback = [this, w = std::weak_ptr(waiting_)](const auto & goal_handle) { + auto s = w.lock(); + if (!s) { return; } - state->done = true; if (!goal_handle) { - state->result = Result>>::Error("goal rejected"); + result = Result>>::Error("goal rejected"); } else { - state->stream->goal_handle_ = goal_handle; - state->stream->client_ = client; - state->result = Result>>::Ok(state->stream); + stream->goal_handle_ = goal_handle; + stream->client_ = client; + result = Result>>::Ok(stream); } // Use post() instead of resume() to defer coroutine resumption. // rclcpp_action holds goal_requests_mutex during this callback // (Jazzy bug: https://github.com/ros2/rclcpp/issues/2796). // Resuming synchronously here would deadlock if the coroutine // immediately calls async_send_goal again. - post_resume(ctx, state->waiting); + post_resume(ctx, w); }; client->async_send_goal(goal, options); - register_cancel( - cancel_cb_, token, ctx, h, [state = state_]() { return state->done; }, - [state = state_]() { - state->done = true; - state->cancelled = true; - }); + register_cancel(cancel_cb_, token, ctx, waiting_, [this]() { cancelled = true; }); } inline void TimerStream::NextAwaiter::await_suspend(std::coroutine_handle<> h) { - stream.waiter_ = h; - cancel_cb_ = std::make_shared(token, [this, h, &cb = cancel_cb_]() { - if (stream.waiter_ != h) { - return; - } - stream.cancel(); - cancelled = true; - stream.ctx_.post([h, weak = std::weak_ptr(cb)]() { - if (weak.lock()) { - h.resume(); - } + waiting_ = std::make_shared(h); + stream.waiter_ = waiting_; + // Stopped on the spot, as Sleep is. + cancel_cb_ = std::make_shared(token, [this]() { + post_resume(stream.ctx_, waiting_, [this]() { + stream.cancel(); + cancelled = true; }); }); } diff --git a/rclcpp_async/include/rclcpp_async/event.hpp b/rclcpp_async/include/rclcpp_async/event.hpp index feff88d..2e10cca 100644 --- a/rclcpp_async/include/rclcpp_async/event.hpp +++ b/rclcpp_async/include/rclcpp_async/event.hpp @@ -33,12 +33,7 @@ class Event Executor & ctx_; bool set_ = false; - struct Waiter - { - std::coroutine_handle<> handle; - std::shared_ptr active; - }; - std::queue waiters_; + std::queue> waiters_; public: explicit Event(Executor & ctx) : ctx_(ctx) {} @@ -54,17 +49,7 @@ class Event Event & event; std::stop_token token; 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; - } - } + std::shared_ptr waiting_; void set_token(std::stop_token t) { token = std::move(t); } @@ -92,24 +77,20 @@ class Event inline void Event::WaitAwaiter::await_suspend(std::coroutine_handle<> h) { - active = std::make_shared(true); - event.waiters_.push({h, active}); - auto a = active; - register_cancel(cancel_cb_, token, event.ctx_, h, [a]() { return !*a; }, [a]() { *a = false; }); + waiting_ = std::make_shared(h); + event.waiters_.push(waiting_); + register_cancel(cancel_cb_, token, event.ctx_, waiting_, []() {}); } inline void Event::set() { set_ = true; - std::queue pending; + std::queue> pending; pending.swap(waiters_); while (!pending.empty()) { - auto w = pending.front(); + auto w = std::move(pending.front()); pending.pop(); - if (*w.active) { - *w.active = false; - ctx_.resume(w.handle); - } + resume_now(ctx_, w); } } diff --git a/rclcpp_async/include/rclcpp_async/executor.hpp b/rclcpp_async/include/rclcpp_async/executor.hpp index 42344bc..dfc56d7 100644 --- a/rclcpp_async/include/rclcpp_async/executor.hpp +++ b/rclcpp_async/include/rclcpp_async/executor.hpp @@ -14,6 +14,7 @@ #pragma once +#include #include #include #include @@ -41,36 +42,64 @@ class Executor }; // A suspended coroutine, owned by its awaiter. Whoever resumes it later holds -// a weak_ptr, which expires with the awaiter. -struct Waiting +// a weak_ptr, which expires with the awaiter. The handle is taken once, so of +// the completion and the cancellation only the first resumes it. +class Waiting { - std::coroutine_handle<> h; + std::atomic> h_; + +public: + explicit Waiting(std::coroutine_handle<> h) : h_(h) {} + std::coroutine_handle<> take() { return h_.exchange(nullptr); } }; -// Resume `w` on the executor unless its awaiter is gone by then. -inline void post_resume(Executor & ctx, std::weak_ptr w) +// Take `w` and resume it here. False when its awaiter is gone or already taken. +inline bool resume_now(Executor & ctx, const std::weak_ptr & w) +{ + auto s = w.lock(); + auto h = s ? s->take() : nullptr; + if (!h) { + return false; + } + ctx.resume(h); + return true; +} + +// Take `w` now, run `claimed`, and resume it on the executor unless its awaiter +// is gone by then. +template +inline void post_resume(Executor & ctx, std::weak_ptr w, Claimed claimed) { - ctx.post([w = std::move(w)]() { - if (auto s = w.lock()) { - s->h.resume(); + auto s = w.lock(); + auto h = s ? s->take() : nullptr; + if (!h) { + return; + } + claimed(); + ctx.post([w = std::move(w), h]() { + if (w.lock()) { + h.resume(); } }); } -template +inline void post_resume(Executor & ctx, std::weak_ptr w) +{ + post_resume(ctx, std::move(w), []() {}); +} + +// On a stop request, resume `waiting` on the executor after `action`, unless +// it was taken or its awaiter destroyed first. `action` may use the awaiter. +template inline void register_cancel( - std::shared_ptr & out, std::stop_token token, Executor & ctx, std::coroutine_handle<> h, - DonePred is_done, CancelAction action) + std::shared_ptr & out, std::stop_token token, Executor & ctx, + const std::shared_ptr & waiting, CancelAction action) { - 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()) { + out = std::make_shared(token, [&ctx, w = std::weak_ptr(waiting), action]() { + ctx.post([&ctx, w, action]() { + auto s = w.lock(); + auto h = s ? s->take() : nullptr; + if (!h) { return; } action(); diff --git a/rclcpp_async/include/rclcpp_async/goal_stream.hpp b/rclcpp_async/include/rclcpp_async/goal_stream.hpp index 0545409..c91d8c2 100644 --- a/rclcpp_async/include/rclcpp_async/goal_stream.hpp +++ b/rclcpp_async/include/rclcpp_async/goal_stream.hpp @@ -61,7 +61,7 @@ class GoalStream CoContext & ctx_; std::queue queue_; - std::coroutine_handle<> waiter_; + std::weak_ptr waiter_; size_t max_depth_; bool completed_ = false; bool auto_cancel_on_stop_ = true; @@ -83,14 +83,10 @@ class GoalStream while (queue_.size() > max_depth_ && queue_.front().type != Event::Type::kComplete) { queue_.pop(); } - if (waiter_) { - auto h = waiter_; - waiter_ = nullptr; - resume_waiter(h); - } + resume_waiter(std::exchange(waiter_, {})); } - void resume_waiter(std::coroutine_handle<> h); + void resume_waiter(const std::weak_ptr & w); // The waiting coroutine was stopped: cancel the goal it was following. void cancel_goal_on_stop() @@ -114,16 +110,14 @@ class GoalStream std::stop_token token; std::shared_ptr cancel_cb_; bool cancelled = false; - std::coroutine_handle<> waiting_{}; + std::shared_ptr 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. + // The deferred cancellation does not run for a destroyed awaiter, so the + // goal it was following is cancelled here instead. ~NextAwaiter() { - if (waiting_ && stream.waiter_ == waiting_) { - stream.waiter_ = nullptr; + if (waiting_ && stream.waiter_.lock() == waiting_) { + stream.waiter_.reset(); stream.cancel_goal_on_stop(); } } @@ -157,7 +151,7 @@ class GoalStream } }; - NextAwaiter next() { return NextAwaiter{*this, {}, {}, false}; } + NextAwaiter next() { return NextAwaiter{*this, {}, {}, false, nullptr}; } using CancelResponse = typename ActionT::Impl::CancelGoalService::Response; @@ -184,44 +178,23 @@ class GoalStream template struct SendGoalAwaiter { - // Mutable state is held via shared_ptr so that callbacks registered on - // the rclcpp_action::Client (notably goal_response_callback) can safely - // observe `done` even if the awaiter (and its coroutine frame) has been - // destroyed after the awaiting coroutine was cancelled before the goal - // response arrived. - struct State - { - bool done = false; - bool cancelled = false; - std::shared_ptr> stream; - Result>> result; - std::weak_ptr waiting; - }; - CoContext & ctx; typename rclcpp_action::Client::SharedPtr client; typename ActionT::Goal goal; size_t max_depth; std::stop_token token; std::shared_ptr cancel_cb_; - std::shared_ptr state_; + bool cancelled = false; + std::shared_ptr> stream; + Result>> result; std::shared_ptr waiting_; - // 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() { if (token.stop_requested()) { - state_->cancelled = true; + cancelled = true; return true; } return false; @@ -232,10 +205,10 @@ struct SendGoalAwaiter Result>> await_resume() { cancel_cb_.reset(); - if (state_->cancelled) { + if (cancelled) { throw CancelledException{}; } - return std::move(state_->result); + return std::move(result); } }; diff --git a/rclcpp_async/include/rclcpp_async/mutex.hpp b/rclcpp_async/include/rclcpp_async/mutex.hpp index 07fcbd4..168b55f 100644 --- a/rclcpp_async/include/rclcpp_async/mutex.hpp +++ b/rclcpp_async/include/rclcpp_async/mutex.hpp @@ -33,12 +33,7 @@ class Mutex Executor & ctx_; bool locked_ = false; - struct Waiter - { - std::coroutine_handle<> handle; - std::shared_ptr active; - }; - std::queue waiters_; + std::queue> waiters_; public: explicit Mutex(Executor & ctx) : ctx_(ctx) {} @@ -50,17 +45,7 @@ class Mutex Mutex & mutex; std::stop_token token; 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; - } - } + std::shared_ptr waiting_; void set_token(std::stop_token t) { token = std::move(t); } @@ -81,7 +66,7 @@ class Mutex void await_resume() { cancel_cb_.reset(); - if (token.stop_requested() && !(active && *active)) { + if (token.stop_requested()) { throw CancelledException{}; } } @@ -94,20 +79,17 @@ class Mutex inline void Mutex::LockAwaiter::await_suspend(std::coroutine_handle<> h) { - active = std::make_shared(true); - mutex.waiters_.push({h, active}); - auto a = active; - register_cancel(cancel_cb_, token, mutex.ctx_, h, [a]() { return !*a; }, [a]() { *a = false; }); + waiting_ = std::make_shared(h); + mutex.waiters_.push(waiting_); + register_cancel(cancel_cb_, token, mutex.ctx_, waiting_, []() {}); } inline void Mutex::unlock() { while (!waiters_.empty()) { - auto w = waiters_.front(); + auto w = std::move(waiters_.front()); waiters_.pop(); - if (*w.active) { - *w.active = false; - ctx_.resume(w.handle); + if (resume_now(ctx_, w)) { return; } } diff --git a/rclcpp_async/include/rclcpp_async/send_request_awaiter.hpp b/rclcpp_async/include/rclcpp_async/send_request_awaiter.hpp index 1935730..d31e3ac 100644 --- a/rclcpp_async/include/rclcpp_async/send_request_awaiter.hpp +++ b/rclcpp_async/include/rclcpp_async/send_request_awaiter.hpp @@ -35,41 +35,21 @@ struct SendRequestAwaiter { using Response = typename ServiceT::Response::SharedPtr; - // Mutable state is held via shared_ptr so that the response callback - // registered on the rclcpp::Client can safely observe `done` even if - // the awaiter (and its coroutine frame) has been destroyed after the - // awaiting coroutine was cancelled. - struct State - { - bool done = false; - bool cancelled = false; - Response response; - }; - CoContext & ctx; typename rclcpp::Client::SharedPtr client; typename ServiceT::Request::SharedPtr request; std::stop_token token; 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; - } - } + bool cancelled = false; + Response response; + std::shared_ptr waiting_; void set_token(std::stop_token t) { token = std::move(t); } bool await_ready() { if (token.stop_requested()) { - state_->cancelled = true; + cancelled = true; return true; } return false; @@ -81,10 +61,10 @@ struct SendRequestAwaiter Response await_resume() { cancel_cb_.reset(); - if (state_->cancelled) { + if (cancelled) { throw CancelledException{}; } - return std::move(state_->response); + return std::move(response); } }; diff --git a/rclcpp_async/include/rclcpp_async/sleep_awaiter.hpp b/rclcpp_async/include/rclcpp_async/sleep_awaiter.hpp index 86132cb..9bbff31 100644 --- a/rclcpp_async/include/rclcpp_async/sleep_awaiter.hpp +++ b/rclcpp_async/include/rclcpp_async/sleep_awaiter.hpp @@ -39,7 +39,7 @@ struct SleepAwaiter std::stop_token token; std::shared_ptr cancel_cb_; bool cancelled = false; - bool done = false; + std::shared_ptr waiting_; void set_token(std::stop_token t) { token = std::move(t); } diff --git a/rclcpp_async/include/rclcpp_async/task.hpp b/rclcpp_async/include/rclcpp_async/task.hpp index fa07f03..9cb81d8 100644 --- a/rclcpp_async/include/rclcpp_async/task.hpp +++ b/rclcpp_async/include/rclcpp_async/task.hpp @@ -56,7 +56,9 @@ struct Task std::coroutine_handle<> await_suspend(std::coroutine_handle h) noexcept { if (auto c = h.promise().continuation.lock()) { - return c->h; + if (auto next = c->take()) { + return next; + } } return std::noop_coroutine(); } @@ -196,7 +198,9 @@ struct Task std::coroutine_handle<> await_suspend(std::coroutine_handle h) noexcept { if (auto c = h.promise().continuation.lock()) { - return c->h; + if (auto next = c->take()) { + return next; + } } return std::noop_coroutine(); } diff --git a/rclcpp_async/include/rclcpp_async/tf_buffer.hpp b/rclcpp_async/include/rclcpp_async/tf_buffer.hpp index 6e0e76f..0ba8a81 100644 --- a/rclcpp_async/include/rclcpp_async/tf_buffer.hpp +++ b/rclcpp_async/include/rclcpp_async/tf_buffer.hpp @@ -46,7 +46,6 @@ class TfBuffer std::string source_frame; tf2::TimePoint time; std::weak_ptr waiting; - std::shared_ptr active; geometry_msgs::msg::TransformStamped result; }; @@ -112,17 +111,6 @@ class TfBuffer 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 - // 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() @@ -147,16 +135,12 @@ class TfBuffer request->time = time; waiting = std::make_shared(h); request->waiting = waiting; - request->active = std::make_shared(true); tf.add_pending(request); - register_cancel( - cancel_cb_, token, tf.ctx_, h, [r = request]() { return !*r->active; }, - [this, &tf_buf = tf, r = request]() { - tf_buf.remove_pending(r); - *r->active = false; - cancelled = true; - }); + register_cancel(cancel_cb_, token, tf.ctx_, waiting, [this]() { + tf.remove_pending(request); + cancelled = true; + }); } geometry_msgs::msg::TransformStamped await_resume() @@ -206,8 +190,8 @@ class TfBuffer { std::lock_guard lock(mutex_); std::erase_if(pending_, [this](const std::shared_ptr & req) { - if (!*req->active) { - return true; + if (req->waiting.expired()) { + return true; // its awaiter is gone } if (buffer_->canTransform(req->target_frame, req->source_frame, req->time)) { try { @@ -215,7 +199,6 @@ class TfBuffer } catch (const tf2::TransformException &) { return false; } - *req->active = false; post_resume(ctx_, req->waiting); return true; } diff --git a/rclcpp_async/include/rclcpp_async/timer_stream.hpp b/rclcpp_async/include/rclcpp_async/timer_stream.hpp index cca136d..a01c05b 100644 --- a/rclcpp_async/include/rclcpp_async/timer_stream.hpp +++ b/rclcpp_async/include/rclcpp_async/timer_stream.hpp @@ -34,7 +34,7 @@ class TimerStream { CoContext & ctx_; rclcpp::TimerBase::SharedPtr timer_; - std::coroutine_handle<> waiter_; + std::weak_ptr waiter_; int pending_ = 0; bool closed_ = false; @@ -49,6 +49,7 @@ class TimerStream 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); } @@ -76,12 +77,12 @@ class TimerStream } }; - NextAwaiter next() { return NextAwaiter{*this, {}, {}, false}; } + NextAwaiter next() { return NextAwaiter{*this, {}, {}, false, nullptr}; } void cancel() { closed_ = true; - waiter_ = nullptr; + waiter_.reset(); if (timer_) { timer_->cancel(); } diff --git a/rclcpp_async/include/rclcpp_async/topic_stream.hpp b/rclcpp_async/include/rclcpp_async/topic_stream.hpp index bef0e2b..975de8b 100644 --- a/rclcpp_async/include/rclcpp_async/topic_stream.hpp +++ b/rclcpp_async/include/rclcpp_async/topic_stream.hpp @@ -38,7 +38,7 @@ class TopicStream { CoContext & ctx_; std::queue> queue_; - std::coroutine_handle<> waiter_; + std::weak_ptr waiter_; typename rclcpp::Subscription::SharedPtr sub_; size_t max_depth_; bool closed_ = false; @@ -59,16 +59,7 @@ 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; - } - } + std::shared_ptr waiting_; void set_token(std::stop_token t) { token = std::move(t); } @@ -98,7 +89,7 @@ class TopicStream } }; - NextAwaiter next() { return NextAwaiter{*this, {}, {}, false}; } + NextAwaiter next() { return NextAwaiter{*this, {}, {}, false, nullptr}; } void close(); }; From 9852436ee4cb294808b4110b51f7f161a1d59650 Mon Sep 17 00:00:00 2001 From: Tamaki Nishino Date: Sun, 18 Oct 2026 15:06:02 +0900 Subject: [PATCH 2/2] Withdraw a Channel waiter in the stop callback, as before register_cancel deferred withdrawing it to the executor, so a send that arrived in between won and returned a value despite the stop request. --- rclcpp_async/include/rclcpp_async/channel.hpp | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/rclcpp_async/include/rclcpp_async/channel.hpp b/rclcpp_async/include/rclcpp_async/channel.hpp index e102227..8e0f277 100644 --- a/rclcpp_async/include/rclcpp_async/channel.hpp +++ b/rclcpp_async/include/rclcpp_async/channel.hpp @@ -101,12 +101,16 @@ bool Channel::NextAwaiter::await_suspend(std::coroutine_handle<> h) waiting_ = std::make_shared(h); ch.waiter_ = waiting_; } // release ch.mutex_ before stop_callback to avoid deadlock - register_cancel(cancel_cb_, token, ch.ctx_, waiting_, [this]() { - std::lock_guard lock(ch.mutex_); - if (ch.waiter_.lock() == waiting_) { + // Withdrawn on the spot, so a send after the stop request cannot win. + cancel_cb_ = std::make_shared(token, [this]() { + { + std::lock_guard lock(ch.mutex_); + if (ch.waiter_.lock() != waiting_) { + return; + } ch.waiter_.reset(); } - cancelled = true; + post_resume(ch.ctx_, waiting_, [this]() { cancelled = true; }); }); return true; }