1   
//
1   
//
2   
// Copyright (c) 2026 Michael Vandeberg
2   
// Copyright (c) 2026 Michael Vandeberg
3   
//
3   
//
4   
// Distributed under the Boost Software License, Version 1.0. (See accompanying
4   
// Distributed under the Boost Software License, Version 1.0. (See accompanying
5   
// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
5   
// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
6   
//
6   
//
7   
// Official repository: https://github.com/cppalliance/capy
7   
// Official repository: https://github.com/cppalliance/capy
8   
//
8   
//
9  
9  
10   
#ifndef BOOST_CAPY_DELAY_HPP
10   
#ifndef BOOST_CAPY_DELAY_HPP
11   
#define BOOST_CAPY_DELAY_HPP
11   
#define BOOST_CAPY_DELAY_HPP
12  
12  
13   
#include <boost/capy/detail/config.hpp>
13   
#include <boost/capy/detail/config.hpp>
14   
#include <boost/capy/ex/executor_ref.hpp>
14   
#include <boost/capy/ex/executor_ref.hpp>
15   
#include <boost/capy/ex/io_env.hpp>
15   
#include <boost/capy/ex/io_env.hpp>
16   
#include <boost/capy/ex/detail/timer_service.hpp>
16   
#include <boost/capy/ex/detail/timer_service.hpp>
17  
17  
18   
#include <atomic>
18   
#include <atomic>
19   
#include <chrono>
19   
#include <chrono>
20   
#include <coroutine>
20   
#include <coroutine>
21   
#include <new>
21   
#include <new>
22   
#include <stop_token>
22   
#include <stop_token>
23   
#include <utility>
23   
#include <utility>
24  
24  
25   
namespace boost {
25   
namespace boost {
26   
namespace capy {
26   
namespace capy {
27  
27  
28   
/** IoAwaitable returned by @ref delay.
28   
/** IoAwaitable returned by @ref delay.
29  
29  
30   
    Suspends the calling coroutine until the deadline elapses
30   
    Suspends the calling coroutine until the deadline elapses
31   
    or the environment's stop token is activated, whichever
31   
    or the environment's stop token is activated, whichever
32   
    comes first. Resumption is always posted through the
32   
    comes first. Resumption is always posted through the
33   
    executor, never inline on the timer thread.
33   
    executor, never inline on the timer thread.
34  
34  
35   
    Not intended to be named directly; use the @ref delay
35   
    Not intended to be named directly; use the @ref delay
36   
    factory function instead.
36   
    factory function instead.
37  
37  
38   
    @par Cancellation
38   
    @par Cancellation
39  
39  
40   
    If `stop_requested()` is true before suspension, the
40   
    If `stop_requested()` is true before suspension, the
41   
    coroutine resumes immediately without scheduling a timer.
41   
    coroutine resumes immediately without scheduling a timer.
42   
    If stop is requested while suspended, the stop callback
42   
    If stop is requested while suspended, the stop callback
43   
    claims the resume and posts it through the executor; the
43   
    claims the resume and posts it through the executor; the
44   
    pending timer is cancelled on the next `await_resume` or
44   
    pending timer is cancelled on the next `await_resume` or
45   
    destructor call.
45   
    destructor call.
46  
46  
47   
    @par Thread Safety
47   
    @par Thread Safety
48  
48  
49   
    A single `delay_awaitable` must not be awaited concurrently.
49   
    A single `delay_awaitable` must not be awaited concurrently.
50   
    Multiple independent `delay()` calls on the same
50   
    Multiple independent `delay()` calls on the same
51   
    execution_context are safe and share one timer thread.
51   
    execution_context are safe and share one timer thread.
52  
52  
53   
    @see delay, timeout
53   
    @see delay, timeout
54   
*/
54   
*/
55   
class delay_awaitable
55   
class delay_awaitable
56   
{
56   
{
57   
    std::chrono::nanoseconds dur_;
57   
    std::chrono::nanoseconds dur_;
58  
58  
59   
    detail::timer_service* ts_ = nullptr;
59   
    detail::timer_service* ts_ = nullptr;
60   
    detail::timer_service::timer_id tid_ = 0;
60   
    detail::timer_service::timer_id tid_ = 0;
61  
61  
62   
    // Declared before stop_cb_buf_: the callback
62   
    // Declared before stop_cb_buf_: the callback
63   
    // accesses these members, so they must still be
63   
    // accesses these members, so they must still be
64   
    // alive if the stop_cb_ destructor blocks.
64   
    // alive if the stop_cb_ destructor blocks.
65   
    std::atomic<bool> claimed_{false};
65   
    std::atomic<bool> claimed_{false};
66   
    bool canceled_ = false;
66   
    bool canceled_ = false;
67   
    bool stop_cb_active_ = false;
67   
    bool stop_cb_active_ = false;
68  
68  
69   
    struct cancel_fn
69   
    struct cancel_fn
70   
    {
70   
    {
71   
        delay_awaitable* self_;
71   
        delay_awaitable* self_;
72   
        executor_ref ex_;
72   
        executor_ref ex_;
73   
        std::coroutine_handle<> h_;
73   
        std::coroutine_handle<> h_;
74  
74  
75   
        void operator()() const noexcept
75   
        void operator()() const noexcept
76   
        {
76   
        {
77   
            if(!self_->claimed_.exchange(
77   
            if(!self_->claimed_.exchange(
78   
                true, std::memory_order_acq_rel))
78   
                true, std::memory_order_acq_rel))
79   
            {
79   
            {
80   
                self_->canceled_ = true;
80   
                self_->canceled_ = true;
81   
                ex_.post(h_);
81   
                ex_.post(h_);
82   
            }
82   
            }
83   
        }
83   
        }
84   
    };
84   
    };
85  
85  
86   
    using stop_cb_t = std::stop_callback<cancel_fn>;
86   
    using stop_cb_t = std::stop_callback<cancel_fn>;
87  
87  
88   
    // Aligned storage for the stop callback.
88   
    // Aligned storage for the stop callback.
89   
    // Declared last: its destructor may block while
89   
    // Declared last: its destructor may block while
90   
    // the callback accesses the members above.
90   
    // the callback accesses the members above.
91 - 
    BOOST_CAPY_MSVC_WARNING_PUSH
91 + 
#ifdef _MSC_VER
92 - 
    BOOST_CAPY_MSVC_WARNING_DISABLE(4324)
92 + 
# pragma warning(push)
 
93 + 
# pragma warning(disable: 4324)
 
94 + 
#endif
93   
    alignas(stop_cb_t)
95   
    alignas(stop_cb_t)
94   
        unsigned char stop_cb_buf_[sizeof(stop_cb_t)];
96   
        unsigned char stop_cb_buf_[sizeof(stop_cb_t)];
95 - 
    BOOST_CAPY_MSVC_WARNING_POP
97 + 
#ifdef _MSC_VER
 
98 + 
# pragma warning(pop)
 
99 + 
#endif
96  
100  
97   
    stop_cb_t& stop_cb_() noexcept
101   
    stop_cb_t& stop_cb_() noexcept
98   
    {
102   
    {
99   
        return *reinterpret_cast<stop_cb_t*>(stop_cb_buf_);
103   
        return *reinterpret_cast<stop_cb_t*>(stop_cb_buf_);
100   
    }
104   
    }
101  
105  
102   
public:
106   
public:
103   
    explicit delay_awaitable(std::chrono::nanoseconds dur) noexcept
107   
    explicit delay_awaitable(std::chrono::nanoseconds dur) noexcept
104   
        : dur_(dur)
108   
        : dur_(dur)
105   
    {
109   
    {
106   
    }
110   
    }
107  
111  
108   
    /// @pre The stop callback must not be active
112   
    /// @pre The stop callback must not be active
109   
    ///      (i.e. the object has not yet been awaited).
113   
    ///      (i.e. the object has not yet been awaited).
110   
    delay_awaitable(delay_awaitable&& o) noexcept
114   
    delay_awaitable(delay_awaitable&& o) noexcept
111   
        : dur_(o.dur_)
115   
        : dur_(o.dur_)
112   
        , ts_(o.ts_)
116   
        , ts_(o.ts_)
113   
        , tid_(o.tid_)
117   
        , tid_(o.tid_)
114   
        , claimed_(o.claimed_.load(std::memory_order_relaxed))
118   
        , claimed_(o.claimed_.load(std::memory_order_relaxed))
115   
        , canceled_(o.canceled_)
119   
        , canceled_(o.canceled_)
116   
        , stop_cb_active_(std::exchange(o.stop_cb_active_, false))
120   
        , stop_cb_active_(std::exchange(o.stop_cb_active_, false))
117   
    {
121   
    {
118   
    }
122   
    }
119  
123  
120   
    ~delay_awaitable()
124   
    ~delay_awaitable()
121   
    {
125   
    {
122   
        if(stop_cb_active_)
126   
        if(stop_cb_active_)
123   
            stop_cb_().~stop_cb_t();
127   
            stop_cb_().~stop_cb_t();
124   
        if(ts_)
128   
        if(ts_)
125   
            ts_->cancel(tid_);
129   
            ts_->cancel(tid_);
126   
    }
130   
    }
127  
131  
128   
    delay_awaitable(delay_awaitable const&) = delete;
132   
    delay_awaitable(delay_awaitable const&) = delete;
129   
    delay_awaitable& operator=(delay_awaitable const&) = delete;
133   
    delay_awaitable& operator=(delay_awaitable const&) = delete;
130   
    delay_awaitable& operator=(delay_awaitable&&) = delete;
134   
    delay_awaitable& operator=(delay_awaitable&&) = delete;
131  
135  
132   
    bool await_ready() const noexcept
136   
    bool await_ready() const noexcept
133   
    {
137   
    {
134   
        return dur_.count() <= 0;
138   
        return dur_.count() <= 0;
135   
    }
139   
    }
136  
140  
137   
    std::coroutine_handle<>
141   
    std::coroutine_handle<>
138   
    await_suspend(
142   
    await_suspend(
139   
        std::coroutine_handle<> h,
143   
        std::coroutine_handle<> h,
140   
        io_env const* env) noexcept
144   
        io_env const* env) noexcept
141   
    {
145   
    {
142   
        // Already stopped: resume immediately
146   
        // Already stopped: resume immediately
143   
        if(env->stop_token.stop_requested())
147   
        if(env->stop_token.stop_requested())
144   
        {
148   
        {
145   
            canceled_ = true;
149   
            canceled_ = true;
146   
            return h;
150   
            return h;
147   
        }
151   
        }
148  
152  
149   
        ts_ = &env->executor.context().use_service<detail::timer_service>();
153   
        ts_ = &env->executor.context().use_service<detail::timer_service>();
150  
154  
151 - 
        // Schedule timer (won't fire inline since deadline is in the future)
155 + 
        // Register stop callback before arming the timer.
152 - 
        tid_ = ts_->schedule_after(dur_,
156 + 
        // Once the timer is armed, another thread can fire it,
 
157 + 
        // resume the coroutine, and destroy this awaitable.
 
158 + 
        stop_cb_active_ = true;
 
159 + 
        ::new(stop_cb_buf_) stop_cb_t(
 
160 + 
            env->stop_token,
 
161 + 
            cancel_fn{this, env->executor, h});
 
162 +
 
163 + 
        // If the stop callback already claimed the resume
 
164 + 
        // (inline invocation), skip the timer entirely.
 
165 + 
        if(claimed_.load(std::memory_order_acquire))
 
166 + 
            return std::noop_coroutine();
 
167 +
 
168 + 
        // Schedule timer using the output-reference overload so
 
169 + 
        // that tid_ is written while the timer_service lock is
 
170 + 
        // held — the timer thread cannot fire the callback until
 
171 + 
        // after the lock is released, at which point tid_ is set.
 
172 + 
        ts_->schedule_after(dur_,
153   
            [this, h, ex = env->executor]()
173   
            [this, h, ex = env->executor]()
154   
            {
174   
            {
155   
                if(!claimed_.exchange(
175   
                if(!claimed_.exchange(
156   
                    true, std::memory_order_acq_rel))
176   
                    true, std::memory_order_acq_rel))
157   
                {
177   
                {
158   
                    ex.post(h);
178   
                    ex.post(h);
159   
                }
179   
                }
160 - 
            });
180 + 
            },
161 -
181 + 
            tid_);
162 - 
        // Register stop callback (may fire inline)
 
163 - 
        ::new(stop_cb_buf_) stop_cb_t(
 
164 - 
            env->stop_token,
 
165 - 
            cancel_fn{this, env->executor, h});
 
166 - 
        stop_cb_active_ = true;
 
167  
182  
168   
        return std::noop_coroutine();
183   
        return std::noop_coroutine();
169   
    }
184   
    }
170  
185  
171   
    void await_resume() noexcept
186   
    void await_resume() noexcept
172   
    {
187   
    {
173   
        if(stop_cb_active_)
188   
        if(stop_cb_active_)
174   
        {
189   
        {
175   
            stop_cb_().~stop_cb_t();
190   
            stop_cb_().~stop_cb_t();
176   
            stop_cb_active_ = false;
191   
            stop_cb_active_ = false;
177   
        }
192   
        }
178   
        if(ts_)
193   
        if(ts_)
179   
            ts_->cancel(tid_);
194   
            ts_->cancel(tid_);
180   
    }
195   
    }
181   
};
196   
};
182  
197  
183   
/** Suspend the current coroutine for a duration.
198   
/** Suspend the current coroutine for a duration.
184  
199  
185   
    Returns an IoAwaitable that completes at or after the
200   
    Returns an IoAwaitable that completes at or after the
186   
    specified duration, or earlier if the environment's stop
201   
    specified duration, or earlier if the environment's stop
187   
    token is activated. Completion is always normal (void
202   
    token is activated. Completion is always normal (void
188   
    return); no exception is thrown on cancellation.
203   
    return); no exception is thrown on cancellation.
189  
204  
190   
    Zero or negative durations complete synchronously without
205   
    Zero or negative durations complete synchronously without
191   
    scheduling a timer.
206   
    scheduling a timer.
192  
207  
193   
    @par Example
208   
    @par Example
194   
    @code
209   
    @code
195   
    co_await delay(std::chrono::milliseconds(100));
210   
    co_await delay(std::chrono::milliseconds(100));
196   
    @endcode
211   
    @endcode
197  
212  
198   
    @param dur The duration to wait.
213   
    @param dur The duration to wait.
199  
214  
200   
    @return A @ref delay_awaitable whose `await_resume`
215   
    @return A @ref delay_awaitable whose `await_resume`
201   
        returns `void`.
216   
        returns `void`.
202  
217  
203   
    @throws Nothing.
218   
    @throws Nothing.
204  
219  
205   
    @see timeout, delay_awaitable
220   
    @see timeout, delay_awaitable
206   
*/
221   
*/
207   
template<typename Rep, typename Period>
222   
template<typename Rep, typename Period>
208   
delay_awaitable
223   
delay_awaitable
209   
delay(std::chrono::duration<Rep, Period> dur) noexcept
224   
delay(std::chrono::duration<Rep, Period> dur) noexcept
210   
{
225   
{
211   
    return delay_awaitable{
226   
    return delay_awaitable{
212   
        std::chrono::duration_cast<std::chrono::nanoseconds>(dur)};
227   
        std::chrono::duration_cast<std::chrono::nanoseconds>(dur)};
213   
}
228   
}
214  
229  
215   
} // capy
230   
} // capy
216   
} // boost
231   
} // boost
217  
232  
218   
#endif
233   
#endif