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();