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