1   
//
1   
//
2   
// Copyright (c) 2025 Vinnie Falco (vinnie.falco@gmail.com)
2   
// Copyright (c) 2025 Vinnie Falco (vinnie.falco@gmail.com)
3   
// Copyright (c) 2026 Steve Gerbino
3   
// Copyright (c) 2026 Steve Gerbino
4   
//
4   
//
5   
// Distributed under the Boost Software License, Version 1.0. (See accompanying
5   
// Distributed under the Boost Software License, Version 1.0. (See accompanying
6   
// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
6   
// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
7   
//
7   
//
8   
// Official repository: https://github.com/cppalliance/corosio
8   
// Official repository: https://github.com/cppalliance/corosio
9   
//
9   
//
10  
10  
11   
#ifndef BOOST_COROSIO_DETAIL_TIMER_SERVICE_HPP
11   
#ifndef BOOST_COROSIO_DETAIL_TIMER_SERVICE_HPP
12   
#define BOOST_COROSIO_DETAIL_TIMER_SERVICE_HPP
12   
#define BOOST_COROSIO_DETAIL_TIMER_SERVICE_HPP
13  
13  
14   
#include <boost/corosio/timer.hpp>
14   
#include <boost/corosio/timer.hpp>
15   
#include <boost/corosio/io_context.hpp>
15   
#include <boost/corosio/io_context.hpp>
16   
#include <boost/corosio/detail/scheduler_op.hpp>
16   
#include <boost/corosio/detail/scheduler_op.hpp>
17   
#include <boost/corosio/native/native_scheduler.hpp>
17   
#include <boost/corosio/native/native_scheduler.hpp>
18   
#include <boost/corosio/detail/intrusive.hpp>
18   
#include <boost/corosio/detail/intrusive.hpp>
19   
#include <boost/corosio/detail/thread_local_ptr.hpp>
19   
#include <boost/corosio/detail/thread_local_ptr.hpp>
20   
#include <boost/capy/error.hpp>
20   
#include <boost/capy/error.hpp>
21   
#include <boost/capy/ex/execution_context.hpp>
21   
#include <boost/capy/ex/execution_context.hpp>
22   
#include <boost/capy/ex/executor_ref.hpp>
22   
#include <boost/capy/ex/executor_ref.hpp>
23   
#include <system_error>
23   
#include <system_error>
24  
24  
25   
#include <atomic>
25   
#include <atomic>
26   
#include <chrono>
26   
#include <chrono>
27   
#include <coroutine>
27   
#include <coroutine>
28   
#include <cstddef>
28   
#include <cstddef>
29   
#include <limits>
29   
#include <limits>
30   
#include <mutex>
30   
#include <mutex>
31   
#include <optional>
31   
#include <optional>
32 - 
#include <utility>
 
