From 7c66b284c02a823de8aa5e398f39d4782dce411a Mon Sep 17 00:00:00 2001 From: Steve Gerbino Date: Tue, 4 Aug 2026 23:02:37 +0200 Subject: [PATCH 1/3] feat: support custom clocks and wait traits in delay Add a delay overload accepting a time point on any clock, with a wait-traits policy controlling how much of the remaining time each underlying steady wait may cover: co_await delay(system_clock::now() + 5min); co_await delay(deadline); The wait is zero-alloc: a sequence of re-armed steady-clock waits on one frame-embedded waiter. waiter_node gains a re-arm hook consulted by the completion op before resuming, and clock_delay_awaitable re-publishes its waiter with the next traits-capped expiry until Clock::now() reaches the deadline. Publication goes through a new timer::publish_wait entry point with no elapsed fast path, so every completion (including cancellation) reaches the hook; re-arms go through timer::rearm_wait, which preserves the wait's work count and stop callback. Steady-clock time points, of any duration type, keep routing to the existing single-wait awaitable. The new public header wait_traits.hpp provides the identity default traits and the WaitTraits concept. The timers guide gains a section on delaying against a different clock. --- doc/modules/ROOT/pages/4.guide/4h.timers.adoc | 25 ++ include/boost/corosio/delay.hpp | 252 +++++++++++-- include/boost/corosio/detail/timer.hpp | 95 ++++- .../boost/corosio/detail/timer_service.hpp | 8 + include/boost/corosio/wait_traits.hpp | 90 +++++ src/corosio/src/timer.cpp | 41 +++ test/doc/snippets/4h_timers.cpp | 38 ++ test/unit/delay.cpp | 336 ++++++++++++++++++ test/unit/wait_traits.cpp | 80 +++++ 9 files changed, 934 insertions(+), 31 deletions(-) create mode 100644 include/boost/corosio/wait_traits.hpp create mode 100644 test/unit/wait_traits.cpp diff --git a/doc/modules/ROOT/pages/4.guide/4h.timers.adoc b/doc/modules/ROOT/pages/4.guide/4h.timers.adoc index e2c3bddf1..904aadb87 100644 --- a/doc/modules/ROOT/pages/4.guide/4h.timers.adoc +++ b/doc/modules/ROOT/pages/4.guide/4h.timers.adoc @@ -55,6 +55,31 @@ include::example$snippets/4h_timers.cpp[tag=delay_timepoint,indent=0] A time point already in the past also completes synchronously. +== Delaying on a Different Clock + +`delay(time_point)` also accepts a time point on clocks other than +`steady_clock` -- for example `std::chrono::system_clock`, when a +deadline is naturally expressed as wall-clock time rather than a +monotonic duration: + +[source,cpp] +---- +include::example$snippets/4h_timers.cpp[tag=delay_wallclock,indent=0] +---- + +Internally this performs one or more bounded `steady_clock` waits, +re-reading the clock between them, so an adjustment to the clock is +observed at the next re-check rather than only at the original +deadline. Pass a custom traits type as `delay(time_point)` to +bound how quickly such an adjustment is observed: + +[source,cpp] +---- +include::example$snippets/4h_timers.cpp[tag=delay_traits,indent=0] + +include::example$snippets/4h_timers.cpp[tag=delay_traits_use,indent=0] +---- + == Cancellation `delay()` honors the stop token of its `co_await` environment. If the diff --git a/include/boost/corosio/delay.hpp b/include/boost/corosio/delay.hpp index c1d2ce4b8..c53158aa0 100644 --- a/include/boost/corosio/delay.hpp +++ b/include/boost/corosio/delay.hpp @@ -13,18 +13,70 @@ #include #include #include +#include #include #include #include #include +#include #include #include #include #include +#include +#include namespace boost::corosio { +namespace detail { + +// Narrow reps wrap if nanoseconds::max() is converted into them; +// a double comparison clamps safely in both directions. +template +std::chrono::nanoseconds +clamp_to_ns(std::chrono::duration dur) noexcept +{ + using namespace std::chrono; + using dsec = duration; + if constexpr (std::is_floating_point_v) + { + // NaN fails both clamp comparisons and would reach the + // cast; treat it as no wait rather than undefined behavior. + if (dur != dur) + return nanoseconds::zero(); + } + return dsec(dur) >= dsec((nanoseconds::max)()) + ? (nanoseconds::max)() + : dsec(dur) <= dsec((nanoseconds::min)()) + ? (nanoseconds::min)() + : duration_cast(dur); +} + +// A non-io_context executor cannot supply a timer service, and +// await_suspend is driven through a noexcept wrapper, so translate +// the service-lookup failure into a clear terminate. +inline void +emplace_delay_timer( + std::optional& t, capy::execution_context& ctx) +{ + try + { + t.emplace(ctx); + } + catch(std::logic_error const&) + { + throw_logic_error( + "delay requires an io_context-backed executor"); + } + catch(std::exception const& e) + { + throw_logic_error(e.what()); + } +} + +} // namespace detail + /** IoAwaitable returned by @ref delay. Suspends the calling coroutine until the deadline elapses or @@ -118,22 +170,7 @@ class delay_awaitable dur_.count() <= 0) return h; - // A non-io_context executor cannot supply a timer service, - // and await_suspend is driven through a noexcept wrapper, so - // translate the service-lookup failure into a clear terminate. - try - { - timer_.emplace(env->executor.context()); - } - catch(std::logic_error const&) - { - detail::throw_logic_error( - "delay requires an io_context-backed executor"); - } - catch(std::exception const& e) - { - detail::throw_logic_error(e.what()); - } + detail::emplace_delay_timer(timer_, env->executor.context()); if(has_deadline_) timer_->expires_at(deadline_); @@ -155,6 +192,136 @@ class delay_awaitable } }; +/** IoAwaitable returned by the clock overloads of @ref delay. + + Suspends the calling coroutine until `Clock::now()` reaches the + deadline or the environment's stop token is activated. The wait + is a sequence of steady-clock timer waits: after each expiry the + clock is re-read and, if the deadline is unreached, the same + frame-embedded waiter is re-published for the next + `Traits::to_wait_duration` cap — without resuming the coroutine + and without allocating. + + Not intended to be named directly; use the @ref delay factory + overloads instead. + + @par Preconditions + The awaiting coroutine's executor must belong to an + `io_context`. Any other execution context terminates with a + diagnostic, because silently running without a timer would + drop the requested delay. + + @par Cancellation + Identical to @ref delay_awaitable: stop already requested + resumes inline with `error::canceled`; stop while suspended + cancels the pending wait, including between re-arms. + + @see delay, wait_traits +*/ +template +class clock_delay_awaitable +{ + typename Clock::time_point deadline_{}; + bool canceled_ = false; + std::optional timer_; + detail::waiter_node w_; + + std::chrono::nanoseconds + next_wait(typename Clock::time_point now) const noexcept + { + return detail::clamp_to_ns( + Traits::to_wait_duration(deadline_ - now)); + } + + // Runs on the scheduler thread executing the completion op, + // before the continuation is posted, so the frame cannot die + // concurrently. + static bool on_fire(void* ctx) noexcept + { + auto* self = static_cast(ctx); + // Canceled: resume and surface the error + if(self->w_.ec_) + return false; + auto now = Clock::now(); + if(now >= self->deadline_) + return false; + // Re-publish and return without touching the node again: + // the wait may complete on another thread immediately after. + if(self->timer_->rearm_wait(self->w_, self->next_wait(now))) + return true; + // Heap growth failed; finish the wait with an error rather + // than strand the frame with an unbalanced work count. + self->w_.ec_ = std::make_error_code(std::errc::not_enough_memory); + return false; + } + +public: + /// Construct an awaitable that waits until `tp` on `Clock`. + explicit clock_delay_awaitable( + typename Clock::time_point tp) noexcept + : deadline_(tp) + { + } + + /// Construct by transferring the deadline from `other`. + // Only moved before await_suspend; w_ is quiescent until then. + clock_delay_awaitable(clock_delay_awaitable&& other) noexcept + : deadline_(other.deadline_) + { + } + + clock_delay_awaitable(clock_delay_awaitable const&) = delete; + clock_delay_awaitable& + operator=(clock_delay_awaitable const&) = delete; + clock_delay_awaitable& + operator=(clock_delay_awaitable&&) = delete; + + /// Return false unconditionally; see await_suspend. + // The elapsed-deadline fast path must run after the stop-token + // check, and only await_suspend receives the env carrying it. + bool await_ready() const noexcept + { + return false; + } + + /// Resume inline if stopped or reached; else wait on a timer. + std::coroutine_handle<> + await_suspend(std::coroutine_handle<> h, capy::io_env const* env) + { + if(env->stop_token.stop_requested()) + { + canceled_ = true; + return h; + } + + auto now = Clock::now(); + if(now >= deadline_) + return h; + + detail::emplace_delay_timer(timer_, env->executor.context()); + + timer_->expires_after(next_wait(now)); + + w_.bind(h, *env); + w_.on_fire_ = &on_fire; + w_.on_fire_ctx_ = this; + // Never the elapsed fast path: a capped expiry that elapses + // before publication must still reach on_fire, not complete + // the clock wait early. + return timer_->publish_wait(w_); + } + + /// Return empty on deadline, `error::canceled` if stop won. + capy::io_result<> await_resume() noexcept + { + if(canceled_) + return {capy::error::canceled}; + if(timer_) + return {w_.ec_}; + return {}; + } +}; + /** Suspend the current coroutine for a duration. Returns an IoAwaitable that completes at or after the @@ -175,16 +342,7 @@ template [[nodiscard]] delay_awaitable delay(std::chrono::duration dur) noexcept { - using namespace std::chrono; - // Narrow reps wrap if nanoseconds::max() is converted into them; - // a double comparison clamps safely in both directions. - using dsec = duration; - auto ns = dsec(dur) >= dsec((nanoseconds::max)()) - ? (nanoseconds::max)() - : dsec(dur) <= dsec((nanoseconds::min)()) - ? (nanoseconds::min)() - : duration_cast(dur); - return delay_awaitable(ns); + return delay_awaitable(detail::clamp_to_ns(dur)); } /** Suspend the current coroutine until a time point. @@ -203,6 +361,48 @@ delay(std::chrono::steady_clock::time_point tp) noexcept return delay_awaitable(tp); } +/** Suspend the current coroutine until a time point on `Clock`. + + Returns an IoAwaitable that completes at or after the first + observation of `Clock::now() >= tp`, or earlier if the + environment's stop token is activated. The wait is one or more + bounded steady-clock waits, re-reading `Clock::now()` after + each; `Traits::to_wait_duration` bounds each one. With the + default @ref wait_traits a single full-length wait is used, so + an adjustment of `Clock` mid-wait is observed only at natural + wakeup; supply capping traits to bound that latency. Time + points already reached complete synchronously. + + @note `Clock::now()` and `Traits::to_wait_duration` are invoked + on the io_context's run thread and must not throw or block. + + @par Example + @code + auto [ec] = co_await delay( + std::chrono::system_clock::now() + std::chrono::minutes(5)); + @endcode + + @tparam Traits The wait-traits policy; `void` selects + @ref wait_traits. + + @param tp The time point to wait until. + + @return A @ref clock_delay_awaitable yielding `io_result<>`. +*/ +template + requires (!std::same_as) && + (std::is_void_v || WaitTraits) +[[nodiscard]] auto +delay(std::chrono::time_point tp) noexcept +{ + using traits_type = std::conditional_t< + std::is_void_v, wait_traits, Traits>; + // ceil preserves completes-at-or-after when Duration is coarser + // than the clock's native duration + return clock_delay_awaitable( + std::chrono::ceil(tp)); +} + } // namespace boost::corosio #endif diff --git a/include/boost/corosio/detail/timer.hpp b/include/boost/corosio/detail/timer.hpp index a8e835fb0..062c2c542 100644 --- a/include/boost/corosio/detail/timer.hpp +++ b/include/boost/corosio/detail/timer.hpp @@ -157,6 +157,20 @@ class BOOST_COROSIO_DECL timer : public io_object // symbol from outside the corosio DLL. BOOST_COROSIO_DECL std::coroutine_handle<> wait(waiter_node& w); + + /** Publish a waiter unconditionally. + + Like `wait`, but never takes the elapsed fast path. The + fast path posts the continuation directly, bypassing the + embedded op; hook-driven waits must observe every + completion through the op, where the re-arm hook runs. + + @par Preconditions + Same as `wait`. + + @param w The waiter to publish. + */ + std::coroutine_handle<> publish(waiter_node& w); }; /// The clock type used for time operations. @@ -409,6 +423,48 @@ class BOOST_COROSIO_DECL timer : public io_object // Defined below wait_awaitable, which needs timer complete. wait_awaitable wait(); + /** Publish a hook-driven wait. + + Bypasses the elapsed fast path so every completion is + delivered through the waiter's embedded op, where the + re-arm hook is consulted. Used by awaitables that + re-publish the waiter to continue a logical wait across + several timer expirations. + + @par Preconditions + @p w is fully initialized ( handle, executor, stop token, + hook fields ) and its storage outlives the wait. + + @param w The waiter to publish. + + @return `std::noop_coroutine()`. + */ + std::coroutine_handle<> publish_wait(waiter_node& w); + + /** Re-arm an already-fired waiter with a new relative expiry. + + Stores the ( saturated ) expiry and re-publishes @p w. The + waiter's original work count and stop callback remain in + effect. Must only be called from the waiter's re-arm hook, + where the waiter has been popped from the service but not + yet resumed. + + @par Preconditions + The timer has no other waiters — this is what makes the + unlocked expiry write race-free. + + Re-publication needs heap capacity and can fail under + allocation pressure. On failure the waiter is left exactly as + the hook received it, so the caller completes the wait through + the normal resume path instead of re-arming. + + @param w The waiter to re-publish. + @param d The next expiry relative to now. + + @return `true` if re-published; `false` if allocation failed. + */ + [[nodiscard]] bool rearm_wait(waiter_node& w, duration d) noexcept; + protected: explicit timer(handle h) noexcept : io_object(std::move(h)) {} @@ -497,6 +553,18 @@ struct BOOST_COROSIO_SYMBOL_VISIBLE waiter_node /// The completion result read by `await_resume`. std::error_code ec_; + // Consulted by the completion op before resuming; lets a + // clock-facade wait re-publish itself instead of completing. + // Never consulted on the shutdown destroy path. Consulted on + // every completion, including cancellation ( `ec_` set ) — the + // hook must inspect `w`'s `ec_` and must not re-arm a canceled + // waiter. Runs inside the completion path; must not throw. + /// Re-arm hook: return true to skip resumption ( wait continues ). + bool (*on_fire_)(void*) noexcept = nullptr; + + /// Context passed to `on_fire_` ( the owning awaitable ). + void* on_fire_ctx_ = nullptr; + /// The embedded completion op posted to the scheduler. completion_op op_; @@ -519,6 +587,24 @@ struct BOOST_COROSIO_SYMBOL_VISIBLE waiter_node waiter_node(waiter_node const&) = delete; waiter_node& operator=(waiter_node const&) = delete; + /** Bind the coroutine and its environment before publication. + + The single definition of the fields every wait must populate + before the node is published; hook-driven waits additionally + set `on_fire_` / `on_fire_ctx_`. + + @param h The coroutine to resume on completion. + @param env The awaiting chain's environment; must outlive + the suspension. + */ + void bind(std::coroutine_handle<> h, capy::io_env const& env) noexcept + { + h_ = h; + cont_.h = h; + d_ = env.executor; + token_ = &env.stop_token; + } + /** Arm the stop callback. @par Preconditions @@ -579,11 +665,11 @@ struct wait_awaitable -> std::coroutine_handle<> { auto& impl = t_.get(); - w_.h_ = h; - w_.cont_.h = h; - w_.d_ = env->executor; + w_.bind(h, *env); - // Inline fast path: already expired and not in the heap + // Inline fast path: already expired and not in the heap. + // Post instead of dispatch so the coroutine yields to the + // scheduler, allowing other queued work to run. if (impl.already_expired()) { w_.ec_ = {}; @@ -591,7 +677,6 @@ struct wait_awaitable return std::noop_coroutine(); } - w_.token_ = &env->stop_token; return impl.wait(w_); } }; diff --git a/include/boost/corosio/detail/timer_service.hpp b/include/boost/corosio/detail/timer_service.hpp index f6c0d1ac2..e45afed4c 100644 --- a/include/boost/corosio/detail/timer_service.hpp +++ b/include/boost/corosio/detail/timer_service.hpp @@ -479,6 +479,14 @@ timer_service::insert_waiter(timer::implementation& impl, waiter_node* w) bool lost_cancel = false; { std::lock_guard lock(mutex_); + // Grow before publishing anything, so the push_back below + // cannot throw: a failure here leaves the waiter untouched, + // the strong guarantee rearm_wait's recovery relies on. + if (impl.heap_index_.load(std::memory_order_relaxed) == + (std::numeric_limits::max)() && + heap_.size() == heap_.capacity()) + heap_.reserve( + heap_.capacity() == 0 ? 16 : 2 * heap_.capacity()); // Publish: from here the waiter is visible to the fire path and // to its own stop callback (impl_ non-null enables cancel_waiter). w->impl_ = &impl; diff --git a/include/boost/corosio/wait_traits.hpp b/include/boost/corosio/wait_traits.hpp new file mode 100644 index 000000000..d6e929f01 --- /dev/null +++ b/include/boost/corosio/wait_traits.hpp @@ -0,0 +1,90 @@ +// +// Copyright (c) 2026 Steve Gerbino +// +// Distributed under the Boost Software License, Version 1.0. (See accompanying +// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt) +// +// Official repository: https://github.com/cppalliance/corosio +// + +#ifndef BOOST_COROSIO_WAIT_TRAITS_HPP +#define BOOST_COROSIO_WAIT_TRAITS_HPP + +#include + +#include + +namespace boost::corosio { + +/** Default wait traits for clock-based delays. + + Controls how much of the remaining time a single underlying + steady-clock wait may cover before `Clock::now()` is re-read. + A larger value costs fewer wakeups; a smaller value bounds how + late an adjustment of `Clock` ( e.g. a stepped time-of-day + clock ) is observed. The default covers the full remaining + duration, which is exact for clocks that advance in lockstep + with the machine's monotonic clock. + + @par Example + @code + // Observe wall-clock steps within one second + struct capped_traits + { + static std::chrono::system_clock::duration + to_wait_duration(std::chrono::system_clock::duration d) + { + return (std::min)(d, + std::chrono::system_clock::duration( + std::chrono::seconds(1))); + } + }; + + auto [ec] = co_await delay( + std::chrono::system_clock::now() + std::chrono::hours(1)); + @endcode + + @tparam Clock The clock type whose durations are converted. + + @see delay +*/ +template +struct wait_traits +{ + /** Convert a remaining duration into a wait duration. + + Should return a positive duration when @p d is positive; a + non-positive result degrades to reactor-rate re-checking. + + @par Preconditions + Must not throw and must not block — invoked on the + io_context's run thread, including from the timer + completion path. + + @param d The remaining time until the deadline. + + @return The duration the next underlying wait may cover. + */ + static typename Clock::duration + to_wait_duration(typename Clock::duration d) + { + return d; + } +}; + +/** Concept for wait-traits policies usable with `Clock`. + + Satisfied when `Traits::to_wait_duration` accepts a + `Clock::duration` and returns something convertible back to it. + `Traits::to_wait_duration` must not throw. +*/ +template +concept WaitTraits = requires(typename Clock::duration d) +{ + { Traits::to_wait_duration(d) } + -> std::convertible_to; +}; + +} // namespace boost::corosio + +#endif diff --git a/src/corosio/src/timer.cpp b/src/corosio/src/timer.cpp index 0cb3acfd4..e4c05aa6f 100644 --- a/src/corosio/src/timer.cpp +++ b/src/corosio/src/timer.cpp @@ -70,7 +70,12 @@ timer::implementation::wait(waiter_node& w) w.d_.post(w.cont_); return std::noop_coroutine(); } + return publish(w); +} +std::coroutine_handle<> +timer::implementation::publish(waiter_node& w) +{ // Publication-last invariant: fully initialize the waiter, count // its work, and arm cancellation BEFORE insert_waiter() publishes // it into the heap/list where a concurrent run() thread can fire @@ -93,6 +98,36 @@ timer::implementation::wait(waiter_node& w) return std::noop_coroutine(); } +std::coroutine_handle<> +timer::publish_wait(waiter_node& w) +{ + return get().publish(w); +} + +bool +timer::rearm_wait(waiter_node& w, duration d) noexcept +{ + // The single waiter was popped before its op ran, so the impl is + // out of the heap with no published waiters: expires_after only + // stores the saturated expiry, and writing it is race-free. + expires_after(d); + auto& impl = get(); + // The drain that popped the waiter cleared the flag. + impl.might_have_pending_waits_.store(true, std::memory_order_relaxed); + try + { + impl.svc_->insert_waiter(impl, &w); + } + catch(std::bad_alloc const&) + { + // insert_waiter grows the heap before publishing anything, + // so the waiter is untouched and the caller can complete + // the wait through the normal resume path. + return false; + } + return true; +} + // 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 @@ -125,6 +160,12 @@ waiter_node::completion_op::operator()() // continuation is the last access, since the frame (and node) // may complete and die on another thread immediately after. auto* w = waiter_; + // A true return means the waiter re-published itself: the frame + // stays suspended, the wait's work count stays live, and the + // node may already be firing on another thread — no access past + // this point. + if (w->on_fire_ && w->on_fire_(w->on_fire_ctx_)) + return; w->reset_stop_cb(); auto d = w->d_; auto& sched = w->svc_->get_scheduler(); diff --git a/test/doc/snippets/4h_timers.cpp b/test/doc/snippets/4h_timers.cpp index bea1d5be0..557d1ac9d 100644 --- a/test/doc/snippets/4h_timers.cpp +++ b/test/doc/snippets/4h_timers.cpp @@ -85,6 +85,44 @@ capy::task<> delay_timepoint_frag() // end::delay_timepoint[] } +// This fragment waits on a real 5-minute wall-clock deadline, so +// (like the connect fragments below) it is compiled but never +// launched. +capy::task<> delay_wallclock_frag() +{ + // tag::delay_wallclock[] + auto deadline = std::chrono::system_clock::now() + + std::chrono::minutes(5); + auto [ec] = co_await corosio::delay(deadline); + // end::delay_wallclock[] +} + +// tag::delay_traits[] +// Re-read the wall clock at least once per second, so a step of +// the clock is observed within that bound +struct capped_traits +{ + static std::chrono::system_clock::duration + to_wait_duration(std::chrono::system_clock::duration d) + { + auto cap = std::chrono::system_clock::duration( + std::chrono::seconds(1)); + return d < cap ? d : cap; + } +}; +// end::delay_traits[] + +// Same real deadline as the wallclock fragment above: compiled but +// never launched. +capy::task<> delay_traits_frag() +{ + // tag::delay_traits_use[] + auto deadline = std::chrono::system_clock::now() + + std::chrono::minutes(5); + auto [ec] = co_await corosio::delay(deadline); + // end::delay_traits_use[] +} + capy::task<> delay_cancel_frag(std::error_code& out) { // tag::delay_cancel[] diff --git a/test/unit/delay.cpp b/test/unit/delay.cpp index d7b733962..ef0176212 100644 --- a/test/unit/delay.cpp +++ b/test/unit/delay.cpp @@ -20,6 +20,7 @@ #include #include #include +#include #include #include #include @@ -31,6 +32,87 @@ namespace boost::corosio { +// Test clock driven by a static counter so facade waits complete +// after a deterministic number of re-check iterations with no +// wall-clock dependence. +struct test_clock +{ + using rep = std::int64_t; + using period = std::nano; + using duration = std::chrono::nanoseconds; + using time_point = std::chrono::time_point; + static constexpr bool is_steady = false; + + static inline std::atomic now_ns{0}; + + static time_point now() noexcept + { + return time_point( + duration(now_ns.load(std::memory_order_relaxed))); + } +}; + +// Advances test_clock 1ms per consultation and requests a zero-length +// steady wait, so each re-check lands on the next reactor pass. +struct stepping_traits +{ + static inline std::atomic calls{0}; + + static test_clock::duration + to_wait_duration(test_clock::duration) + { + calls.fetch_add(1, std::memory_order_relaxed); + test_clock::now_ns.fetch_add( + 1'000'000, std::memory_order_relaxed); + return {}; + } +}; + +// Requests a wait far longer than any test runs, so cancellation is +// the only way out. +struct hold_traits +{ + static test_clock::duration + to_wait_duration(test_clock::duration) + { + return std::chrono::seconds(10); + } +}; + +// Distinct clock type backed by the monotonic clock: routes to the +// facade ( not the steady fast path ) but advances in real time, +// exercising default traits end to end. +struct wall_clock +{ + using rep = std::chrono::steady_clock::rep; + using period = std::chrono::steady_clock::period; + using duration = std::chrono::steady_clock::duration; + using time_point = std::chrono::time_point; + static constexpr bool is_steady = true; + + static time_point now() noexcept + { + return time_point( + std::chrono::steady_clock::now().time_since_epoch()); + } +}; + +// Steady time points, of any duration, must keep the zero-iteration +// awaitable; other clocks route to the facade. +static_assert(std::same_as< + decltype(delay(std::chrono::steady_clock::time_point{})), + delay_awaitable>); +static_assert(std::same_as< + decltype(delay(std::chrono::time_point{})), + delay_awaitable>); +static_assert(std::same_as< + decltype(delay(test_clock::time_point{})), + clock_delay_awaitable>>); +static_assert(std::same_as< + decltype(delay(test_clock::time_point{})), + clock_delay_awaitable>); + template struct delay_test { @@ -418,6 +500,24 @@ struct delay_test BOOST_TEST(ok); } + void testFloatingNaNDurationCompletes() + { + // A NaN floating-rep duration must not reach duration_cast; + // the clamp treats it as no wait and completes synchronously. + io_context ioc(Backend); + bool ok = false; + + auto t = [](bool& ok_out) -> capy::task<> { + auto [ec] = co_await delay(std::chrono::duration( + std::numeric_limits::quiet_NaN())); + ok_out = !ec; + }; + capy::run_async(ioc.get_executor())(t(ok)); + + ioc.run(); + BOOST_TEST(ok); + } + void testPositiveExtremeDurationArmsThenCancels() { // hours::max() must clamp and arm a real timer without overflow @@ -580,6 +680,232 @@ struct delay_test } } + void testClockDeadlineCompletes() + { + io_context ioc(Backend); + test_clock::now_ns.store(0); + stepping_traits::calls.store(0); + bool ok = false; + + auto t = [](bool& ok_out) -> capy::task<> { + auto tp = test_clock::now() + std::chrono::milliseconds(5); + auto [ec] = co_await delay(tp); + ok_out = !ec; + }; + capy::run_async(ioc.get_executor())(t(ok)); + + ioc.run(); + BOOST_TEST(ok); + // The re-check loop must have run: 5ms of clock at 1ms per + // consultation is at least four re-arms after the initial one. + BOOST_TEST(stepping_traits::calls.load() >= 4); + BOOST_TEST(test_clock::now() >= test_clock::time_point( + std::chrono::milliseconds(5))); + } + + void testClockPastDeadlineCompletesImmediately() + { + io_context ioc(Backend); + test_clock::now_ns.store(1'000'000'000); + stepping_traits::calls.store(0); + bool ok = false; + + auto t = [](bool& ok_out) -> capy::task<> { + auto tp = test_clock::now() - std::chrono::seconds(1); + auto [ec] = co_await delay(tp); + ok_out = !ec; + }; + capy::run_async(ioc.get_executor())(t(ok)); + + ioc.run(); + BOOST_TEST(ok); + // Elapsed deadline resumes inline without consulting traits + BOOST_TEST_EQ(stepping_traits::calls.load(), 0); + } + + void testClockDefaultTraitsCompletes() + { + io_context ioc(Backend); + bool ok = false; + + auto t = [](bool& ok_out) -> capy::task<> { + auto tp = wall_clock::now() + std::chrono::milliseconds(5); + auto [ec] = co_await delay(tp); + ok_out = !ec && wall_clock::now() >= tp; + }; + capy::run_async(ioc.get_executor())(t(ok)); + + ioc.run(); + BOOST_TEST(ok); + } + + void testClockCoarseDurationCompletes() + { + // A time_point coarser than Clock::duration must convert + // ( ceil ) and still complete at or after the deadline. + io_context ioc(Backend); + test_clock::now_ns.store(0); + bool ok = false; + + auto t = [](bool& ok_out) -> capy::task<> { + auto tp = std::chrono::time_point(std::chrono::milliseconds(3)); + auto [ec] = co_await delay(tp); + ok_out = !ec && test_clock::now() >= + test_clock::time_point(std::chrono::milliseconds(3)); + }; + capy::run_async(ioc.get_executor())(t(ok)); + + ioc.run(); + BOOST_TEST(ok); + } + + void testClockCancellation() + { + io_context ioc(Backend); + test_clock::now_ns.store(0); + std::stop_source src; + bool canceled = false; + + auto t = [](bool& canceled_out) -> capy::task<> { + auto tp = test_clock::now() + std::chrono::hours(1); + auto [ec] = co_await delay(tp); + canceled_out = (ec == capy::cond::canceled); + }; + capy::run_async(ioc.get_executor(), src.get_token())(t(canceled)); + + // Let the wait suspend, then cancel + ioc.run_one(); + src.request_stop(); + ioc.run(); + BOOST_TEST(canceled); + } + + void testClockAlreadyStoppedCompletesCanceled() + { + io_context ioc(Backend); + test_clock::now_ns.store(0); + std::stop_source src; + src.request_stop(); + bool canceled = false; + + auto t = [](bool& canceled_out) -> capy::task<> { + auto tp = test_clock::now() + std::chrono::hours(1); + auto [ec] = co_await delay(tp); + canceled_out = (ec == capy::cond::canceled); + }; + capy::run_async(ioc.get_executor(), src.get_token())(t(canceled)); + + ioc.run(); + BOOST_TEST(canceled); + } + + void testClockPastDeadlineWithStopRequested() + { + // Stop must win over an elapsed deadline, mirroring the + // steady overloads' ordering. + io_context ioc(Backend); + test_clock::now_ns.store(1'000'000'000); + std::stop_source src; + src.request_stop(); + bool canceled = false; + + auto t = [](bool& canceled_out) -> capy::task<> { + auto tp = test_clock::now() - std::chrono::seconds(1); + auto [ec] = co_await delay(tp); + canceled_out = (ec == capy::cond::canceled); + }; + capy::run_async(ioc.get_executor(), src.get_token())(t(canceled)); + + ioc.run(); + BOOST_TEST(canceled); + } + + void testClockShutdownWithSuspendedWait() + { + // Destroying the io_context while a facade wait is suspended + // must drain the waiter and destroy the frame ( guard runs ), + // like any pending steady wait. + int destroyed = 0; + test_clock::now_ns.store(0); + + { + io_context ioc(Backend); + + auto task = [](int& counter) -> capy::task<> { + struct guard + { + int& c_; + ~guard() { ++c_; } + }; + guard g{counter}; + auto tp = test_clock::now() + std::chrono::hours(1); + auto [ec] = co_await delay(tp); + (void)ec; + }; + + capy::run_async(ioc.get_executor())(task(destroyed)); + ioc.poll(); + } + + BOOST_TEST_EQ(destroyed, 1); + } + + void testClockRearmStopRace() + { + // Hammer the rearm/cancel interleavings: race_traits spins the + // facade at reactor rate while a foreign thread requests stop + // on every waiter. The clock still advances 1us per re-arm, so + // the test terminates even if every stop were lost. Every + // co_await must complete and the context must drain. + struct race_traits + { + static test_clock::duration + to_wait_duration(test_clock::duration) + { + test_clock::now_ns.fetch_add( + 1'000, std::memory_order_relaxed); + return {}; + } + }; + + constexpr int N = 100; + + for(int iter = 0; iter < 5; ++iter) + { + io_context ioc(Backend, 2u); // multi-threaded, not hint 1 + auto ex = ioc.get_executor(); + test_clock::now_ns.store(0); + + std::atomic completed{0}; + std::vector srcs(N); + + auto task = [](std::atomic& done) -> capy::task<> { + auto tp = test_clock::now() + + std::chrono::milliseconds(50); + auto [ec] = co_await delay(tp); + (void)ec; // success or canceled — both acceptable + done.fetch_add(1, std::memory_order_relaxed); + }; + + for(int i = 0; i < N; ++i) + capy::run_async(ex, srcs[i].get_token())(task(completed)); + + std::thread stopper([&] { + for(auto& s : srcs) + s.request_stop(); + }); + std::thread r1([&] { ioc.run(); }); + std::thread r2([&] { ioc.run(); }); + + r1.join(); + r2.join(); + stopper.join(); + + BOOST_TEST_EQ(completed.load(), N); + } + } + void run() { testDurationCompletes(); @@ -600,11 +926,21 @@ struct delay_test testShutdownDrainsPostedCompletion(); testNarrowRepDurationClamp(); testNegativeExtremeDurationCompletes(); + testFloatingNaNDurationCompletes(); testPositiveExtremeDurationArmsThenCancels(); testMultiTimerExpiryOrder(); testAbruptStopWithPendingDelays(); testShutdownReentrantThreeFrames(); testInitiationStopRace(); + testClockDeadlineCompletes(); + testClockPastDeadlineCompletesImmediately(); + testClockDefaultTraitsCompletes(); + testClockCoarseDurationCompletes(); + testClockCancellation(); + testClockAlreadyStoppedCompletesCanceled(); + testClockPastDeadlineWithStopRequested(); + testClockShutdownWithSuspendedWait(); + testClockRearmStopRace(); } }; diff --git a/test/unit/wait_traits.cpp b/test/unit/wait_traits.cpp new file mode 100644 index 000000000..cd49e8455 --- /dev/null +++ b/test/unit/wait_traits.cpp @@ -0,0 +1,80 @@ +// +// Copyright (c) 2026 Steve Gerbino +// +// Distributed under the Boost Software License, Version 1.0. (See accompanying +// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt) +// +// Official repository: https://github.com/cppalliance/corosio +// + +// Test that header file is self-contained. +#include + +#include +#include + +#include "test_suite.hpp" + +namespace boost::corosio { + +struct wait_traits_test +{ + struct minute_clock + { + using rep = int; + using period = std::ratio<60>; + using duration = std::chrono::duration; + using time_point = std::chrono::time_point; + static constexpr bool is_steady = false; + static time_point now() noexcept { return {}; } + }; + + struct capped + { + static std::chrono::system_clock::duration + to_wait_duration(std::chrono::system_clock::duration d) + { + return (std::min)(d, + std::chrono::system_clock::duration( + std::chrono::seconds(1))); + } + }; + + struct not_traits + { + }; + + void testDefaultIsIdentity() + { + auto d = std::chrono::system_clock::duration( + std::chrono::seconds(5)); + BOOST_TEST(wait_traits< + std::chrono::system_clock>::to_wait_duration(d) == d); + BOOST_TEST(wait_traits::to_wait_duration( + minute_clock::duration(3)) == minute_clock::duration(3)); + BOOST_TEST(wait_traits::to_wait_duration( + minute_clock::duration(-3)) == minute_clock::duration(-3)); + } + + void testConcept() + { + static_assert(WaitTraits< + wait_traits, + std::chrono::system_clock>); + static_assert(WaitTraits< + wait_traits, minute_clock>); + static_assert(WaitTraits); + static_assert(!WaitTraits); + static_assert(!WaitTraits); + } + + void run() + { + testDefaultIsIdentity(); + testConcept(); + } +}; + +TEST_SUITE(wait_traits_test, "boost.corosio.wait_traits"); + +} // namespace boost::corosio From c78b0406238dbc2d7b84346c3741ca025ea1cd8d Mon Sep 17 00:00:00 2001 From: Steve Gerbino Date: Wed, 5 Aug 2026 17:35:37 +0200 Subject: [PATCH 2/3] refactor(timer): drop the unreachable cancel and reschedule surface --- include/boost/corosio/detail/timer.hpp | 91 +++--------- .../boost/corosio/detail/timer_service.hpp | 131 +----------------- src/corosio/src/timer.cpp | 18 --- 3 files changed, 29 insertions(+), 211 deletions(-) diff --git a/include/boost/corosio/detail/timer.hpp b/include/boost/corosio/detail/timer.hpp index 062c2c542..a7473ecc2 100644 --- a/include/boost/corosio/detail/timer.hpp +++ b/include/boost/corosio/detail/timer.hpp @@ -52,10 +52,10 @@ struct wait_awaitable; awaitable types. The timer can be used to schedule operations to occur after a specified duration or at a specific time point. - Multiple coroutines may wait concurrently on the same timer. - When the timer expires, all waiters complete with success. When - the timer is cancelled, all waiters complete with an error that - compares equal to `capy::cond::canceled`. + Each timer carries at most one wait: `delay` and `timeout` own a + private timer per `co_await`. When the timer expires the waiter + completes with success; a cancelled wait completes with an error + that compares equal to `capy::cond::canceled`. Each timer operation participates in the affine awaitable protocol, ensuring coroutines resume on the correct executor. @@ -301,35 +301,6 @@ class BOOST_COROSIO_DECL timer : public io_object timer(timer const&) = delete; timer& operator=(timer const&) = delete; - /** Cancel all pending asynchronous wait operations. - - All outstanding operations complete with an error code that - compares equal to `capy::cond::canceled`. - - @return The number of operations that were cancelled. - */ - std::size_t cancel() - { - if (!get().might_have_pending_waits_.load(std::memory_order_relaxed)) - return 0; - return do_cancel(); - } - - /** Cancel one pending asynchronous wait operation. - - The oldest pending wait is cancelled (FIFO order). It - completes with an error code that compares equal to - `capy::cond::canceled`. - - @return The number of operations that were cancelled (0 or 1). - */ - std::size_t cancel_one() - { - if (!get().might_have_pending_waits_.load(std::memory_order_relaxed)) - return 0; - return do_cancel_one(); - } - /** Return the timer's expiry time as an absolute time. @return The expiry time point. If no expiry has been set, @@ -342,34 +313,33 @@ class BOOST_COROSIO_DECL timer : public io_object /** Set the timer's expiry time as an absolute time. - Any pending asynchronous wait operations will be cancelled. + @par Preconditions + No wait is published on this timer. @param t The expiry time to be used for the timer. - - @return The number of pending operations that were cancelled. */ - std::size_t expires_at(time_point t) + void expires_at(time_point t) { - auto& impl = get(); + auto& impl = get(); + BOOST_COROSIO_ASSERT( + impl.heap_index_.load(std::memory_order_relaxed) == + implementation::npos); impl.expiry_ = t; - if (impl.heap_index_.load(std::memory_order_relaxed) == - implementation::npos && - !impl.might_have_pending_waits_.load(std::memory_order_relaxed)) - return 0; - return do_update_expiry(); } /** Set the timer's expiry time relative to now. - Any pending asynchronous wait operations will be cancelled. + @par Preconditions + No wait is published on this timer. @param d The expiry time relative to now. - - @return The number of pending operations that were cancelled. */ - std::size_t expires_after(duration d) + void expires_after(duration d) { auto& impl = get(); + BOOST_COROSIO_ASSERT( + impl.heap_index_.load(std::memory_order_relaxed) == + implementation::npos); if (d <= duration::zero()) impl.expiry_ = (time_point::min)(); else @@ -382,39 +352,29 @@ class BOOST_COROSIO_DECL timer : public io_object ? (time_point::max)() : now + d; } - if (impl.heap_index_.load(std::memory_order_relaxed) == - implementation::npos && - !impl.might_have_pending_waits_.load(std::memory_order_relaxed)) - return 0; - return do_update_expiry(); } /** Set the timer's expiry time relative to now. This is a convenience overload that accepts any duration type - and converts it to the timer's native duration type. Any - pending asynchronous wait operations will be cancelled. + and converts it to the timer's native duration type. @param d The expiry time relative to now. - - @return The number of pending operations that were cancelled. */ template - std::size_t expires_after(std::chrono::duration d) + void expires_after(std::chrono::duration d) { - return expires_after(std::chrono::duration_cast(d)); + expires_after(std::chrono::duration_cast(d)); } /** Wait for the timer to expire. - Multiple coroutines may wait on the same timer concurrently. - When the timer expires, all waiters complete with success. + At most one wait may be outstanding at a time. The operation supports cancellation via `std::stop_token` through the affine awaitable protocol. If the associated stop token is triggered, only that waiter completes with an error that - compares equal to `capy::cond::canceled`; other waiters are - unaffected. + compares equal to `capy::cond::canceled`. This timer must outlive the returned awaitable. @@ -469,13 +429,6 @@ class BOOST_COROSIO_DECL timer : public io_object explicit timer(handle h) noexcept : io_object(std::move(h)) {} private: - // Defined in src/corosio/src/timer.cpp, which includes both this - // header and timer_service.hpp, so the timer_service_* free - // functions are visible there. - std::size_t do_cancel(); - std::size_t do_cancel_one(); - std::size_t do_update_expiry(); - /// Return the underlying implementation. implementation& get() const noexcept { diff --git a/include/boost/corosio/detail/timer_service.hpp b/include/boost/corosio/detail/timer_service.hpp index e45afed4c..47af42786 100644 --- a/include/boost/corosio/detail/timer_service.hpp +++ b/include/boost/corosio/detail/timer_service.hpp @@ -48,8 +48,8 @@ struct scheduler; 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. + index, and an intrusive_list of waiter_nodes. Each timer holds + at most one waiter. timer_service owns a min-heap of active timers and a free list of recycled impls. The heap is ordered by expiry time; the @@ -185,22 +185,15 @@ class BOOST_COROSIO_DECL timer_service final /// Cancel and recycle a timer implementation. inline void destroy_impl(timer::implementation& impl); - /// Update the timer expiry, cancelling existing waiters. - inline std::size_t update_timer( - timer::implementation& impl, time_point new_time); - /// Insert a waiter into the timer's waiter list and the heap. inline void insert_waiter(timer::implementation& impl, waiter_node* w); /// Cancel all waiters on a timer. - inline std::size_t cancel_timer(timer::implementation& impl); + inline void cancel_timer(timer::implementation& impl); /// Cancel one specific waiter ( stop_token callback path ). inline void cancel_waiter(waiter_node* w); - /// 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. inline std::size_t process_expired(); @@ -413,65 +406,6 @@ timer_service::destroy_impl(timer::implementation& impl) free_list_ = &impl; } -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: 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)()); - if (!in_heap && - !impl.might_have_pending_waits_.load(std::memory_order_relaxed)) - return 0; - - bool notify = false; - intrusive_list canceled; - - { - std::lock_guard lock(mutex_); - - while (auto* w = impl.waiters_.pop_front()) - { - w->impl_ = nullptr; - canceled.push_back(w); - } - - std::size_t idx = impl.heap_index_.load(std::memory_order_relaxed); - if (idx < heap_.size()) - { - time_point old_time = heap_[idx].time_; - heap_[idx].time_ = new_time; - - if (new_time < old_time) - up_heap(idx); - else - down_heap(idx); - - notify = - (impl.heap_index_.load(std::memory_order_relaxed) == 0); - } - - refresh_cached_nearest(); - } - - std::size_t count = 0; - while (auto* w = canceled.pop_front()) - { - w->ec_ = make_error_code(capy::error::canceled); - sched_->post(&w->op_); - ++count; - } - - if (notify) - on_earliest_changed_(); - - return count; -} - inline void timer_service::insert_waiter(timer::implementation& impl, waiter_node* w) { @@ -529,11 +463,11 @@ timer_service::insert_waiter(timer::implementation& impl, waiter_node* w) } } -inline std::size_t +inline void timer_service::cancel_timer(timer::implementation& impl) { if (!impl.might_have_pending_waits_.load(std::memory_order_relaxed)) - return 0; + return; // No unlocked already-done fast-out here: it would need the // non-atomic waiters_ (a race with concurrent drains), and an @@ -553,20 +487,16 @@ timer_service::cancel_timer(timer::implementation& impl) canceled.push_back(w); } // Store false as the final touch of the impl under the lock so - // update_timer's pre-lock false-flag trust holds unqualified. + // a pre-lock false-flag check trusts it unqualified. impl.might_have_pending_waits_.store(false, std::memory_order_relaxed); refresh_cached_nearest(); } - std::size_t count = 0; while (auto* w = canceled.pop_front()) { w->ec_ = make_error_code(capy::error::canceled); sched_->post(&w->op_); - ++count; } - - return count; } inline void @@ -575,8 +505,7 @@ timer_service::cancel_waiter(waiter_node* w) { std::lock_guard lock(mutex_); // Already removed by another drain: cancel_timer, - // cancel_one_waiter, update_timer, process_expired, or - // insert_waiter's lost-cancel recheck + // process_expired, or insert_waiter's lost-cancel recheck if (!w->impl_) return; auto* impl = w->impl_; @@ -595,34 +524,6 @@ timer_service::cancel_waiter(waiter_node* w) sched_->post(&w->op_); } -inline std::size_t -timer_service::cancel_one_waiter(timer::implementation& impl) -{ - if (!impl.might_have_pending_waits_.load(std::memory_order_relaxed)) - return 0; - - waiter_node* w = nullptr; - - { - std::lock_guard lock(mutex_); - w = impl.waiters_.pop_front(); - if (!w) - return 0; - w->impl_ = nullptr; - if (impl.waiters_.empty()) - { - remove_timer_impl(impl); - impl.might_have_pending_waits_.store( - false, std::memory_order_relaxed); - } - refresh_cached_nearest(); - } - - w->ec_ = make_error_code(capy::error::canceled); - sched_->post(&w->op_); - return 1; -} - inline std::size_t timer_service::process_expired() { @@ -749,24 +650,6 @@ timer_service::swap_heap(std::size_t i1, std::size_t i2) // Free functions -inline std::size_t -timer_service_update_expiry(timer::implementation& impl) -{ - return impl.svc_->update_timer(impl, impl.expiry_); -} - -inline std::size_t -timer_service_cancel(timer::implementation& impl) noexcept -{ - return impl.svc_->cancel_timer(impl); -} - -inline std::size_t -timer_service_cancel_one(timer::implementation& impl) noexcept -{ - return impl.svc_->cancel_one_waiter(impl); -} - inline timer_service& get_timer_service(capy::execution_context& ctx, scheduler& sched) { diff --git a/src/corosio/src/timer.cpp b/src/corosio/src/timer.cpp index e4c05aa6f..1f84b462d 100644 --- a/src/corosio/src/timer.cpp +++ b/src/corosio/src/timer.cpp @@ -35,24 +35,6 @@ timer::operator=(timer&& other) noexcept return *this; } -std::size_t -timer::do_cancel() -{ - return detail::timer_service_cancel(get()); -} - -std::size_t -timer::do_cancel_one() -{ - return detail::timer_service_cancel_one(get()); -} - -std::size_t -timer::do_update_expiry() -{ - return detail::timer_service_update_expiry(get()); -} - // Not inline: wait_awaitable::await_suspend (defined in timer.hpp) calls // this from translation units that may never include timer_service.hpp, // so this must be the one strong definition the linker can always find From d18e6a7329b693de686155f32959ece985a1dd7c Mon Sep 17 00:00:00 2001 From: Steve Gerbino Date: Wed, 5 Aug 2026 17:50:47 +0200 Subject: [PATCH 3/3] refactor(timer): store the single published waiter as a pointer --- include/boost/corosio/detail/timer.hpp | 30 +++++---- .../boost/corosio/detail/timer_service.hpp | 65 +++++++++---------- src/corosio/src/timer.cpp | 2 +- 3 files changed, 50 insertions(+), 47 deletions(-) diff --git a/include/boost/corosio/detail/timer.hpp b/include/boost/corosio/detail/timer.hpp index a7473ecc2..f038ba5e7 100644 --- a/include/boost/corosio/detail/timer.hpp +++ b/include/boost/corosio/detail/timer.hpp @@ -39,9 +39,9 @@ namespace boost::corosio::detail { // 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. +// forward-declared as a nested type. implementation stores only a +// waiter_node pointer, so this forward declaration suffices for its +// data layout. class timer_service; struct waiter_node; struct wait_awaitable; @@ -77,7 +77,7 @@ class BOOST_COROSIO_DECL timer : public io_object public: /** Backend state and wait entry point for a timer. - Holds per-timer state (expiry, heap position, waiter list) and + Holds per-timer state ( expiry, heap position, the single waiter ) and the `wait` entry point used by the awaitable returned from @ref timer::wait. There is exactly one concrete timer backend, so `wait` is a plain member function rather than a virtual @@ -98,7 +98,7 @@ class BOOST_COROSIO_DECL timer : public io_object // heap_index_ and might_have_pending_waits_ are cross-thread // hints, not authoritative state: the real state lives in the - // heap and waiter list under timer_service::mutex_. Every + // heap and the published waiter under timer_service::mutex_. Every // unlocked fast-out that reads them is either re-validated under // the mutex or safe under a stale value in both directions, and // any locked writer / locked reader pair is already ordered by @@ -108,14 +108,19 @@ class BOOST_COROSIO_DECL timer : public io_object /// Index in the timer service's min-heap, or `npos`. std::atomic heap_index_{npos}; + // false implies waiter_ is null: both are cleared together + // under the service mutex. /// True if `wait()` has been called since last cancel. std::atomic might_have_pending_waits_{false}; /// The timer service that owns this implementation. timer_service* svc_ = nullptr; - /// Coroutines currently waiting on this timer's expiry. - intrusive_list waiters_; + // Exactly one wait may be outstanding: delay and timeout own + // a private timer per co_await, and the service's drains rely + // on the one-to-one pairing. + /// The waiter published on this timer, or `nullptr`. + waiter_node* waiter_ = nullptr; /// Free list linkage, reused when this impl is recycled. implementation* next_free_ = nullptr; @@ -140,10 +145,11 @@ class BOOST_COROSIO_DECL timer : public io_object /** 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. + Publishes the waiter into the service's heap and the + timer's waiter slot, 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 @@ -482,7 +488,7 @@ struct BOOST_COROSIO_SYMBOL_VISIBLE waiter_node using stop_cb_type = std::stop_callback; - // nullptr once removed from timer's waiter list (concurrency marker) + // nullptr once unpublished from the timer ( concurrency marker ) /// The timer this waiter is published on, or `nullptr`. timer::implementation* impl_ = nullptr; diff --git a/include/boost/corosio/detail/timer_service.hpp b/include/boost/corosio/detail/timer_service.hpp index 47af42786..f9f599cad 100644 --- a/include/boost/corosio/detail/timer_service.hpp +++ b/include/boost/corosio/detail/timer_service.hpp @@ -48,8 +48,10 @@ struct scheduler; frame — waits perform no allocation. timer::implementation holds per-timer state: expiry, heap - index, and an intrusive_list of waiter_nodes. Each timer holds - at most one waiter. + index, and the single published waiter. Each timer holds + at most one waiter; process_expired's local cross-timer drain + list still threads waiters through their intrusive hooks when + collecting several timers' waiters past the lock. timer_service owns a min-heap of active timers and a free list of recycled impls. The heap is ordered by expiry time; the @@ -185,10 +187,10 @@ class BOOST_COROSIO_DECL timer_service final /// Cancel and recycle a timer implementation. inline void destroy_impl(timer::implementation& impl); - /// Insert a waiter into the timer's waiter list and the heap. + /// Publish the timer's waiter and insert the timer into the heap. inline void insert_waiter(timer::implementation& impl, waiter_node* w); - /// Cancel all waiters on a timer. + /// Cancel the timer's published waiter, if any. inline void cancel_timer(timer::implementation& impl); /// Cancel one specific waiter ( stop_token callback path ). @@ -306,7 +308,7 @@ timer_service::shutdown() // this is harmless. for (auto* impl : impls) { - while (auto* w = impl->waiters_.pop_front()) + if (auto* w = std::exchange(impl->waiter_, nullptr)) { w->reset_stop_cb(); auto h = std::exchange(w->h_, {}); @@ -341,6 +343,7 @@ timer_service::construct() (std::numeric_limits::max)(), std::memory_order_relaxed); impl->might_have_pending_waits_.store(false, std::memory_order_relaxed); + BOOST_COROSIO_ASSERT(impl->waiter_ == nullptr); return impl; } @@ -356,6 +359,7 @@ timer_service::construct() (std::numeric_limits::max)(), std::memory_order_relaxed); impl->might_have_pending_waits_.store(false, std::memory_order_relaxed); + BOOST_COROSIO_ASSERT(impl->waiter_ == nullptr); } else { @@ -434,21 +438,19 @@ timer_service::insert_waiter(timer::implementation& impl, waiter_node* w) (impl.heap_index_.load(std::memory_order_relaxed) == 0); refresh_cached_nearest(); } - impl.waiters_.push_back(w); + BOOST_COROSIO_ASSERT(impl.waiter_ == nullptr); + impl.waiter_ = 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()) { - w->impl_ = nullptr; - impl.waiters_.remove(w); - if (impl.waiters_.empty()) - { - remove_timer_impl(impl); - impl.might_have_pending_waits_.store( - false, std::memory_order_relaxed); - } + w->impl_ = nullptr; + impl.waiter_ = nullptr; + remove_timer_impl(impl); + impl.might_have_pending_waits_.store( + false, std::memory_order_relaxed); refresh_cached_nearest(); lost_cancel = true; notify = false; // insertion undone; nearest unchanged @@ -470,32 +472,30 @@ timer_service::cancel_timer(timer::implementation& impl) return; // No unlocked already-done fast-out here: it would need the - // non-atomic waiters_ (a race with concurrent drains), and an + // non-atomic waiter_ (a race with concurrent drains), and an // index-only check is lifetime-unsafe because npos is stored // before the drain finishes touching the impl. A stale-true // flag is rare with the stateless API; the locked path below // re-validates. - intrusive_list canceled; + waiter_node* canceled = nullptr; { std::lock_guard lock(mutex_); remove_timer_impl(impl); - while (auto* w = impl.waiters_.pop_front()) - { - w->impl_ = nullptr; - canceled.push_back(w); - } + canceled = std::exchange(impl.waiter_, nullptr); + if (canceled) + canceled->impl_ = nullptr; // Store false as the final touch of the impl under the lock so // a pre-lock false-flag check trusts it unqualified. impl.might_have_pending_waits_.store(false, std::memory_order_relaxed); refresh_cached_nearest(); } - while (auto* w = canceled.pop_front()) + if (canceled) { - w->ec_ = make_error_code(capy::error::canceled); - sched_->post(&w->op_); + canceled->ec_ = make_error_code(capy::error::canceled); + sched_->post(&canceled->op_); } } @@ -508,15 +508,12 @@ timer_service::cancel_waiter(waiter_node* w) // process_expired, or insert_waiter's lost-cancel recheck if (!w->impl_) return; - auto* impl = w->impl_; - w->impl_ = nullptr; - impl->waiters_.remove(w); - if (impl->waiters_.empty()) - { - remove_timer_impl(*impl); - impl->might_have_pending_waits_.store( - false, std::memory_order_relaxed); - } + auto* impl = w->impl_; + w->impl_ = nullptr; + impl->waiter_ = nullptr; + remove_timer_impl(*impl); + impl->might_have_pending_waits_.store( + false, std::memory_order_relaxed); refresh_cached_nearest(); } @@ -537,7 +534,7 @@ timer_service::process_expired() { timer::implementation* t = heap_[0].timer_; remove_timer_impl(*t); - while (auto* w = t->waiters_.pop_front()) + if (auto* w = std::exchange(t->waiter_, nullptr)) { w->impl_ = nullptr; w->ec_ = {}; diff --git a/src/corosio/src/timer.cpp b/src/corosio/src/timer.cpp index 1f84b462d..669284d8e 100644 --- a/src/corosio/src/timer.cpp +++ b/src/corosio/src/timer.cpp @@ -60,7 +60,7 @@ timer::implementation::publish(waiter_node& w) { // Publication-last invariant: fully initialize the waiter, count // its work, and arm cancellation BEFORE insert_waiter() publishes - // it into the heap/list where a concurrent run() thread can fire + // it into the heap/waiter slot where a concurrent run() thread can fire // it. impl_ stays null until insert_waiter() sets it under the // mutex, so a stop callback that fires early (cancel_waiter) sees a // null impl_ and is a safe no-op. To avoid losing such an early