From ba75e1bef49dbe68603848b0932a06607d6e876d Mon Sep 17 00:00:00 2001 From: Steve Gerbino Date: Fri, 17 Jul 2026 16:48:32 +0200 Subject: [PATCH] perf(timer): embed the waiter node in the awaitable frame Each co_await on a timer previously allocated a waiter_node from the service, amortized through a thread-local cache and a mutex-protected free list. The node now lives inside the awaitable on the suspended coroutine's frame, so a wait performs no allocation and never touches the free-list mutex. The stop token is borrowed from the awaiting chain's io_env and the stop callback is constructed in place in a raw buffer, so the node owns no destructible state. Completion paths destroy the callback and finish touching the node before posting the continuation or destroying the coroutine, whose frame is the node's storage. completion_op and canceller are defined non-inline in timer.cpp because the inline waiter_node constructor references do_complete and the vtable from translation units that reach timer.hpp through delay.hpp without including timer_service.hpp. The waiter free list, thread-local waiter cache, and per-wait create/destroy machinery are removed. --- include/boost/corosio/detail/timer.hpp | 272 +++++++++++++---- .../boost/corosio/detail/timer_service.hpp | 282 ++++-------------- src/corosio/src/timer.cpp | 101 +++++-- test/unit/delay.cpp | 33 ++ 4 files changed, 375 insertions(+), 313 deletions(-) diff --git a/include/boost/corosio/detail/timer.hpp b/include/boost/corosio/detail/timer.hpp index 67da610f2..a8e835fb0 100644 --- a/include/boost/corosio/detail/timer.hpp +++ b/include/boost/corosio/detail/timer.hpp @@ -13,6 +13,7 @@ #include #include +#include #include #include #include @@ -28,19 +29,22 @@ #include #include #include +#include #include #include #include namespace boost::corosio::detail { -// Defined in timer_service.hpp, which includes this header. The -// implementation::wait() body needs waiter_node and timer_service to -// be complete, so it is declared here and defined there; intrusive_list -// only stores waiter_node pointers, so the forward declaration suffices -// for implementation's data layout. +// timer_service is defined in timer_service.hpp, which includes this +// header. waiter_node and wait_awaitable are defined below the timer +// class: waiter_node stores a timer::implementation*, which cannot be +// forward-declared as a nested type. intrusive_list only stores +// waiter_node pointers, so this forward declaration suffices for +// implementation's data layout. class timer_service; struct waiter_node; +struct wait_awaitable; /** An asynchronous timer for coroutine I/O. @@ -68,46 +72,7 @@ struct waiter_node; */ class BOOST_COROSIO_DECL timer : public io_object { - struct wait_awaitable - { - timer& t_; - std::error_code ec_; - capy::continuation cont_; - - explicit wait_awaitable(timer& t) noexcept : t_(t) {} - - bool await_ready() const noexcept - { - return false; - } - - // Cancellation surfaces through ec_: the stop_token path in - // wait() completes the waiter with error::canceled written to - // ec_, so there is no separate token to consult here. - capy::io_result<> await_resume() const noexcept - { - return {ec_}; - } - - auto await_suspend(std::coroutine_handle<> h, capy::io_env const* env) - -> std::coroutine_handle<> - { - cont_.h = h; - auto& impl = t_.get(); - // Inline fast path: already expired and not in the heap - if (impl.heap_index_.load(std::memory_order_relaxed) == - implementation::npos && - (impl.expiry_ == (time_point::min)() || - impl.expiry_ <= clock_type::now())) - { - ec_ = {}; - auto d = env->executor; - d.post(cont_); - return std::noop_coroutine(); - } - return impl.wait(h, env->executor, env->stop_token, &ec_, &cont_); - } - }; + friend struct wait_awaitable; public: /** Backend state and wait entry point for a timer. @@ -158,18 +123,40 @@ class BOOST_COROSIO_DECL timer : public io_object /// Construct bound to the given timer service. explicit implementation(timer_service& svc) noexcept : svc_(&svc) {} - /// Initiate an asynchronous wait for the timer to expire. + /** Check whether the timer is expired and absent from the heap. + + The single definition of the already-expired fast-path + predicate: `await_suspend` tests it inline and `wait()` + re-tests it because the expiry can elapse between the two + reads. + */ + bool already_expired() const noexcept + { + return heap_index_.load(std::memory_order_relaxed) == npos && + (expiry_ == + (std::chrono::steady_clock::time_point::min)() || + expiry_ <= std::chrono::steady_clock::now()); + } + + /** Asynchronously wait for the timer to expire. + + Publishes the waiter into the service's heap and waiter + list, after which it may complete on any thread. If the + timer is already expired and not in the heap, completes + by posting the continuation without publishing. + + @par Preconditions + @p w is fully initialized, and its storage (the awaitable + on the suspended coroutine's frame) outlives the wait. + + @param w The waiter to publish. + */ // Exported at member level: dllexport on the enclosing timer // class does not extend to nested classes, and header-inline // callers (wait_awaitable::await_suspend) reference this // symbol from outside the corosio DLL. BOOST_COROSIO_DECL - std::coroutine_handle<> wait( - std::coroutine_handle<>, - capy::executor_ref, - std::stop_token, - std::error_code*, - capy::continuation*); + std::coroutine_handle<> wait(waiter_node& w); }; /// The clock type used for time operations. @@ -419,10 +406,8 @@ class BOOST_COROSIO_DECL timer : public io_object @return An awaitable that completes with `io_result<>`. */ - auto wait() - { - return wait_awaitable(*this); - } + // Defined below wait_awaitable, which needs timer complete. + wait_awaitable wait(); protected: explicit timer(handle h) noexcept : io_object(std::move(h)) {} @@ -442,6 +427,181 @@ class BOOST_COROSIO_DECL timer : public io_object } }; +/** Frame-resident per-wait state for a timer wait. + + One node exists per `co_await` on a timer, embedded in the + awaitable on the suspended coroutine's frame — never allocated. + Once published by `implementation::wait()` the node may be + completed from any thread; every completion path finishes + touching the node before resuming or destroying the coroutine, + because either act may end the node's storage. + + The node owns no resources: the stop token is borrowed from the + awaiting chain's `io_env` (which outlives the suspension) and + the stop callback is managed manually in `cb_buf_`, destroyed on + every completion path before the frame can die. +*/ +struct BOOST_COROSIO_SYMBOL_VISIBLE waiter_node + : intrusive_list::node +{ + // Embedded completion op — avoids heap allocation per fire/cancel. + // Members are exported and defined non-inline in timer.cpp: the + // inline waiter_node constructor references do_complete and the + // vtable from translation units that reach this header through + // delay.hpp without ever including timer_service.hpp, so the one + // strong definition must live in a TU that is always linked. + struct BOOST_COROSIO_SYMBOL_VISIBLE completion_op final : scheduler_op + { + waiter_node* waiter_ = nullptr; + + BOOST_COROSIO_DECL + static void do_complete( + void* owner, scheduler_op* base, std::uint32_t, std::uint32_t); + + completion_op() noexcept : scheduler_op(&do_complete) {} + + BOOST_COROSIO_DECL void operator()() override; + BOOST_COROSIO_DECL void destroy() override; + }; + + // Per-waiter stop_token cancellation + struct canceller + { + waiter_node* waiter_; + BOOST_COROSIO_DECL void operator()() const; + }; + + using stop_cb_type = std::stop_callback; + + // nullptr once removed from timer's waiter list (concurrency marker) + /// The timer this waiter is published on, or `nullptr`. + timer::implementation* impl_ = nullptr; + + /// The timer service that completes this waiter. + timer_service* svc_ = nullptr; + + /// The suspended coroutine, destroyed by the shutdown drains. + std::coroutine_handle<> h_; + + /// The continuation posted to resume the coroutine. + capy::continuation cont_; + + /// The executor the continuation is posted through. + capy::executor_ref d_; + + // Borrowed from the awaiting chain's io_env, which outlives the + // suspension; the node holds no owning state. + /// The stop token observed for cancellation. + std::stop_token const* token_ = nullptr; + + /// The completion result read by `await_resume`. + std::error_code ec_; + + /// The embedded completion op posted to the scheduler. + completion_op op_; + + // stop_callback is neither movable nor assignable; construct it + // in place once the node is pinned on the coroutine frame, and + // destroy it manually on every completion path. + /// Storage for the armed stop callback. + alignas(stop_cb_type) unsigned char cb_buf_[sizeof(stop_cb_type)]; + + /// True while `cb_buf_` holds a live stop callback. + bool cb_active_ = false; + + waiter_node() noexcept + { + op_.waiter_ = this; + } + + // The embedded op self-points and the list hooks are published + // to other threads; the node never moves. + waiter_node(waiter_node const&) = delete; + waiter_node& operator=(waiter_node const&) = delete; + + /** Arm the stop callback. + + @par Preconditions + `token_` is set. + */ + void arm_stop_cb() + { + new (cb_buf_) stop_cb_type(*token_, canceller{this}); + cb_active_ = true; + } + + /// Destroy the stop callback if armed. + void reset_stop_cb() noexcept + { + if (cb_active_) + { + std::launder(reinterpret_cast(cb_buf_)) + ->~stop_cb_type(); + cb_active_ = false; + } + } +}; + +/** Awaitable returned by `timer::wait()`. + + Carries the waiter node so a wait performs no allocation. The + awaitable is movable only before `await_suspend` publishes the + node (a move builds a fresh, quiescent node); afterwards it is + pinned on the coroutine frame until the wait completes. +*/ +struct wait_awaitable +{ + timer& t_; + waiter_node w_; + + explicit wait_awaitable(timer& t) noexcept : t_(t) {} + + wait_awaitable(wait_awaitable&& o) noexcept : t_(o.t_) {} + + wait_awaitable(wait_awaitable const&) = delete; + wait_awaitable& operator=(wait_awaitable const&) = delete; + wait_awaitable& operator=(wait_awaitable&&) = delete; + + bool await_ready() const noexcept + { + return false; + } + + // Cancellation surfaces through w_.ec_: the stop_token path in + // wait() completes the waiter with error::canceled written to + // it, so there is no separate token to consult here. + capy::io_result<> await_resume() const noexcept + { + return {w_.ec_}; + } + + auto await_suspend(std::coroutine_handle<> h, capy::io_env const* env) + -> std::coroutine_handle<> + { + auto& impl = t_.get(); + w_.h_ = h; + w_.cont_.h = h; + w_.d_ = env->executor; + + // Inline fast path: already expired and not in the heap + if (impl.already_expired()) + { + w_.ec_ = {}; + w_.d_.post(w_.cont_); + return std::noop_coroutine(); + } + + w_.token_ = &env->stop_token; + return impl.wait(w_); + } +}; + +inline wait_awaitable +timer::wait() +{ + return wait_awaitable(*this); +} + } // namespace boost::corosio::detail #endif diff --git a/include/boost/corosio/detail/timer_service.hpp b/include/boost/corosio/detail/timer_service.hpp index 9b77a8895..f6c0d1ac2 100644 --- a/include/boost/corosio/detail/timer_service.hpp +++ b/include/boost/corosio/detail/timer_service.hpp @@ -27,7 +27,6 @@ #include #include #include -#include #include #include #include @@ -42,37 +41,40 @@ struct scheduler; Data Structures --------------- - waiter_node holds per-waiter state: coroutine handle, executor, - error output, stop_token, embedded completion_op. Each concurrent - co_await t.wait() allocates one waiter_node. + waiter_node (defined in timer.hpp) holds per-waiter state: + coroutine handle, executor, error output, embedded + completion_op. Each concurrent co_await t.wait() embeds one + waiter_node in the awaitable on the suspended coroutine's + frame — waits perform no allocation. timer::implementation holds per-timer state: expiry, heap index, and an intrusive_list of waiter_nodes. Multiple coroutines can wait on the same timer simultaneously. - timer_service owns a min-heap of active timers, a free list - of recycled impls, and a free list of recycled waiter_nodes. The - heap is ordered by expiry time; the scheduler queries - nearest_expiry() to set the epoll/timerfd timeout. + timer_service owns a min-heap of active timers and a free list + of recycled impls. The heap is ordered by expiry time; the + scheduler queries nearest_expiry() to set the epoll/timerfd + timeout. Optimization Strategy --------------------- 1. Deferred heap insertion — expires_after() stores the expiry but does not insert into the heap. Insertion happens in wait(). 2. Thread-local impl cache — single-slot per-thread cache. - 3. Embedded completion_op — eliminates heap allocation per fire/cancel. + 3. Frame-resident waiter_node with embedded completion_op — + eliminates heap allocation per wait/fire/cancel. 4. Cached nearest expiry — atomic avoids mutex in nearest_expiry(). 5. might_have_pending_waits_ flag — skips lock when no wait issued. - 6. Thread-local waiter cache — single-slot per-thread cache. Concurrency ----------- stop_token callbacks can fire from any thread. The impl_ pointer on waiter_node is used as a "still in list" marker. + A waiter_node's storage is the suspended coroutine's frame: + every completion path must finish touching the node before + posting the continuation or destroying the handle. */ -struct BOOST_COROSIO_SYMBOL_VISIBLE waiter_node; - inline void timer_service_invalidate_cache() noexcept; // timer_service class body — member function definitions are @@ -125,7 +127,6 @@ class BOOST_COROSIO_DECL timer_service final mutable std::mutex mutex_; std::vector heap_; timer::implementation* free_list_ = nullptr; - waiter_node* waiter_free_list_ = nullptr; callback on_earliest_changed_; bool shutting_down_ = false; // Avoids mutex in nearest_expiry() and empty() @@ -184,12 +185,6 @@ class BOOST_COROSIO_DECL timer_service final /// Cancel and recycle a timer implementation. inline void destroy_impl(timer::implementation& impl); - /// Create or recycle a waiter node. - inline waiter_node* create_waiter(); - - /// Return a waiter node to the cache or free list. - inline void destroy_waiter(waiter_node* w); - /// Update the timer expiry, cancelling existing waiters. inline std::size_t update_timer( timer::implementation& impl, time_point new_time); @@ -200,10 +195,10 @@ class BOOST_COROSIO_DECL timer_service final /// Cancel all waiters on a timer. inline std::size_t cancel_timer(timer::implementation& impl); - /// Cancel a single waiter ( stop_token callback path ). + /// Cancel one specific waiter ( stop_token callback path ). inline void cancel_waiter(waiter_node* w); - /// Cancel one waiter on a timer. + /// Cancel the oldest pending waiter on a timer ( FIFO ). inline std::size_t cancel_one_waiter(timer::implementation& impl); /// Complete all waiters whose timers have expired. @@ -223,73 +218,25 @@ class BOOST_COROSIO_DECL timer_service final inline void swap_heap(std::size_t i1, std::size_t i2); }; -struct BOOST_COROSIO_SYMBOL_VISIBLE waiter_node - : intrusive_list::node -{ - // Embedded completion op — avoids heap allocation per fire/cancel - struct completion_op final : scheduler_op - { - waiter_node* waiter_ = nullptr; - - static void do_complete( - void* owner, scheduler_op* base, std::uint32_t, std::uint32_t); - - completion_op() noexcept : scheduler_op(&do_complete) {} - - void operator()() override; - void destroy() override; - }; - - // Per-waiter stop_token cancellation - struct canceller - { - waiter_node* waiter_; - void operator()() const; - }; - - // nullptr once removed from timer's waiter list (concurrency marker) - timer::implementation* impl_ = nullptr; - timer_service* svc_ = nullptr; - std::coroutine_handle<> h_; - capy::continuation* cont_ = nullptr; - capy::executor_ref d_; - std::error_code* ec_out_ = nullptr; - std::stop_token token_; - std::optional> stop_cb_; - completion_op op_; - std::error_code ec_value_; - waiter_node* next_free_ = nullptr; - - waiter_node() noexcept - { - op_.waiter_ = this; - } -}; - -// Thread-local caches avoid hot-path mutex acquisitions: -// 1. Impl cache — single-slot, validated by comparing svc_ -// 2. Waiter cache — single-slot, no service affinity -// All caches are cleared by timer_service_invalidate_cache() during shutdown. +// Thread-local cache avoids hot-path mutex acquisitions: +// single-slot impl cache, validated by comparing svc_. Cleared by +// timer_service_invalidate_cache() during shutdown. inline thread_local_ptr tl_cached_impl; -inline thread_local_ptr tl_cached_waiter; - -// The POD TLS slots above never run destructors, so a short-lived -// run() thread would leak its cached impl and waiter. Each push -// arms this owner, whose destructor frees the slots at thread exit. -// Cached entries are quiescent heap objects (stop callbacks reset, -// nothing in the heap or free lists) and deletion touches no -// service state, so it is safe after the owning service is gone -// (the stale-entry path in try_pop_tl_cache deletes the same way). + +// The POD TLS slot above never runs destructors, so a short-lived +// run() thread would leak its cached impl. Each push arms this +// owner, whose destructor frees the slot at thread exit. A cached +// entry is a quiescent heap object (nothing in the heap or free +// list) and deletion touches no service state, so it is safe after +// the owning service is gone (the stale-entry path in +// try_pop_tl_cache deletes the same way). struct tl_cache_owner { ~tl_cache_owner() { delete tl_cached_impl.get(); tl_cached_impl.set(nullptr); - - delete tl_cached_waiter.get(); - tl_cached_waiter.set(nullptr); } }; @@ -327,38 +274,11 @@ try_push_tl_cache(timer::implementation* impl) noexcept return false; } -inline waiter_node* -try_pop_waiter_tl_cache() noexcept -{ - auto* w = tl_cached_waiter.get(); - if (w) - { - tl_cached_waiter.set(nullptr); - return w; - } - return nullptr; -} - -inline bool -try_push_waiter_tl_cache(waiter_node* w) noexcept -{ - if (!tl_cached_waiter.get()) - { - arm_tl_cache_cleanup(); - tl_cached_waiter.set(w); - return true; - } - return false; -} - inline void timer_service_invalidate_cache() noexcept { delete tl_cached_impl.get(); tl_cached_impl.set(nullptr); - - delete tl_cached_waiter.get(); - tl_cached_waiter.set(nullptr); } // timer_service out-of-class member function definitions @@ -395,12 +315,12 @@ timer_service::shutdown() { while (auto* w = impl->waiters_.pop_front()) { - w->stop_cb_.reset(); + w->reset_stop_cb(); auto h = std::exchange(w->h_, {}); sched_->work_finished(); + // Destroying the frame also ends the node's storage if (h) h.destroy(); - delete w; } delete impl; } @@ -412,14 +332,6 @@ timer_service::shutdown() delete free_list_; free_list_ = next; } - - // Delete free-listed waiters - while (waiter_free_list_) - { - auto* next = waiter_free_list_->next_free_; - delete waiter_free_list_; - waiter_free_list_ = next; - } } inline io_object::implementation* @@ -428,7 +340,10 @@ timer_service::construct() timer::implementation* impl = try_pop_tl_cache(this); if (impl) { - impl->svc_ = this; + impl->svc_ = this; + // Reset expiry_ too: a recycled impl must behave like a fresh + // one, whose default expiry reads as already elapsed + impl->expiry_ = {}; impl->heap_index_.store( (std::numeric_limits::max)(), std::memory_order_relaxed); @@ -443,6 +358,7 @@ timer_service::construct() free_list_ = impl->next_free_; impl->next_free_ = nullptr; impl->svc_ = this; + impl->expiry_ = {}; impl->heap_index_.store( (std::numeric_limits::max)(), std::memory_order_relaxed); @@ -497,42 +413,14 @@ timer_service::destroy_impl(timer::implementation& impl) free_list_ = &impl; } -inline waiter_node* -timer_service::create_waiter() -{ - if (auto* w = try_pop_waiter_tl_cache()) - return w; - - std::lock_guard lock(mutex_); - if (waiter_free_list_) - { - auto* w = waiter_free_list_; - waiter_free_list_ = w->next_free_; - w->next_free_ = nullptr; - return w; - } - - return new waiter_node(); -} - -inline void -timer_service::destroy_waiter(waiter_node* w) -{ - if (try_push_waiter_tl_cache(w)) - return; - - std::lock_guard lock(mutex_); - w->next_free_ = waiter_free_list_; - waiter_free_list_ = w; -} - inline std::size_t timer_service::update_timer(timer::implementation& impl, time_point new_time) { // Gate on the flag, not waiters_: reading the non-atomic list // here would race a concurrent drain. A false flag is safe to - // trust pre-lock because every draining critical section stores - // it false as its final touch of the impl. + // trust pre-lock: wait() stores it true before publishing, and + // it is cleared only under the mutex when the waiter list is + // empty, so false implies no published waiters. bool in_heap = (impl.heap_index_.load(std::memory_order_relaxed) != (std::numeric_limits::max)()); @@ -573,7 +461,7 @@ timer_service::update_timer(timer::implementation& impl, time_point new_time) std::size_t count = 0; while (auto* w = canceled.pop_front()) { - w->ec_value_ = make_error_code(capy::error::canceled); + w->ec_ = make_error_code(capy::error::canceled); sched_->post(&w->op_); ++count; } @@ -609,7 +497,7 @@ timer_service::insert_waiter(timer::implementation& impl, waiter_node* w) // Lost-cancel re-check: a stop requested after the canceller was // armed in wait() but before this publication found impl_ null // and returned a no-op. Observe it now and undo the insertion. - if (w->token_.stop_requested()) + if (w->token_->stop_requested()) { w->impl_ = nullptr; impl.waiters_.remove(w); @@ -628,7 +516,7 @@ timer_service::insert_waiter(timer::implementation& impl, waiter_node* w) on_earliest_changed_(); if (lost_cancel) { - w->ec_value_ = make_error_code(capy::error::canceled); + w->ec_ = make_error_code(capy::error::canceled); sched_->post(&w->op_); } } @@ -665,7 +553,7 @@ timer_service::cancel_timer(timer::implementation& impl) std::size_t count = 0; while (auto* w = canceled.pop_front()) { - w->ec_value_ = make_error_code(capy::error::canceled); + w->ec_ = make_error_code(capy::error::canceled); sched_->post(&w->op_); ++count; } @@ -678,7 +566,9 @@ timer_service::cancel_waiter(waiter_node* w) { { std::lock_guard lock(mutex_); - // Already removed by cancel_timer or process_expired + // Already removed by another drain: cancel_timer, + // cancel_one_waiter, update_timer, process_expired, or + // insert_waiter's lost-cancel recheck if (!w->impl_) return; auto* impl = w->impl_; @@ -693,7 +583,7 @@ timer_service::cancel_waiter(waiter_node* w) refresh_cached_nearest(); } - w->ec_value_ = make_error_code(capy::error::canceled); + w->ec_ = make_error_code(capy::error::canceled); sched_->post(&w->op_); } @@ -720,7 +610,7 @@ timer_service::cancel_one_waiter(timer::implementation& impl) refresh_cached_nearest(); } - w->ec_value_ = make_error_code(capy::error::canceled); + w->ec_ = make_error_code(capy::error::canceled); sched_->post(&w->op_); return 1; } @@ -740,8 +630,8 @@ timer_service::process_expired() remove_timer_impl(*t); while (auto* w = t->waiters_.pop_front()) { - w->impl_ = nullptr; - w->ec_value_ = {}; + w->impl_ = nullptr; + w->ec_ = {}; expired.push_back(w); } t->might_have_pending_waits_.store( @@ -835,79 +725,9 @@ timer_service::swap_heap(std::size_t i1, std::size_t i2) heap_[i2].timer_->heap_index_.store(i2, std::memory_order_relaxed); } -// waiter_node out-of-class member function definitions - -inline void -waiter_node::canceller::operator()() const -{ - waiter_->svc_->cancel_waiter(waiter_); -} - -inline void -waiter_node::completion_op::do_complete( - [[maybe_unused]] void* owner, - scheduler_op* base, - std::uint32_t, - std::uint32_t) -{ - // owner is always non-null here. The destroy path (owner == nullptr) - // is unreachable because completion_op overrides destroy() directly, - // bypassing scheduler_op::destroy() which would call func_(nullptr, ...). - BOOST_COROSIO_ASSERT(owner); - static_cast(base)->operator()(); -} - -inline void -waiter_node::completion_op::operator()() -{ - auto* w = waiter_; - w->stop_cb_.reset(); - if (w->ec_out_) - *w->ec_out_ = w->ec_value_; - - auto* cont = w->cont_; - auto d = w->d_; - auto* svc = w->svc_; - auto& sched = svc->get_scheduler(); - - svc->destroy_waiter(w); - - d.post(*cont); - sched.work_finished(); -} - -// GCC 14 false-positive: inlining ~optional through -// delete loses track that stop_cb_ was already .reset() above. -#if defined(__GNUC__) && !defined(__clang__) -#pragma GCC diagnostic push -#pragma GCC diagnostic ignored "-Wmaybe-uninitialized" -#endif -inline void -waiter_node::completion_op::destroy() -{ - // Called during scheduler shutdown drain when this completion_op is - // in the scheduler's ready queue (posted by cancel_timer() or - // process_expired()). Balances the work_started() from - // implementation::wait(). The scheduler drain loop separately - // balances the work_started() from post(). On IOCP both decrements - // are required for outstanding_work_ to reach zero; on other - // backends this is harmless. - // - // This override also prevents scheduler_op::destroy() from calling - // do_complete(nullptr, ...). See also: timer_service::shutdown() - // which drains waiters still in the timer heap (the other path). - auto* w = waiter_; - w->stop_cb_.reset(); - auto h = std::exchange(w->h_, {}); - auto& sched = w->svc_->get_scheduler(); - delete w; - sched.work_finished(); - if (h) - h.destroy(); -} -#if defined(__GNUC__) && !defined(__clang__) -#pragma GCC diagnostic pop -#endif +// waiter_node's completion_op and canceller members are defined in +// timer.cpp alongside implementation::wait(), for the same reason +// wait() lives there (see below). // timer::implementation::wait() is defined in timer.cpp, not here. // It must be a non-inline definition in a translation unit that is diff --git a/src/corosio/src/timer.cpp b/src/corosio/src/timer.cpp index 436638f9d..0cb3acfd4 100644 --- a/src/corosio/src/timer.cpp +++ b/src/corosio/src/timer.cpp @@ -59,25 +59,16 @@ timer::do_update_expiry() // wherever a detail::timer is used (every such user also needs timer's // constructors, defined in this same translation unit). std::coroutine_handle<> -timer::implementation::wait( - std::coroutine_handle<> h, - capy::executor_ref d, - std::stop_token token, - std::error_code* ec, - capy::continuation* cont) +timer::implementation::wait(waiter_node& w) { - // Already-expired fast path — no waiter_node, no mutex. + // Already-expired fast path — no publication, no mutex. // Post instead of dispatch so the coroutine yields to the // scheduler, allowing other queued work to run. - if (heap_index_.load(std::memory_order_relaxed) == npos) + if (already_expired()) { - if (expiry_ == (time_point::min)() || expiry_ <= clock_type::now()) - { - if (ec) - *ec = {}; - d.post(*cont); - return std::noop_coroutine(); - } + w.ec_ = {}; + w.d_.post(w.cont_); + return std::noop_coroutine(); } // Publication-last invariant: fully initialize the waiter, count @@ -88,24 +79,82 @@ timer::implementation::wait( // null impl_ and is a safe no-op. To avoid losing such an early // cancel, insert_waiter() re-checks stop_requested() under the lock // and completes as canceled if it fires in this window. - auto* w = svc_->create_waiter(); - w->impl_ = nullptr; - w->svc_ = svc_; - w->h_ = h; - w->cont_ = cont; - w->d_ = d; - w->token_ = std::move(token); - w->ec_out_ = ec; + w.impl_ = nullptr; + w.svc_ = svc_; might_have_pending_waits_.store(true, std::memory_order_relaxed); svc_->get_scheduler().work_started(); - if (w->token_.stop_possible()) - w->stop_cb_.emplace(w->token_, waiter_node::canceller{w}); + if (w.token_->stop_possible()) + w.arm_stop_cb(); - svc_->insert_waiter(*this, w); + svc_->insert_waiter(*this, &w); return std::noop_coroutine(); } +// completion_op and canceller definitions live here, non-inline, for +// the same reason wait() does: the inline waiter_node constructor in +// timer.hpp references do_complete and the vtable from translation +// units that never include timer_service.hpp. + +void +waiter_node::canceller::operator()() const +{ + waiter_->svc_->cancel_waiter(waiter_); +} + +void +waiter_node::completion_op::do_complete( + [[maybe_unused]] void* owner, + scheduler_op* base, + std::uint32_t, + std::uint32_t) +{ + // owner is always non-null here. The destroy path (owner == nullptr) + // is unreachable because completion_op overrides destroy() directly, + // bypassing scheduler_op::destroy() which would call func_(nullptr, ...). + BOOST_COROSIO_ASSERT(owner); + static_cast(base)->operator()(); +} + +void +waiter_node::completion_op::operator()() +{ + // The node lives in the resuming coroutine's frame: posting the + // continuation is the last access, since the frame (and node) + // may complete and die on another thread immediately after. + auto* w = waiter_; + w->reset_stop_cb(); + auto d = w->d_; + auto& sched = w->svc_->get_scheduler(); + d.post(w->cont_); + sched.work_finished(); +} + +void +waiter_node::completion_op::destroy() +{ + // Called during scheduler shutdown drain when this completion_op is + // in the scheduler's ready queue (posted by cancel_timer() or + // process_expired()). Balances the work_started() from + // implementation::wait(). The scheduler drain loop separately + // balances the work_started() from post(). On IOCP both decrements + // are required for outstanding_work_ to reach zero; on other + // backends this is harmless. + // + // This override also prevents scheduler_op::destroy() from calling + // do_complete(nullptr, ...). See also: timer_service::shutdown() + // which drains waiters still in the timer heap (the other path). + // Destroying the frame also ends the node's storage, so it is + // the last access. + auto* w = waiter_; + w->reset_stop_cb(); + auto h = std::exchange(w->h_, {}); + auto& sched = w->svc_->get_scheduler(); + sched.work_finished(); + if (h) + h.destroy(); +} + } // namespace boost::corosio::detail diff --git a/test/unit/delay.cpp b/test/unit/delay.cpp index b3c5ae271..d7b733962 100644 --- a/test/unit/delay.cpp +++ b/test/unit/delay.cpp @@ -341,6 +341,38 @@ struct delay_test BOOST_TEST_EQ(destroyed, 3); } + void testShutdownDrainsPostedCompletion() + { + // A cancelled delay posts its completion op to the scheduler + // ready queue; the io_context destructs before running it, so + // the scheduler drain must destroy the suspended frame through + // the queued op rather than the timer service's heap drain. + int destroyed = 0; + std::stop_source src; + + { + io_context ioc(Backend); + + auto task = [](int& counter) -> capy::task<> { + struct guard + { + int& c_; + ~guard() { ++c_; } + }; + guard g{counter}; + auto [ec] = co_await delay(std::chrono::hours(1)); + (void)ec; + }; + + capy::run_async(ioc.get_executor(), src.get_token())( + task(destroyed)); + ioc.poll(); + src.request_stop(); + } + + BOOST_TEST_EQ(destroyed, 1); + } + void testNarrowRepDurationClamp() { // A narrow-rep duration must survive the overflow clamp intact. @@ -565,6 +597,7 @@ struct delay_test testConcurrentDelaysHeapRemoval(); testShutdownWithSuspendedDelay(); testShutdownDrainsHeapWaiters(); + testShutdownDrainsPostedCompletion(); testNarrowRepDurationClamp(); testNegativeExtremeDurationCompletes(); testPositiveExtremeDurationArmsThenCancels();