33   
#include <stop_token>
32   
#include <stop_token>
34   
#include <vector>
33   
#include <vector>
35  
34  
36   
namespace boost::corosio::detail {
35   
namespace boost::corosio::detail {
37  
36  
38   
struct scheduler;
37   
struct scheduler;
39  
38  
40   
/*
39   
/*
41   
    Timer Service
40   
    Timer Service
42   
    =============
41   
    =============
43  
42  
44   
    Data Structures
43   
    Data Structures
45   
    ---------------
44   
    ---------------
46   
    waiter_node holds per-waiter state: coroutine handle, executor,
45   
    waiter_node holds per-waiter state: coroutine handle, executor,
47   
    error output, stop_token, embedded completion_op. Each concurrent
46   
    error output, stop_token, embedded completion_op. Each concurrent
48   
    co_await t.wait() allocates one waiter_node.
47   
    co_await t.wait() allocates one waiter_node.
49  
48  
50   
    timer_service::implementation holds per-timer state: expiry,
49   
    timer_service::implementation holds per-timer state: expiry,
51   
    heap index, and an intrusive_list of waiter_nodes. Multiple
50   
    heap index, and an intrusive_list of waiter_nodes. Multiple
52   
    coroutines can wait on the same timer simultaneously.
51   
    coroutines can wait on the same timer simultaneously.
53  
52  
54   
    timer_service owns a min-heap of active timers, a free list
53   
    timer_service owns a min-heap of active timers, a free list
55   
    of recycled impls, and a free list of recycled waiter_nodes. The
54   
    of recycled impls, and a free list of recycled waiter_nodes. The
56   
    heap is ordered by expiry time; the scheduler queries
55   
    heap is ordered by expiry time; the scheduler queries
57   
    nearest_expiry() to set the epoll/timerfd timeout.
56   
    nearest_expiry() to set the epoll/timerfd timeout.
58  
57  
59   
    Optimization Strategy
58   
    Optimization Strategy
60   
    ---------------------
59   
    ---------------------
61   
    1. Deferred heap insertion — expires_after() stores the expiry
60   
    1. Deferred heap insertion — expires_after() stores the expiry
62   
       but does not insert into the heap. Insertion happens in wait().
61   
       but does not insert into the heap. Insertion happens in wait().
63   
    2. Thread-local impl cache — single-slot per-thread cache.
62   
    2. Thread-local impl cache — single-slot per-thread cache.
64   
    3. Embedded completion_op — eliminates heap allocation per fire/cancel.
63   
    3. Embedded completion_op — eliminates heap allocation per fire/cancel.
65   
    4. Cached nearest expiry — atomic avoids mutex in nearest_expiry().
64   
    4. Cached nearest expiry — atomic avoids mutex in nearest_expiry().
66   
    5. might_have_pending_waits_ flag — skips lock when no wait issued.
65   
    5. might_have_pending_waits_ flag — skips lock when no wait issued.
67   
    6. Thread-local waiter cache — single-slot per-thread cache.
66   
    6. Thread-local waiter cache — single-slot per-thread cache.
68  
67  
69   
    Concurrency
68   
    Concurrency
70   
    -----------
69   
    -----------
71   
    stop_token callbacks can fire from any thread. The impl_
70   
    stop_token callbacks can fire from any thread. The impl_
72   
    pointer on waiter_node is used as a "still in list" marker.
71   
    pointer on waiter_node is used as a "still in list" marker.
73   
*/
72   
*/
74  
73  
75   
struct BOOST_COROSIO_SYMBOL_VISIBLE waiter_node;
74   
struct BOOST_COROSIO_SYMBOL_VISIBLE waiter_node;
76  
75  
77   
inline void timer_service_invalidate_cache() noexcept;
76   
inline void timer_service_invalidate_cache() noexcept;
78  
77  
79   
// timer_service class body — member function definitions are
78   
// timer_service class body — member function definitions are
80   
// out-of-class (after implementation and waiter_node are complete)
79   
// out-of-class (after implementation and waiter_node are complete)
81   
class BOOST_COROSIO_DECL timer_service final
80   
class BOOST_COROSIO_DECL timer_service final
82   
    : public capy::execution_context::service
81   
    : public capy::execution_context::service
83   
    , public io_object::io_service
82   
    , public io_object::io_service
84   
{
83   
{
85   
public:
84   
public:
86   
    using clock_type = std::chrono::steady_clock;
85   
    using clock_type = std::chrono::steady_clock;
87   
    using time_point = clock_type::time_point;
86   
    using time_point = clock_type::time_point;
88  
87  
89   
    class callback
88   
    class callback
90   
    {
89   
    {
91   
        void* ctx_         = nullptr;
90   
        void* ctx_         = nullptr;
92   
        void (*fn_)(void*) = nullptr;
91   
        void (*fn_)(void*) = nullptr;
93  
92  
94   
    public:
93   
    public:
95   
        callback() = default;
94   
        callback() = default;
96   
        callback(void* ctx, void (*fn)(void*)) noexcept : ctx_(ctx), fn_(fn) {}
95   
        callback(void* ctx, void (*fn)(void*)) noexcept : ctx_(ctx), fn_(fn) {}
97  
96  
98   
        explicit operator bool() const noexcept
97   
        explicit operator bool() const noexcept
99   
        {
98   
        {
100   
            return fn_ != nullptr;
99   
            return fn_ != nullptr;
101   
        }
100   
        }
102   
        void operator()() const
101   
        void operator()() const
103   
        {
102   
        {
104   
            if (fn_)
103   
            if (fn_)
105   
                fn_(ctx_);
104   
                fn_(ctx_);
106   
        }
105   
        }
107   
    };
106   
    };
108  
107  
109   
    struct implementation;
108   
    struct implementation;
110  
109  
111   
private:
110   
private:
112   
    struct heap_entry
111   
    struct heap_entry
113   
    {
112   
    {
114   
        time_point time_;
113   
        time_point time_;
115   
        implementation* timer_;
114   
        implementation* timer_;
116   
    };
115   
    };
117  
116  
118   
    scheduler* sched_ = nullptr;
117   
    scheduler* sched_ = nullptr;
119   
    mutable std::mutex mutex_;
118   
    mutable std::mutex mutex_;
120   
    std::vector<heap_entry> heap_;
119   
    std::vector<heap_entry> heap_;
121   
    implementation* free_list_     = nullptr;
120   
    implementation* free_list_     = nullptr;
122   
    waiter_node* waiter_free_list_ = nullptr;
121   
    waiter_node* waiter_free_list_ = nullptr;
123   
    callback on_earliest_changed_;
122   
    callback on_earliest_changed_;
124   
    // Avoids mutex in nearest_expiry() and empty()
123   
    // Avoids mutex in nearest_expiry() and empty()
125   
    mutable std::atomic<std::int64_t> cached_nearest_ns_{
124   
    mutable std::atomic<std::int64_t> cached_nearest_ns_{
126   
        (std::numeric_limits<std::int64_t>::max)()};
125   
        (std::numeric_limits<std::int64_t>::max)()};
127  
126  
128   
public:
127   
public:
129   
    inline timer_service(capy::execution_context&, scheduler& sched)
128   
    inline timer_service(capy::execution_context&, scheduler& sched)
130   
        : sched_(&sched)
129   
        : sched_(&sched)
131   
    {
130   
    {
132   
    }
131   
    }
133  
132  
134   
    inline scheduler& get_scheduler() noexcept
133   
    inline scheduler& get_scheduler() noexcept
135   
    {
134   
    {
136   
        return *sched_;
135   
        return *sched_;
137   
    }
136   
    }
138  
137  
139   
    ~timer_service() override = default;
138   
    ~timer_service() override = default;
140  
139  
141   
    timer_service(timer_service const&)            = delete;
140   
    timer_service(timer_service const&)            = delete;
142   
    timer_service& operator=(timer_service const&) = delete;
141   
    timer_service& operator=(timer_service const&) = delete;
143  
142  
144   
    inline void set_on_earliest_changed(callback cb)
143   
    inline void set_on_earliest_changed(callback cb)
145   
    {
144   
    {
146   
        on_earliest_changed_ = cb;
145   
        on_earliest_changed_ = cb;
147   
    }
146   
    }
148  
147  
149   
    inline bool empty() const noexcept
148   
    inline bool empty() const noexcept
150   
    {
149   
    {
151   
        return cached_nearest_ns_.load(std::memory_order_acquire) ==
150   
        return cached_nearest_ns_.load(std::memory_order_acquire) ==
152   
            (std::numeric_limits<std::int64_t>::max)();
151   
            (std::numeric_limits<std::int64_t>::max)();
153   
    }
152   
    }
154  
153  
155   
    inline time_point nearest_expiry() const noexcept
154   
    inline time_point nearest_expiry() const noexcept
156   
    {
155   
    {
157   
        auto ns = cached_nearest_ns_.load(std::memory_order_acquire);
156   
        auto ns = cached_nearest_ns_.load(std::memory_order_acquire);
158   
        return time_point(time_point::duration(ns));
157   
        return time_point(time_point::duration(ns));
159   
    }
158   
    }
160  
159  
161   
    inline void shutdown() override;
160   
    inline void shutdown() override;
162   
    inline io_object::implementation* construct() override;
161   
    inline io_object::implementation* construct() override;
163   
    inline void destroy(io_object::implementation* p) override;
162   
    inline void destroy(io_object::implementation* p) override;
164   
    inline void destroy_impl(implementation& impl);
163   
    inline void destroy_impl(implementation& impl);
165   
    inline waiter_node* create_waiter();
164   
    inline waiter_node* create_waiter();
166   
    inline void destroy_waiter(waiter_node* w);
165   
    inline void destroy_waiter(waiter_node* w);
167   
    inline std::size_t update_timer(implementation& impl, time_point new_time);
166   
    inline std::size_t update_timer(implementation& impl, time_point new_time);
168   
    inline void insert_waiter(implementation& impl, waiter_node* w);
167   
    inline void insert_waiter(implementation& impl, waiter_node* w);
169   
    inline std::size_t cancel_timer(implementation& impl);
168   
    inline std::size_t cancel_timer(implementation& impl);
170   
    inline void cancel_waiter(waiter_node* w);
169   
    inline void cancel_waiter(waiter_node* w);
171   
    inline std::size_t cancel_one_waiter(implementation& impl);
170   
    inline std::size_t cancel_one_waiter(implementation& impl);
172   
    inline std::size_t process_expired();
171   
    inline std::size_t process_expired();
173  
172  
174   
private:
173   
private:
175   
    inline void refresh_cached_nearest() noexcept
174   
    inline void refresh_cached_nearest() noexcept
176   
    {
175   
    {
177   
        auto ns = heap_.empty() ? (std::numeric_limits<std::int64_t>::max)()
176   
        auto ns = heap_.empty() ? (std::numeric_limits<std::int64_t>::max)()
178   
                                : heap_[0].time_.time_since_epoch().count();
177   
                                : heap_[0].time_.time_since_epoch().count();
179   
        cached_nearest_ns_.store(ns, std::memory_order_release);
178   
        cached_nearest_ns_.store(ns, std::memory_order_release);
180   
    }
179   
    }
181  
180  
182   
    inline void remove_timer_impl(implementation& impl);
181   
    inline void remove_timer_impl(implementation& impl);
183   
    inline void up_heap(std::size_t index);
182   
    inline void up_heap(std::size_t index);
184   
    inline void down_heap(std::size_t index);
183   
    inline void down_heap(std::size_t index);
185   
    inline void swap_heap(std::size_t i1, std::size_t i2);
184   
    inline void swap_heap(std::size_t i1, std::size_t i2);
186   
};
185   
};
187  
186  
188   
struct BOOST_COROSIO_SYMBOL_VISIBLE waiter_node
187   
struct BOOST_COROSIO_SYMBOL_VISIBLE waiter_node
189   
    : intrusive_list<waiter_node>::node
188   
    : intrusive_list<waiter_node>::node
190   
{
189   
{
191   
    // Embedded completion op — avoids heap allocation per fire/cancel
190   
    // Embedded completion op — avoids heap allocation per fire/cancel
192   
    struct completion_op final : scheduler_op
191   
    struct completion_op final : scheduler_op
193   
    {
192   
    {
194   
        waiter_node* waiter_ = nullptr;
193   
        waiter_node* waiter_ = nullptr;
195  
194  
196   
        static void do_complete(
195   
        static void do_complete(
197   
            void* owner, scheduler_op* base, std::uint32_t, std::uint32_t);
196   
            void* owner, scheduler_op* base, std::uint32_t, std::uint32_t);
198  
197  
199   
        completion_op() noexcept : scheduler_op(&do_complete) {}
198   
        completion_op() noexcept : scheduler_op(&do_complete) {}
200  
199  
201   
        void operator()() override;
200   
        void operator()() override;
202 - 
        void destroy() override;
201 + 
        // No-op — lifetime owned by waiter_node, not the scheduler queue
 
202 + 
        void destroy() override {}
203   
    };
203   
    };
204  
204  
205   
    // Per-waiter stop_token cancellation
205   
    // Per-waiter stop_token cancellation
206   
    struct canceller
206   
    struct canceller
207   
    {
207   
    {
208   
        waiter_node* waiter_;
208   
        waiter_node* waiter_;
209   
        void operator()() const;
209   
        void operator()() const;
210   
    };
210   
    };
211  
211  
212   
    // nullptr once removed from timer's waiter list (concurrency marker)
212   
    // nullptr once removed from timer's waiter list (concurrency marker)
213   
    timer_service::implementation* impl_ = nullptr;
213   
    timer_service::implementation* impl_ = nullptr;
214   
    timer_service* svc_                  = nullptr;
214   
    timer_service* svc_                  = nullptr;
215   
    std::coroutine_handle<> h_;
215   
    std::coroutine_handle<> h_;
216   
    capy::executor_ref d_;
216   
    capy::executor_ref d_;
217   
    std::error_code* ec_out_ = nullptr;
217   
    std::error_code* ec_out_ = nullptr;
218   
    std::stop_token token_;
218   
    std::stop_token token_;
219   
    std::optional<std::stop_callback<canceller>> stop_cb_;
219   
    std::optional<std::stop_callback<canceller>> stop_cb_;
220   
    completion_op op_;
220   
    completion_op op_;
221   
    std::error_code ec_value_;
221   
    std::error_code ec_value_;
222   
    waiter_node* next_free_ = nullptr;
222   
    waiter_node* next_free_ = nullptr;
223  
223  
224   
    waiter_node() noexcept
224   
    waiter_node() noexcept
225   
    {
225   
    {
226   
        op_.waiter_ = this;
226   
        op_.waiter_ = this;
227   
    }
227   
    }
228   
};
228   
};
229  
229  
230   
struct timer_service::implementation final : timer::implementation
230   
struct timer_service::implementation final : timer::implementation
231   
{
231   
{
232   
    using clock_type = std::chrono::steady_clock;
232   
    using clock_type = std::chrono::steady_clock;
233   
    using time_point = clock_type::time_point;
233   
    using time_point = clock_type::time_point;
234   
    using duration   = clock_type::duration;
234   
    using duration   = clock_type::duration;
235  
235  
236   
    timer_service* svc_ = nullptr;
236   
    timer_service* svc_ = nullptr;
237   
    intrusive_list<waiter_node> waiters_;
237   
    intrusive_list<waiter_node> waiters_;
238  
238  
239   
    // Free list linkage (reused when impl is on free_list)
239   
    // Free list linkage (reused when impl is on free_list)
240   
    implementation* next_free_ = nullptr;
240   
    implementation* next_free_ = nullptr;
241  
241  
242   
    inline explicit implementation(timer_service& svc) noexcept;
242   
    inline explicit implementation(timer_service& svc) noexcept;
243  
243  
244   
    inline std::coroutine_handle<> wait(
244   
    inline std::coroutine_handle<> wait(
245   
        std::coroutine_handle<>,
245   
        std::coroutine_handle<>,
246   
        capy::executor_ref,
246   
        capy::executor_ref,
247   
        std::stop_token,
247   
        std::stop_token,
248   
        std::error_code*) override;
248   
        std::error_code*) override;
249   
};
249   
};
250  
250  
251   
// Thread-local caches avoid hot-path mutex acquisitions:
251   
// Thread-local caches avoid hot-path mutex acquisitions:
252   
// 1. Impl cache — single-slot, validated by comparing svc_
252   
// 1. Impl cache — single-slot, validated by comparing svc_
253   
// 2. Waiter cache — single-slot, no service affinity
253   
// 2. Waiter cache — single-slot, no service affinity
254   
// All caches are cleared by timer_service_invalidate_cache() during shutdown.
254   
// All caches are cleared by timer_service_invalidate_cache() during shutdown.
255  
255  
256   
inline thread_local_ptr<timer_service::implementation> tl_cached_impl;
256   
inline thread_local_ptr<timer_service::implementation> tl_cached_impl;
257   
inline thread_local_ptr<waiter_node> tl_cached_waiter;
257   
inline thread_local_ptr<waiter_node> tl_cached_waiter;
258  
258  
259   
inline timer_service::implementation*
259   
inline timer_service::implementation*
260   
try_pop_tl_cache(timer_service* svc) noexcept
260   
try_pop_tl_cache(timer_service* svc) noexcept
261   
{
261   
{
262   
    auto* impl = tl_cached_impl.get();
262   
    auto* impl = tl_cached_impl.get();
263   
    if (impl)
263   
    if (impl)
264   
    {
264   
    {
265   
        tl_cached_impl.set(nullptr);
265   
        tl_cached_impl.set(nullptr);
266   
        if (impl->svc_ == svc)
266   
        if (impl->svc_ == svc)
267   
            return impl;
267   
            return impl;
268   
        // Stale impl from a destroyed service
268   
        // Stale impl from a destroyed service
269   
        delete impl;
269   
        delete impl;
270   
    }
270   
    }
271   
    return nullptr;
271   
    return nullptr;
272   
}
272   
}
273  
273  
274   
inline bool
274   
inline bool
275   
try_push_tl_cache(timer_service::implementation* impl) noexcept
275   
try_push_tl_cache(timer_service::implementation* impl) noexcept
276   
{
276   
{
277   
    if (!tl_cached_impl.get())
277   
    if (!tl_cached_impl.get())
278   
    {
278   
    {
279   
        tl_cached_impl.set(impl);
279   
        tl_cached_impl.set(impl);
280   
        return true;
280   
        return true;
281   
    }
281   
    }
282   
    return false;
282   
    return false;
283   
}
283   
}
284  
284  
285   
inline waiter_node*
285   
inline waiter_node*
286   
try_pop_waiter_tl_cache() noexcept
286   
try_pop_waiter_tl_cache() noexcept
287   
{
287   
{
288   
    auto* w = tl_cached_waiter.get();
288   
    auto* w = tl_cached_waiter.get();
289   
    if (w)
289   
    if (w)
290   
    {
290   
    {
291   
        tl_cached_waiter.set(nullptr);
291   
        tl_cached_waiter.set(nullptr);
292   
        return w;
292   
        return w;
293   
    }
293   
    }
294   
    return nullptr;
294   
    return nullptr;
295   
}
295   
}
296  
296  
297   
inline bool
297   
inline bool
298   
try_push_waiter_tl_cache(waiter_node* w) noexcept
298   
try_push_waiter_tl_cache(waiter_node* w) noexcept
299   
{
299   
{
300   
    if (!tl_cached_waiter.get())
300   
    if (!tl_cached_waiter.get())
301   
    {
301   
    {
302   
        tl_cached_waiter.set(w);
302   
        tl_cached_waiter.set(w);
303   
        return true;
303   
        return true;
304   
    }
304   
    }
305   
    return false;
305   
    return false;
306   
}
306   
}
307  
307  
308   
inline void
308   
inline void
309   
timer_service_invalidate_cache() noexcept
309   
timer_service_invalidate_cache() noexcept
310   
{
310   
{
311   
    delete tl_cached_impl.get();
311   
    delete tl_cached_impl.get();
312   
    tl_cached_impl.set(nullptr);
312   
    tl_cached_impl.set(nullptr);
313  
313  
314   
    delete tl_cached_waiter.get();
314   
    delete tl_cached_waiter.get();
315   
    tl_cached_waiter.set(nullptr);
315   
    tl_cached_waiter.set(nullptr);
316   
}
316   
}
317  
317  
318   
// timer_service out-of-class member function definitions
318   
// timer_service out-of-class member function definitions
319  
319  
320   
inline timer_service::implementation::implementation(
320   
inline timer_service::implementation::implementation(
321   
    timer_service& svc) noexcept
321   
    timer_service& svc) noexcept
322   
    : svc_(&svc)
322   
    : svc_(&svc)
323   
{
323   
{
324   
}
324   
}
325  
325  
326   
inline void
326   
inline void
327   
timer_service::shutdown()
327   
timer_service::shutdown()
328   
{
328   
{
329   
    timer_service_invalidate_cache();
329   
    timer_service_invalidate_cache();
330  
330  
331 - 
    // Cancel waiting timers still in the heap.
331 + 
    // Cancel waiting timers still in the heap
332 - 
    // Each waiter called work_started() in implementation::wait().
 
333 - 
    // On IOCP the scheduler shutdown loop exits when outstanding_work_
 
334 - 
    // reaches zero, so we must call work_finished() here to balance it.
 
335 - 
    // On other backends this is harmless (their drain loops exit when
 
336 - 
    // the queue is empty, not based on outstanding_work_).
 
337   
    for (auto& entry : heap_)
332   
    for (auto& entry : heap_)
338   
    {
333   
    {
339   
        auto* impl = entry.timer_;
334   
        auto* impl = entry.timer_;
340   
        while (auto* w = impl->waiters_.pop_front())
335   
        while (auto* w = impl->waiters_.pop_front())
341   
        {
336   
        {
342   
            w->stop_cb_.reset();
337   
            w->stop_cb_.reset();
343 - 
            auto h = std::exchange(w->h_, {});
338 + 
            w->h_ = {};
344 - 
            if (h)
 
345 - 
                h.destroy();
 
346   
            sched_->work_finished();
339   
            sched_->work_finished();
347   
            delete w;
340   
            delete w;
348   
        }
341   
        }
349   
        impl->heap_index_ = (std::numeric_limits<std::size_t>::max)();
342   
        impl->heap_index_ = (std::numeric_limits<std::size_t>::max)();
350   
        delete impl;
343   
        delete impl;
351   
    }
344   
    }
352   
    heap_.clear();
345   
    heap_.clear();
353   
    cached_nearest_ns_.store(
346   
    cached_nearest_ns_.store(
354   
        (std::numeric_limits<std::int64_t>::max)(), std::memory_order_release);
347   
        (std::numeric_limits<std::int64_t>::max)(), std::memory_order_release);
355  
348  
356   
    // Delete free-listed impls
349   
    // Delete free-listed impls
357   
    while (free_list_)
350   
    while (free_list_)
358   
    {
351   
    {
359   
        auto* next = free_list_->next_free_;
352   
        auto* next = free_list_->next_free_;
360   
        delete free_list_;
353   
        delete free_list_;
361   
        free_list_ = next;
354   
        free_list_ = next;
362   
    }
355   
    }
363  
356  
364   
    // Delete free-listed waiters
357   
    // Delete free-listed waiters
365   
    while (waiter_free_list_)
358   
    while (waiter_free_list_)
366   
    {
359   
    {
367   
        auto* next = waiter_free_list_->next_free_;
360   
        auto* next = waiter_free_list_->next_free_;
368   
        delete waiter_free_list_;
361   
        delete waiter_free_list_;
369   
        waiter_free_list_ = next;
362   
        waiter_free_list_ = next;
370   
    }
363   
    }
371   
}
364   
}
372  
365  
373   
inline io_object::implementation*
366   
inline io_object::implementation*
374   
timer_service::construct()
367   
timer_service::construct()
375   
{
368   
{
376   
    implementation* impl = try_pop_tl_cache(this);
369   
    implementation* impl = try_pop_tl_cache(this);
377   
    if (impl)
370   
    if (impl)
378   
    {
371   
    {
379   
        impl->svc_        = this;
372   
        impl->svc_        = this;
380   
        impl->heap_index_ = (std::numeric_limits<std::size_t>::max)();
373   
        impl->heap_index_ = (std::numeric_limits<std::size_t>::max)();
381   
        impl->might_have_pending_waits_ = false;
374   
        impl->might_have_pending_waits_ = false;
382   
        return impl;
375   
        return impl;
383   
    }
376   
    }
384  
377  
385   
    std::lock_guard lock(mutex_);
378   
    std::lock_guard lock(mutex_);
386   
    if (free_list_)
379   
    if (free_list_)
387   
    {
380   
    {
388   
        impl              = free_list_;
381   
        impl              = free_list_;
389   
        free_list_        = impl->next_free_;
382   
        free_list_        = impl->next_free_;
390   
        impl->next_free_  = nullptr;
383   
        impl->next_free_  = nullptr;
391   
        impl->svc_        = this;
384   
        impl->svc_        = this;
392   
        impl->heap_index_ = (std::numeric_limits<std::size_t>::max)();
385   
        impl->heap_index_ = (std::numeric_limits<std::size_t>::max)();
393   
        impl->might_have_pending_waits_ = false;
386   
        impl->might_have_pending_waits_ = false;
394   
    }
387   
    }
395   
    else
388   
    else
396   
    {
389   
    {
397   
        impl = new implementation(*this);
390   
        impl = new implementation(*this);
398   
    }
391   
    }
399   
    return impl;
392   
    return impl;
400   
}
393   
}
401  
394  
402   
inline void
395   
inline void
403   
timer_service::destroy(io_object::implementation* p)
396   
timer_service::destroy(io_object::implementation* p)
404   
{
397   
{
405   
    destroy_impl(static_cast<implementation&>(*p));
398   
    destroy_impl(static_cast<implementation&>(*p));
406   
}
399   
}
407  
400  
408   
inline void
401   
inline void
409   
timer_service::destroy_impl(implementation& impl)
402   
timer_service::destroy_impl(implementation& impl)
410   
{
403   
{
411   
    cancel_timer(impl);
404   
    cancel_timer(impl);
412  
405  
413   
    if (impl.heap_index_ != (std::numeric_limits<std::size_t>::max)())
406   
    if (impl.heap_index_ != (std::numeric_limits<std::size_t>::max)())
414   
    {
407   
    {
415   
        std::lock_guard lock(mutex_);
408   
        std::lock_guard lock(mutex_);
416   
        remove_timer_impl(impl);
409   
        remove_timer_impl(impl);
417   
        refresh_cached_nearest();
410   
        refresh_cached_nearest();
418   
    }
411   
    }
419  
412  
420   
    if (try_push_tl_cache(&impl))
413   
    if (try_push_tl_cache(&impl))
421   
        return;
414   
        return;
422  
415  
423   
    std::lock_guard lock(mutex_);
416   
    std::lock_guard lock(mutex_);
424   
    impl.next_free_ = free_list_;
417   
    impl.next_free_ = free_list_;
425   
    free_list_      = &impl;
418   
    free_list_      = &impl;
426   
}
419   
}
427  
420  
428   
inline waiter_node*
421   
inline waiter_node*
429   
timer_service::create_waiter()
422   
timer_service::create_waiter()
430   
{
423   
{
431   
    if (auto* w = try_pop_waiter_tl_cache())
424   
    if (auto* w = try_pop_waiter_tl_cache())
432   
        return w;
425   
        return w;
433  
426  
434   
    std::lock_guard lock(mutex_);
427   
    std::lock_guard lock(mutex_);
435   
    if (waiter_free_list_)
428   
    if (waiter_free_list_)
436   
    {
429   
    {
437   
        auto* w           = waiter_free_list_;
430   
        auto* w           = waiter_free_list_;
438   
        waiter_free_list_ = w->next_free_;
431   
        waiter_free_list_ = w->next_free_;
439   
        w->next_free_     = nullptr;
432   
        w->next_free_     = nullptr;
440   
        return w;
433   
        return w;
441   
    }
434   
    }
442  
435  
443   
    return new waiter_node();
436   
    return new waiter_node();
444   
}
437   
}
445  
438  
446   
inline void
439   
inline void
447   
timer_service::destroy_waiter(waiter_node* w)
440   
timer_service::destroy_waiter(waiter_node* w)
448   
{
441   
{
449   
    if (try_push_waiter_tl_cache(w))
442   
    if (try_push_waiter_tl_cache(w))
450   
        return;
443   
        return;
451  
444  
452   
    std::lock_guard lock(mutex_);
445   
    std::lock_guard lock(mutex_);
453   
    w->next_free_     = waiter_free_list_;
446   
    w->next_free_     = waiter_free_list_;
454   
    waiter_free_list_ = w;
447   
    waiter_free_list_ = w;
455   
}
448   
}
456  
449  
457   
inline std::size_t
450   
inline std::size_t
458   
timer_service::update_timer(implementation& impl, time_point new_time)
451   
timer_service::update_timer(implementation& impl, time_point new_time)
459   
{
452   
{
460   
    bool in_heap =
453   
    bool in_heap =
461   
        (impl.heap_index_ != (std::numeric_limits<std::size_t>::max)());
454   
        (impl.heap_index_ != (std::numeric_limits<std::size_t>::max)());
462   
    if (!in_heap && impl.waiters_.empty())
455   
    if (!in_heap && impl.waiters_.empty())
463   
        return 0;
456   
        return 0;
464  
457  
465   
    bool notify = false;
458   
    bool notify = false;
466   
    intrusive_list<waiter_node> canceled;
459   
    intrusive_list<waiter_node> canceled;
467  
460  
468   
    {
461   
    {
469   
        std::lock_guard lock(mutex_);
462   
        std::lock_guard lock(mutex_);
470  
463  
471   
        while (auto* w = impl.waiters_.pop_front())
464   
        while (auto* w = impl.waiters_.pop_front())
472   
        {
465   
        {
473   
            w->impl_ = nullptr;
466   
            w->impl_ = nullptr;
474   
            canceled.push_back(w);
467   
            canceled.push_back(w);
475   
        }
468   
        }
476  
469  
477   
        if (impl.heap_index_ < heap_.size())
470   
        if (impl.heap_index_ < heap_.size())
478   
        {
471   
        {
479   
            time_point old_time           = heap_[impl.heap_index_].time_;
472   
            time_point old_time           = heap_[impl.heap_index_].time_;
480   
            heap_[impl.heap_index_].time_ = new_time;
473   
            heap_[impl.heap_index_].time_ = new_time;
481  
474  
482   
            if (new_time < old_time)
475   
            if (new_time < old_time)
483   
                up_heap(impl.heap_index_);
476   
                up_heap(impl.heap_index_);
484   
            else
477   
            else
485   
                down_heap(impl.heap_index_);
478   
                down_heap(impl.heap_index_);
486  
479  
487   
            notify = (impl.heap_index_ == 0);
480   
            notify = (impl.heap_index_ == 0);
488   
        }
481   
        }
489  
482  
490   
        refresh_cached_nearest();
483   
        refresh_cached_nearest();
491   
    }
484   
    }
492  
485  
493   
    std::size_t count = 0;
486   
    std::size_t count = 0;
494   
    while (auto* w = canceled.pop_front())
487   
    while (auto* w = canceled.pop_front())
495   
    {
488   
    {
496   
        w->ec_value_ = make_error_code(capy::error::canceled);
489   
        w->ec_value_ = make_error_code(capy::error::canceled);
497   
        sched_->post(&w->op_);
490   
        sched_->post(&w->op_);
498   
        ++count;
491   
        ++count;
499   
    }
492   
    }
500  
493  
501   
    if (notify)
494   
    if (notify)
502   
        on_earliest_changed_();
495   
        on_earliest_changed_();
503  
496  
504   
    return count;
497   
    return count;
505   
}
498   
}
506  
499  
507   
inline void
500   
inline void
508   
timer_service::insert_waiter(implementation& impl, waiter_node* w)
501   
timer_service::insert_waiter(implementation& impl, waiter_node* w)
509   
{
502   
{
510   
    bool notify = false;
503   
    bool notify = false;
511   
    {
504   
    {
512   
        std::lock_guard lock(mutex_);
505   
        std::lock_guard lock(mutex_);
513   
        if (impl.heap_index_ == (std::numeric_limits<std::size_t>::max)())
506   
        if (impl.heap_index_ == (std::numeric_limits<std::size_t>::max)())
514   
        {
507   
        {
515   
            impl.heap_index_ = heap_.size();
508   
            impl.heap_index_ = heap_.size();
516   
            heap_.push_back({impl.expiry_, &impl});
509   
            heap_.push_back({impl.expiry_, &impl});
517   
            up_heap(heap_.size() - 1);
510   
            up_heap(heap_.size() - 1);
518   
            notify = (impl.heap_index_ == 0);
511   
            notify = (impl.heap_index_ == 0);
519   
            refresh_cached_nearest();
512   
            refresh_cached_nearest();
520   
        }
513   
        }
521   
        impl.waiters_.push_back(w);
514   
        impl.waiters_.push_back(w);
522   
    }
515   
    }
523   
    if (notify)
516   
    if (notify)
524   
        on_earliest_changed_();
517   
        on_earliest_changed_();
525   
}
518   
}
526  
519  
527   
inline std::size_t
520   
inline std::size_t
528   
timer_service::cancel_timer(implementation& impl)
521   
timer_service::cancel_timer(implementation& impl)
529   
{
522   
{
530   
    if (!impl.might_have_pending_waits_)
523   
    if (!impl.might_have_pending_waits_)
531   
        return 0;
524   
        return 0;
532  
525  
533   
    // Not in heap and no waiters — just clear the flag
526   
    // Not in heap and no waiters — just clear the flag
534   
    if (impl.heap_index_ == (std::numeric_limits<std::size_t>::max)() &&
527   
    if (impl.heap_index_ == (std::numeric_limits<std::size_t>::max)() &&
535   
        impl.waiters_.empty())
528   
        impl.waiters_.empty())
536   
    {
529   
    {
537   
        impl.might_have_pending_waits_ = false;
530   
        impl.might_have_pending_waits_ = false;
538   
        return 0;
531   
        return 0;
539   
    }
532   
    }
540  
533  
541   
    intrusive_list<waiter_node> canceled;
534   
    intrusive_list<waiter_node> canceled;
542  
535  
543   
    {
536   
    {
544   
        std::lock_guard lock(mutex_);
537   
        std::lock_guard lock(mutex_);
545   
        remove_timer_impl(impl);
538   
        remove_timer_impl(impl);
546   
        while (auto* w = impl.waiters_.pop_front())
539   
        while (auto* w = impl.waiters_.pop_front())
547   
        {
540   
        {
548   
            w->impl_ = nullptr;
541   
            w->impl_ = nullptr;
549   
            canceled.push_back(w);
542   
            canceled.push_back(w);
550   
        }
543   
        }
551   
        refresh_cached_nearest();
544   
        refresh_cached_nearest();
552   
    }
545   
    }
553  
546  
554   
    impl.might_have_pending_waits_ = false;
547   
    impl.might_have_pending_waits_ = false;
555  
548  
556   
    std::size_t count = 0;
549   
    std::size_t count = 0;
557   
    while (auto* w = canceled.pop_front())
550   
    while (auto* w = canceled.pop_front())
558   
    {
551   
    {
559   
        w->ec_value_ = make_error_code(capy::error::canceled);
552   
        w->ec_value_ = make_error_code(capy::error::canceled);
560   
        sched_->post(&w->op_);
553   
        sched_->post(&w->op_);
561   
        ++count;
554   
        ++count;
562   
    }
555   
    }
563  
556  
564   
    return count;
557   
    return count;
565   
}
558   
}
566  
559  
567   
inline void
560   
inline void
568   
timer_service::cancel_waiter(waiter_node* w)
561   
timer_service::cancel_waiter(waiter_node* w)
569   
{
562   
{
570   
    {
563   
    {
571   
        std::lock_guard lock(mutex_);
564   
        std::lock_guard lock(mutex_);
572   
        // Already removed by cancel_timer or process_expired
565   
        // Already removed by cancel_timer or process_expired
573   
        if (!w->impl_)
566   
        if (!w->impl_)
574   
            return;
567   
            return;
575   
        auto* impl = w->impl_;
568   
        auto* impl = w->impl_;
576   
        w->impl_   = nullptr;
569   
        w->impl_   = nullptr;
577   
        impl->waiters_.remove(w);
570   
        impl->waiters_.remove(w);
578   
        if (impl->waiters_.empty())
571   
        if (impl->waiters_.empty())
579   
        {
572   
        {
580   
            remove_timer_impl(*impl);
573   
            remove_timer_impl(*impl);
581   
            impl->might_have_pending_waits_ = false;
574   
            impl->might_have_pending_waits_ = false;
582   
        }
575   
        }
583   
        refresh_cached_nearest();
576   
        refresh_cached_nearest();
584   
    }
577   
    }
585  
578  
586   
    w->ec_value_ = make_error_code(capy::error::canceled);
579   
    w->ec_value_ = make_error_code(capy::error::canceled);
587   
    sched_->post(&w->op_);
580   
    sched_->post(&w->op_);
588   
}
581   
}
589  
582  
590   
inline std::size_t
583   
inline std::size_t
591   
timer_service::cancel_one_waiter(implementation& impl)
584   
timer_service::cancel_one_waiter(implementation& impl)
592   
{
585   
{
593   
    if (!impl.might_have_pending_waits_)
586   
    if (!impl.might_have_pending_waits_)
594   
        return 0;
587   
        return 0;
595  
588  
596   
    waiter_node* w = nullptr;
589   
    waiter_node* w = nullptr;
597  
590  
598   
    {
591   
    {
599   
        std::lock_guard lock(mutex_);
592   
        std::lock_guard lock(mutex_);
600   
        w = impl.waiters_.pop_front();
593   
        w = impl.waiters_.pop_front();
601   
        if (!w)
594   
        if (!w)
602   
            return 0;
595   
            return 0;
603   
        w->impl_ = nullptr;
596   
        w->impl_ = nullptr;
604   
        if (impl.waiters_.empty())
597   
        if (impl.waiters_.empty())
605   
        {
598   
        {
606   
            remove_timer_impl(impl);
599   
            remove_timer_impl(impl);
607   
            impl.might_have_pending_waits_ = false;
600   
            impl.might_have_pending_waits_ = false;
608   
        }
601   
        }
609   
        refresh_cached_nearest();
602   
        refresh_cached_nearest();
610   
    }
603   
    }
611  
604  
612   
    w->ec_value_ = make_error_code(capy::error::canceled);
605   
    w->ec_value_ = make_error_code(capy::error::canceled);
613   
    sched_->post(&w->op_);
606   
    sched_->post(&w->op_);
614   
    return 1;
607   
    return 1;
615   
}
608   
}
616  
609  
617   
inline std::size_t
610   
inline std::size_t
618   
timer_service::process_expired()
611   
timer_service::process_expired()
619   
{
612   
{
620   
    intrusive_list<waiter_node> expired;
613   
    intrusive_list<waiter_node> expired;
621  
614  
622   
    {
615   
    {
623   
        std::lock_guard lock(mutex_);
616   
        std::lock_guard lock(mutex_);
624   
        auto now = clock_type::now();
617   
        auto now = clock_type::now();
625  
618  
626   
        while (!heap_.empty() && heap_[0].time_ <= now)
619   
        while (!heap_.empty() && heap_[0].time_ <= now)
627   
        {
620   
        {
628   
            implementation* t = heap_[0].timer_;
621   
            implementation* t = heap_[0].timer_;
629   
            remove_timer_impl(*t);
622   
            remove_timer_impl(*t);
630   
            while (auto* w = t->waiters_.pop_front())
623   
            while (auto* w = t->waiters_.pop_front())
631   
            {
624   
            {
632   
                w->impl_     = nullptr;
625   
                w->impl_     = nullptr;
633   
                w->ec_value_ = {};
626   
                w->ec_value_ = {};
634   
                expired.push_back(w);
627   
                expired.push_back(w);
635   
            }
628   
            }
636   
            t->might_have_pending_waits_ = false;
629   
            t->might_have_pending_waits_ = false;
637   
        }
630   
        }
638  
631  
639   
        refresh_cached_nearest();
632   
        refresh_cached_nearest();
640   
    }
633   
    }
641  
634  
642   
    std::size_t count = 0;
635   
    std::size_t count = 0;
643   
    while (auto* w = expired.pop_front())
636   
    while (auto* w = expired.pop_front())
644   
    {
637   
    {
645   
        sched_->post(&w->op_);
638   
        sched_->post(&w->op_);
646   
        ++count;
639   
        ++count;
647   
    }
640   
    }
648  
641  
649   
    return count;
642   
    return count;
650   
}
643   
}
651  
644  
652   
inline void
645   
inline void
653   
timer_service::remove_timer_impl(implementation& impl)
646   
timer_service::remove_timer_impl(implementation& impl)
654   
{
647   
{
655   
    std::size_t index = impl.heap_index_;
648   
    std::size_t index = impl.heap_index_;
656   
    if (index >= heap_.size())
649   
    if (index >= heap_.size())
657   
        return; // Not in heap
650   
        return; // Not in heap
658  
651  
659   
    if (index == heap_.size() - 1)
652   
    if (index == heap_.size() - 1)
660   
    {
653   
    {
661   
        // Last element, just pop
654   
        // Last element, just pop
662   
        impl.heap_index_ = (std::numeric_limits<std::size_t>::max)();
655   
        impl.heap_index_ = (std::numeric_limits<std::size_t>::max)();
663   
        heap_.pop_back();
656   
        heap_.pop_back();
664   
    }
657   
    }
665   
    else
658   
    else
666   
    {
659   
    {
667   
        // Swap with last and reheapify
660   
        // Swap with last and reheapify
668   
        swap_heap(index, heap_.size() - 1);
661   
        swap_heap(index, heap_.size() - 1);
669   
        impl.heap_index_ = (std::numeric_limits<std::size_t>::max)();
662   
        impl.heap_index_ = (std::numeric_limits<std::size_t>::max)();
670   
        heap_.pop_back();
663   
        heap_.pop_back();
671  
664  
672   
        if (index > 0 && heap_[index].time_ < heap_[(index - 1) / 2].time_)
665   
        if (index > 0 && heap_[index].time_ < heap_[(index - 1) / 2].time_)
673   
            up_heap(index);
666   
            up_heap(index);
674   
        else
667   
        else
675   
            down_heap(index);
668   
            down_heap(index);
676   
    }
669   
    }
677   
}
670   
}
678  
671  
679   
inline void
672   
inline void
680   
timer_service::up_heap(std::size_t index)
673   
timer_service::up_heap(std::size_t index)
681   
{
674   
{
682   
    while (index > 0)
675   
    while (index > 0)
683   
    {
676   
    {
684   
        std::size_t parent = (index - 1) / 2;
677   
        std::size_t parent = (index - 1) / 2;
685   
        if (!(heap_[index].time_ < heap_[parent].time_))
678   
        if (!(heap_[index].time_ < heap_[parent].time_))
686   
            break;
679   
            break;
687   
        swap_heap(index, parent);
680   
        swap_heap(index, parent);
688   
        index = parent;
681   
        index = parent;
689   
    }
682   
    }
690   
}
683   
}
691  
684  
692   
inline void
685   
inline void
693   
timer_service::down_heap(std::size_t index)
686   
timer_service::down_heap(std::size_t index)
694   
{
687   
{
695   
    std::size_t child = index * 2 + 1;
688   
    std::size_t child = index * 2 + 1;
696   
    while (child < heap_.size())
689   
    while (child < heap_.size())
697   
    {
690   
    {
698   
        std::size_t min_child = (child + 1 == heap_.size() ||
691   
        std::size_t min_child = (child + 1 == heap_.size() ||
699   
                                 heap_[child].time_ < heap_[child + 1].time_)
692   
                                 heap_[child].time_ < heap_[child + 1].time_)
700   
            ? child
693   
            ? child
701   
            : child + 1;
694   
            : child + 1;
702  
695  
703   
        if (heap_[index].time_ < heap_[min_child].time_)
696   
        if (heap_[index].time_ < heap_[min_child].time_)
704   
            break;
697   
            break;
705  
698  
706   
        swap_heap(index, min_child);
699   
        swap_heap(index, min_child);
707   
        index = min_child;
700   
        index = min_child;
708   
        child = index * 2 + 1;
701   
        child = index * 2 + 1;
709   
    }
702   
    }
710   
}
703   
}
711  
704  
712   
inline void
705   
inline void
713   
timer_service::swap_heap(std::size_t i1, std::size_t i2)
706   
timer_service::swap_heap(std::size_t i1, std::size_t i2)
714   
{
707   
{
715   
    heap_entry tmp                = heap_[i1];
708   
    heap_entry tmp                = heap_[i1];
716   
    heap_[i1]                     = heap_[i2];
709   
    heap_[i1]                     = heap_[i2];
717   
    heap_[i2]                     = tmp;
710   
    heap_[i2]                     = tmp;
718   
    heap_[i1].timer_->heap_index_ = i1;
711   
    heap_[i1].timer_->heap_index_ = i1;
719   
    heap_[i2].timer_->heap_index_ = i2;
712   
    heap_[i2].timer_->heap_index_ = i2;
720   
}
713   
}
721  
714  
722   
// waiter_node out-of-class member function definitions
715   
// waiter_node out-of-class member function definitions
723  
716  
724   
inline void
717   
inline void
725   
waiter_node::canceller::operator()() const
718   
waiter_node::canceller::operator()() const
726   
{
719   
{
727   
    waiter_->svc_->cancel_waiter(waiter_);
720   
    waiter_->svc_->cancel_waiter(waiter_);
728   
}
721   
}
729  
722  
730   
inline void
723   
inline void
731   
waiter_node::completion_op::do_complete(
724   
waiter_node::completion_op::do_complete(
732 - 
    [[maybe_unused]] void* owner, scheduler_op* base, std::uint32_t, std::uint32_t)
725 + 
    void* owner, scheduler_op* base, std::uint32_t, std::uint32_t)
733   
{
726   
{
734 - 
    // owner is always non-null here. The destroy path (owner == nullptr)
727 + 
    if (!owner)
735 - 
    // is unreachable because completion_op overrides destroy() directly,
728 + 
        return;
736 - 
    // bypassing scheduler_op::destroy() which would call func_(nullptr, ...).
 
737 - 
    BOOST_COROSIO_ASSERT(owner);
 
738   
    static_cast<completion_op*>(base)->operator()();
729   
    static_cast<completion_op*>(base)->operator()();
739   
}
730   
}
740  
731  
741   
inline void
732   
inline void
742   
waiter_node::completion_op::operator()()
733   
waiter_node::completion_op::operator()()
743   
{
734   
{
744   
    auto* w = waiter_;
735   
    auto* w = waiter_;
745   
    w->stop_cb_.reset();
736   
    w->stop_cb_.reset();
746   
    if (w->ec_out_)
737   
    if (w->ec_out_)
747   
        *w->ec_out_ = w->ec_value_;
738   
        *w->ec_out_ = w->ec_value_;
748  
739  
749   
    auto h      = w->h_;
740   
    auto h      = w->h_;
750   
    auto d      = w->d_;
741   
    auto d      = w->d_;
751   
    auto* svc   = w->svc_;
742   
    auto* svc   = w->svc_;
752   
    auto& sched = svc->get_scheduler();
743   
    auto& sched = svc->get_scheduler();
753  
744  
754   
    svc->destroy_waiter(w);
745   
    svc->destroy_waiter(w);
755  
746  
756   
    d.post(h);
747   
    d.post(h);
757 - 
}
 
758 -
 
759 - 
inline void
 
760 - 
waiter_node::completion_op::destroy()
 
761 - 
{
 
762 - 
    // Called during scheduler shutdown drain when this completion_op is
 
763 - 
    // in the scheduler's ready queue (posted by cancel_timer() or
 
764 - 
    // process_expired()). Balances the work_started() from
 
765 - 
    // implementation::wait(). The scheduler drain loop separately
 
766 - 
    // balances the work_started() from post(). On IOCP both decrements
 
767 - 
    // are required for outstanding_work_ to reach zero; on other
 
768 - 
    // backends this is harmless.
 
769 - 
    //
 
770 - 
    // This override also prevents scheduler_op::destroy() from calling
 
771 - 
    // do_complete(nullptr, ...). See also: timer_service::shutdown()
 
772 - 
    // which drains waiters still in the timer heap (the other path).
 
773 - 
    auto* w = waiter_;
 
774 - 
    w->stop_cb_.reset();
 
775 - 
    auto h = std::exchange(w->h_, {});
 
776 - 
    auto& sched = w->svc_->get_scheduler();
 
777 - 
    delete w;
 
778 - 
    sched.work_finished();
 
779 - 
    if (h)
 
780 - 
        h.destroy();
 
781   
    sched.work_finished();
748   
    sched.work_finished();
782   
}
749   
}
783  
750  
784   
inline std::coroutine_handle<>
751   
inline std::coroutine_handle<>
785   
timer_service::implementation::wait(
752   
timer_service::implementation::wait(
786   
    std::coroutine_handle<> h,
753   
    std::coroutine_handle<> h,
787   
    capy::executor_ref d,
754   
    capy::executor_ref d,
788   
    std::stop_token token,
755   
    std::stop_token token,
789   
    std::error_code* ec)
756   
    std::error_code* ec)
790   
{
757   
{
791   
    // Already-expired fast path — no waiter_node, no mutex.
758   
    // Already-expired fast path — no waiter_node, no mutex.
792   
    // Post instead of dispatch so the coroutine yields to the
759   
    // Post instead of dispatch so the coroutine yields to the
793   
    // scheduler, allowing other queued work to run.
760   
    // scheduler, allowing other queued work to run.
794   
    if (heap_index_ == (std::numeric_limits<std::size_t>::max)())
761   
    if (heap_index_ == (std::numeric_limits<std::size_t>::max)())
795   
    {
762   
    {
796   
        if (expiry_ == (time_point::min)() || expiry_ <= clock_type::now())
763   
        if (expiry_ == (time_point::min)() || expiry_ <= clock_type::now())
797   
        {
764   
        {
798   
            if (ec)
765   
            if (ec)
799   
                *ec = {};
766   
                *ec = {};
800   
            d.post(h);
767   
            d.post(h);
801   
            return std::noop_coroutine();
768   
            return std::noop_coroutine();
802   
        }
769   
        }
803   
    }
770   
    }
804  
771  
805   
    auto* w    = svc_->create_waiter();
772   
    auto* w    = svc_->create_waiter();
806   
    w->impl_   = this;
773   
    w->impl_   = this;
807   
    w->svc_    = svc_;
774   
    w->svc_    = svc_;
808   
    w->h_      = h;
775   
    w->h_      = h;
809   
    w->d_      = d;
776   
    w->d_      = d;
810   
    w->token_  = std::move(token);
777   
    w->token_  = std::move(token);
811   
    w->ec_out_ = ec;
778   
    w->ec_out_ = ec;
812  
779  
813   
    svc_->insert_waiter(*this, w);
780   
    svc_->insert_waiter(*this, w);
814   
    might_have_pending_waits_ = true;
781   
    might_have_pending_waits_ = true;
815   
    svc_->get_scheduler().work_started();
782   
    svc_->get_scheduler().work_started();
816  
783  
817   
    if (w->token_.stop_possible())
784   
    if (w->token_.stop_possible())
818   
        w->stop_cb_.emplace(w->token_, waiter_node::canceller{w});
785   
        w->stop_cb_.emplace(w->token_, waiter_node::canceller{w});
819  
786  
820   
    return std::noop_coroutine();
787   
    return std::noop_coroutine();
821   
}
788   
}
822  
789  
823   
// Free functions
790   
// Free functions
824  
791  
825   
struct timer_service_access
792   
struct timer_service_access
826   
{
793   
{
827   
    static native_scheduler& get_scheduler(io_context& ctx) noexcept
794   
    static native_scheduler& get_scheduler(io_context& ctx) noexcept
828   
    {
795   
    {
829   
        return static_cast<native_scheduler&>(*ctx.sched_);
796   
        return static_cast<native_scheduler&>(*ctx.sched_);
830   
    }
797   
    }
831   
};
798   
};
832  
799  
833   
// Bypass find_service() mutex by reading the scheduler's cached pointer
800   
// Bypass find_service() mutex by reading the scheduler's cached pointer
834   
inline io_object::io_service&
801   
inline io_object::io_service&
835   
timer_service_direct(capy::execution_context& ctx) noexcept
802   
timer_service_direct(capy::execution_context& ctx) noexcept
836   
{
803   
{
837   
    return *timer_service_access::get_scheduler(static_cast<io_context&>(ctx))
804   
    return *timer_service_access::get_scheduler(static_cast<io_context&>(ctx))
838   
                .timer_svc_;
805   
                .timer_svc_;
839   
}
806   
}
840  
807  
841   
inline std::size_t
808   
inline std::size_t
842   
timer_service_update_expiry(timer::implementation& base)
809   
timer_service_update_expiry(timer::implementation& base)
843   
{
810   
{
844   
    auto& impl = static_cast<timer_service::implementation&>(base);
811   
    auto& impl = static_cast<timer_service::implementation&>(base);
845   
    return impl.svc_->update_timer(impl, impl.expiry_);
812   
    return impl.svc_->update_timer(impl, impl.expiry_);
846   
}
813   
}
847  
814  
848   
inline std::size_t
815   
inline std::size_t
849   
timer_service_cancel(timer::implementation& base) noexcept
816   
timer_service_cancel(timer::implementation& base) noexcept
850   
{
817   
{
851   
    auto& impl = static_cast<timer_service::implementation&>(base);
818   
    auto& impl = static_cast<timer_service::implementation&>(base);
852   
    return impl.svc_->cancel_timer(impl);
819   
    return impl.svc_->cancel_timer(impl);
853   
}
820   
}
854  
821  
855   
inline std::size_t
822   
inline std::size_t
856   
timer_service_cancel_one(timer::implementation& base) noexcept
823   
timer_service_cancel_one(timer::implementation& base) noexcept
857   
{
824   
{
858   
    auto& impl = static_cast<timer_service::implementation&>(base);
825   
    auto& impl = static_cast<timer_service::implementation&>(base);
859   
    return impl.svc_->cancel_one_waiter(impl);
826   
    return impl.svc_->cancel_one_waiter(impl);
860   
}
827   
}
861  
828  
862   
inline timer_service&
829   
inline timer_service&
863   
get_timer_service(capy::execution_context& ctx, scheduler& sched)
830   
get_timer_service(capy::execution_context& ctx, scheduler& sched)
864   
{
831   
{
865   
    return ctx.make_service<timer_service>(sched);
832   
    return ctx.make_service<timer_service>(sched);
866   
}
833   
}
867  
834  
868   
} // namespace boost::corosio::detail
835   
} // namespace boost::corosio::detail
869  
836  
870   
#endif
837   
#endif