. /include/boost/corosio /detail /timer_service.hpp
72.30% Lines (214/296) 84.38% Functions (27/32)
[ ] prev/next file n p prev/next uncovered line
TLA Baseline Branch
Line Hits Code Line Hits Code
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/detail/timer.hpp> 14   #include <boost/corosio/detail/timer.hpp>
15   #include <boost/corosio/detail/scheduler.hpp> 15   #include <boost/corosio/detail/scheduler.hpp>
16   #include <boost/corosio/detail/scheduler_op.hpp> 16   #include <boost/corosio/detail/scheduler_op.hpp>
17   #include <boost/corosio/detail/intrusive.hpp> 17   #include <boost/corosio/detail/intrusive.hpp>
18   #include <boost/corosio/detail/thread_local_ptr.hpp> 18   #include <boost/corosio/detail/thread_local_ptr.hpp>
19   #include <boost/capy/error.hpp> 19   #include <boost/capy/error.hpp>
20   #include <boost/capy/ex/execution_context.hpp> 20   #include <boost/capy/ex/execution_context.hpp>
21   #include <boost/capy/ex/executor_ref.hpp> 21   #include <boost/capy/ex/executor_ref.hpp>
22   #include <system_error> 22   #include <system_error>
23   23  
24   #include <atomic> 24   #include <atomic>
25   #include <chrono> 25   #include <chrono>
26   #include <coroutine> 26   #include <coroutine>
27   #include <cstddef> 27   #include <cstddef>
28   #include <limits> 28   #include <limits>
29 - #include <optional>  
30   #include <mutex> 29   #include <mutex>
31   #include <stop_token> 30   #include <stop_token>
32   #include <utility> 31   #include <utility>
33   #include <vector> 32   #include <vector>
34   33  
35   namespace boost::corosio::detail { 34   namespace boost::corosio::detail {
36   35  
37   struct scheduler; 36   struct scheduler;
38   37  
39   /* 38   /*
40   Timer Service 39   Timer Service
41   ============= 40   =============
42   41  
43   Data Structures 42   Data Structures
44   --------------- 43   ---------------
45 - waiter_node holds per-waiter state: coroutine handle, executor, 44 + waiter_node (defined in timer.hpp) holds per-waiter state:
46 - error output, stop_token, embedded completion_op. Each concurrent 45 + coroutine handle, executor, error output, embedded
47 - co_await t.wait() allocates one waiter_node. 46 + completion_op. Each concurrent co_await t.wait() embeds one
  47 + waiter_node in the awaitable on the suspended coroutine's
  48 + frame — waits perform no allocation.
48   49  
49   timer::implementation holds per-timer state: expiry, heap 50   timer::implementation holds per-timer state: expiry, heap
50   index, and an intrusive_list of waiter_nodes. Multiple 51   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 and a free list
54 - of recycled impls, and a free list of recycled waiter_nodes. The 55 + of recycled impls. The heap is ordered by expiry time; the
55 - heap is ordered by expiry time; the scheduler queries 56 + scheduler queries nearest_expiry() to set the epoll/timerfd
56 - nearest_expiry() to set the epoll/timerfd timeout. 57 + 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. Frame-resident waiter_node with embedded completion_op —
  65 + eliminates heap allocation per wait/fire/cancel.
64   4. Cached nearest expiry — atomic avoids mutex in nearest_expiry(). 66   4. Cached nearest expiry — atomic avoids mutex in nearest_expiry().
65 - 6. Thread-local waiter cache — single-slot per-thread cache.  
66   5. might_have_pending_waits_ flag — skips lock when no wait issued. 67   5. might_have_pending_waits_ flag — skips lock when no wait issued.
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.
  73 + A waiter_node's storage is the suspended coroutine's frame:
  74 + every completion path must finish touching the node before
  75 + posting the continuation or destroying the handle.
72   */ 76   */
73 - struct BOOST_COROSIO_SYMBOL_VISIBLE waiter_node;  
74 -  
75   77  
76   inline void timer_service_invalidate_cache() noexcept; 78   inline void timer_service_invalidate_cache() noexcept;
77   79  
78   // timer_service class body — member function definitions are 80   // timer_service class body — member function definitions are
79   // out-of-class (after implementation and waiter_node are complete) 81   // out-of-class (after implementation and waiter_node are complete)
80   class BOOST_COROSIO_DECL timer_service final 82   class BOOST_COROSIO_DECL timer_service final
81   : public capy::execution_context::service 83   : public capy::execution_context::service
82   , public io_object::io_service 84   , public io_object::io_service
83   { 85   {
84   public: 86   public:
85   using clock_type = std::chrono::steady_clock; 87   using clock_type = std::chrono::steady_clock;
86   using time_point = clock_type::time_point; 88   using time_point = clock_type::time_point;
87   89  
88   /// Type-erased callback for earliest-expiry-changed notifications. 90   /// Type-erased callback for earliest-expiry-changed notifications.
89   class callback 91   class callback
90   { 92   {
91   void* ctx_ = nullptr; 93   void* ctx_ = nullptr;
92   void (*fn_)(void*) = nullptr; 94   void (*fn_)(void*) = nullptr;
93   95  
94   public: 96   public:
95   /// Construct an empty callback. 97   /// Construct an empty callback.
HITCBC 96   1225 callback() = default; 98   1219 callback() = default;
97   99  
98   /// Construct a callback with the given context and function. 100   /// Construct a callback with the given context and function.
HITCBC 99   1225 callback(void* ctx, void (*fn)(void*)) noexcept : ctx_(ctx), fn_(fn) {} 101   1219 callback(void* ctx, void (*fn)(void*)) noexcept : ctx_(ctx), fn_(fn) {}
100   102  
101   /// Return true if the callback is non-empty. 103   /// Return true if the callback is non-empty.
102   explicit operator bool() const noexcept 104   explicit operator bool() const noexcept
103   { 105   {
104   return fn_ != nullptr; 106   return fn_ != nullptr;
105   } 107   }
106   108  
107   /// Invoke the callback. 109   /// Invoke the callback.
HITCBC 108   8305 void operator()() const 110   8422 void operator()() const
109   { 111   {
HITCBC 110   8305 if (fn_) 112   8422 if (fn_)
HITCBC 111   8305 fn_(ctx_); 113   8422 fn_(ctx_);
HITCBC 112   8305 } 114   8422 }
113   }; 115   };
114   116  
115   private: 117   private:
116   struct heap_entry 118   struct heap_entry
117   { 119   {
118   time_point time_; 120   time_point time_;
119   timer::implementation* timer_; 121   timer::implementation* timer_;
120   }; 122   };
121   123  
122   scheduler* sched_ = nullptr; 124   scheduler* sched_ = nullptr;
123   BOOST_COROSIO_MSVC_WARNING_PUSH 125   BOOST_COROSIO_MSVC_WARNING_PUSH
124   BOOST_COROSIO_MSVC_WARNING_DISABLE(4251) // std:: members, dll-interface 126   BOOST_COROSIO_MSVC_WARNING_DISABLE(4251) // std:: members, dll-interface
125   mutable std::mutex mutex_; 127   mutable std::mutex mutex_;
126   std::vector<heap_entry> heap_; 128   std::vector<heap_entry> heap_;
127 - waiter_node* waiter_free_list_ = nullptr;  
128   timer::implementation* free_list_ = nullptr; 129   timer::implementation* free_list_ = nullptr;
129   callback on_earliest_changed_; 130   callback on_earliest_changed_;
130   bool shutting_down_ = false; 131   bool shutting_down_ = false;
131   // Avoids mutex in nearest_expiry() and empty() 132   // Avoids mutex in nearest_expiry() and empty()
132   mutable std::atomic<std::int64_t> cached_nearest_ns_{ 133   mutable std::atomic<std::int64_t> cached_nearest_ns_{
133   (std::numeric_limits<std::int64_t>::max)()}; 134   (std::numeric_limits<std::int64_t>::max)()};
134   BOOST_COROSIO_MSVC_WARNING_POP 135   BOOST_COROSIO_MSVC_WARNING_POP
135   136  
136   public: 137   public:
137   /// Construct the timer service bound to a scheduler. 138   /// Construct the timer service bound to a scheduler.
HITCBC 138   1225 inline timer_service(capy::execution_context&, scheduler& sched) 139   1219 inline timer_service(capy::execution_context&, scheduler& sched)
HITCBC 139   1225 : sched_(&sched) 140   1219 : sched_(&sched)
140   { 141   {
HITCBC 141   1225 } 142   1219 }
142   143  
143   /// Return the associated scheduler. 144   /// Return the associated scheduler.
HITCBC 144   17166 inline scheduler& get_scheduler() noexcept 145   16984 inline scheduler& get_scheduler() noexcept
145   { 146   {
HITCBC 146   17166 return *sched_; 147   16984 return *sched_;
147   } 148   }
148   149  
149   /// Destroy the timer service. 150   /// Destroy the timer service.
HITCBC 150   2450 ~timer_service() override = default; 151   2438 ~timer_service() override = default;
151   152  
152   timer_service(timer_service const&) = delete; 153   timer_service(timer_service const&) = delete;
153   timer_service& operator=(timer_service const&) = delete; 154   timer_service& operator=(timer_service const&) = delete;
154   155  
155   /// Register a callback invoked when the earliest expiry changes. 156   /// Register a callback invoked when the earliest expiry changes.
HITCBC 156   1225 inline void set_on_earliest_changed(callback cb) 157   1219 inline void set_on_earliest_changed(callback cb)
157   { 158   {
HITCBC 158   1225 on_earliest_changed_ = cb; 159   1219 on_earliest_changed_ = cb;
HITCBC 159   1225 } 160   1219 }
160   161  
161   /// Return true if no timers are in the heap. 162   /// Return true if no timers are in the heap.
162   inline bool empty() const noexcept 163   inline bool empty() const noexcept
163   { 164   {
164   return cached_nearest_ns_.load(std::memory_order_acquire) == 165   return cached_nearest_ns_.load(std::memory_order_acquire) ==
165   (std::numeric_limits<std::int64_t>::max)(); 166   (std::numeric_limits<std::int64_t>::max)();
166   } 167   }
167   168  
168   /// Return the nearest timer expiry without acquiring the mutex. 169   /// Return the nearest timer expiry without acquiring the mutex.
HITCBC 169   229461 inline time_point nearest_expiry() const noexcept 170   220457 inline time_point nearest_expiry() const noexcept
170   { 171   {
HITCBC 171   229461 auto ns = cached_nearest_ns_.load(std::memory_order_acquire); 172   220457 auto ns = cached_nearest_ns_.load(std::memory_order_acquire);
HITCBC 172   229461 return time_point(time_point::duration(ns)); 173   220457 return time_point(time_point::duration(ns));
173   } 174   }
174   175  
175   /// Cancel all pending timers and free cached resources. 176   /// Cancel all pending timers and free cached resources.
176   inline void shutdown() override; 177   inline void shutdown() override;
177   178  
178   /// Construct a new timer implementation. 179   /// Construct a new timer implementation.
179   inline io_object::implementation* construct() override; 180   inline io_object::implementation* construct() override;
180   181  
181   /// Destroy a timer implementation, cancelling pending waiters. 182   /// Destroy a timer implementation, cancelling pending waiters.
182   inline void destroy(io_object::implementation* p) override; 183   inline void destroy(io_object::implementation* p) override;
183   184  
184   /// Cancel and recycle a timer implementation. 185   /// Cancel and recycle a timer implementation.
185   inline void destroy_impl(timer::implementation& impl); 186   inline void destroy_impl(timer::implementation& impl);
186 - /// Create or recycle a waiter node.  
187 - inline waiter_node* create_waiter();  
188 -  
189 - /// Return a waiter node to the cache or free list.  
190 - inline void destroy_waiter(waiter_node* w);  
191 -  
192   187  
193   /// Update the timer expiry, cancelling existing waiters. 188   /// Update the timer expiry, cancelling existing waiters.
194   inline std::size_t update_timer( 189   inline std::size_t update_timer(
195   timer::implementation& impl, time_point new_time); 190   timer::implementation& impl, time_point new_time);
196   191  
197   /// Insert a waiter into the timer's waiter list and the heap. 192   /// Insert a waiter into the timer's waiter list and the heap.
198   inline void insert_waiter(timer::implementation& impl, waiter_node* w); 193   inline void insert_waiter(timer::implementation& impl, waiter_node* w);
199   194  
200   /// Cancel all waiters on a timer. 195   /// Cancel all waiters on a timer.
201   inline std::size_t cancel_timer(timer::implementation& impl); 196   inline std::size_t cancel_timer(timer::implementation& impl);
202   197  
203 - /// Cancel a single waiter ( stop_token callback path ). 198 + /// Cancel one specific waiter ( stop_token callback path ).
204   inline void cancel_waiter(waiter_node* w); 199   inline void cancel_waiter(waiter_node* w);
205   200  
206 - /// Cancel one waiter on a timer. 201 + /// Cancel the oldest pending waiter on a timer ( FIFO ).
207   inline std::size_t cancel_one_waiter(timer::implementation& impl); 202   inline std::size_t cancel_one_waiter(timer::implementation& impl);
208   203  
209   /// Complete all waiters whose timers have expired. 204   /// Complete all waiters whose timers have expired.
210   inline std::size_t process_expired(); 205   inline std::size_t process_expired();
211   206  
212   private: 207   private:
HITCBC 213   258739 inline void refresh_cached_nearest() noexcept 208   250840 inline void refresh_cached_nearest() noexcept
214   { 209   {
HITCBC 215   258739 auto ns = heap_.empty() ? (std::numeric_limits<std::int64_t>::max)() 210   250840 auto ns = heap_.empty() ? (std::numeric_limits<std::int64_t>::max)()
HITCBC 216   255742 : heap_[0].time_.time_since_epoch().count(); 211   247655 : heap_[0].time_.time_since_epoch().count();
HITCBC 217   258739 cached_nearest_ns_.store(ns, std::memory_order_release); 212   250840 cached_nearest_ns_.store(ns, std::memory_order_release);
HITCBC 218   258739 } 213   250840 }
219   214  
220   inline void remove_timer_impl(timer::implementation& impl); 215   inline void remove_timer_impl(timer::implementation& impl);
221   inline void up_heap(std::size_t index); 216   inline void up_heap(std::size_t index);
222   inline void down_heap(std::size_t index); 217   inline void down_heap(std::size_t index);
223   inline void swap_heap(std::size_t i1, std::size_t i2); 218   inline void swap_heap(std::size_t i1, std::size_t i2);
224   }; 219   };
225   220  
226 - struct BOOST_COROSIO_SYMBOL_VISIBLE waiter_node 221 + // Thread-local cache avoids hot-path mutex acquisitions:
227 - : intrusive_list<waiter_node>::node 222 + // single-slot impl cache, validated by comparing svc_. Cleared by
228 - { 223 + // timer_service_invalidate_cache() during shutdown.
229 - // Embedded completion op — avoids heap allocation per fire/cancel  
230 - struct completion_op final : scheduler_op  
231 - {  
232 - waiter_node* waiter_ = nullptr;  
233 -  
234 - static void do_complete(  
235 - void* owner, scheduler_op* base, std::uint32_t, std::uint32_t);  
236 -  
DCB 237 - 1631 completion_op() noexcept : scheduler_op(&do_complete) {}  
238 -  
239 - void operator()() override;  
240 - void destroy() override;  
241 - };  
242 -  
243 - // Per-waiter stop_token cancellation  
244 - struct canceller  
245 - {  
246 - waiter_node* waiter_;  
247 - void operator()() const;  
248 - };  
249 -  
250 - // nullptr once removed from timer's waiter list (concurrency marker)  
251 - timer::implementation* impl_ = nullptr;  
252 - timer_service* svc_ = nullptr;  
253 - std::coroutine_handle<> h_;  
254 - capy::continuation* cont_ = nullptr;  
255 - capy::executor_ref d_;  
256 - std::error_code* ec_out_ = nullptr;  
257 - std::stop_token token_;  
258 - std::optional<std::stop_callback<canceller>> stop_cb_;  
259 - completion_op op_;  
260 - std::error_code ec_value_;  
261 - waiter_node* next_free_ = nullptr;  
262 -  
DCB 263 - 1631 waiter_node() noexcept  
DCB 264 - 1631 {  
DCB 265 - 1631 op_.waiter_ = this;  
DCB 266 - 1631 }  
267 - };  
268 -  
269 - // Thread-local caches avoid hot-path mutex acquisitions:  
270 - // 1. Impl cache — single-slot, validated by comparing svc_  
271 - // 2. Waiter cache — single-slot, no service affinity  
272 - // All caches are cleared by timer_service_invalidate_cache() during shutdown.  
273   224  
274 - inline thread_local_ptr<waiter_node> tl_cached_waiter;  
275   inline thread_local_ptr<timer::implementation> tl_cached_impl; 225   inline thread_local_ptr<timer::implementation> tl_cached_impl;
276   226  
277 - // The POD TLS slots above never run destructors, so a short-lived 227 + // The POD TLS slot above never runs destructors, so a short-lived
278 - // run() thread would leak its cached impl and waiter. Each push 228 + // run() thread would leak its cached impl. Each push arms this
279 - // arms this owner, whose destructor frees the slots at thread exit. 229 + // owner, whose destructor frees the slot at thread exit. A cached
280 - // Cached entries are quiescent heap objects (stop callbacks reset, 230 + // entry is a quiescent heap object (nothing in the heap or free
281 - // nothing in the heap or free lists) and deletion touches no 231 + // list) and deletion touches no service state, so it is safe after
282 - // service state, so it is safe after the owning service is gone 232 + // the owning service is gone (the stale-entry path in
283 - // (the stale-entry path in try_pop_tl_cache deletes the same way). 233 + // try_pop_tl_cache deletes the same way).
284   struct tl_cache_owner 234   struct tl_cache_owner
285   { 235   {
HITCBC 286   40 ~tl_cache_owner() 236   35 ~tl_cache_owner()
287   { 237   {
HITCBC 288   40 delete tl_cached_impl.get(); 238   35 delete tl_cached_impl.get();
DCB 289 - 40  
290 - delete tl_cached_waiter.get();  
DCB 291 - 40 tl_cached_waiter.set(nullptr);  
HITCBC 292   40 tl_cached_impl.set(nullptr); 239   35 tl_cached_impl.set(nullptr);
HITCBC 293   40 } 240   35 }
294   }; 241   };
295   242  
296   inline void 243   inline void
HITCBC 297   16532 arm_tl_cache_cleanup() noexcept 244   9290 arm_tl_cache_cleanup() noexcept
298   { 245   {
HITCBC 299   16532 thread_local tl_cache_owner owner; 246   9290 thread_local tl_cache_owner owner;
300   (void)owner; 247   (void)owner;
HITCBC 301   16532 } 248   9290 }
302   249  
303   inline timer::implementation* 250   inline timer::implementation*
HITCBC 304   9450 try_pop_tl_cache(timer_service* svc) noexcept 251   9360 try_pop_tl_cache(timer_service* svc) noexcept
305   { 252   {
HITCBC 306   9450 auto* impl = tl_cached_impl.get(); 253   9360 auto* impl = tl_cached_impl.get();
HITCBC 307   9450 if (impl) 254   9360 if (impl)
308   { 255   {
HITCBC 309   9128 tl_cached_impl.set(nullptr); 256   9070 tl_cached_impl.set(nullptr);
HITCBC 310   9128 if (impl->svc_ == svc) 257   9070 if (impl->svc_ == svc)
HITCBC 311   9128 return impl; 258   9070 return impl;
312   // Stale impl from a destroyed service 259   // Stale impl from a destroyed service
MISUBC 313   ✗ delete impl; 260   ✗ delete impl;
314   } 261   }
HITCBC 315   322 return nullptr; 262   290 return nullptr;
316   } 263   }
317   264  
318   inline bool 265   inline bool
HITCBC 319   9424 try_push_tl_cache(timer::implementation* impl) noexcept 266   9334 try_push_tl_cache(timer::implementation* impl) noexcept
320   { 267   {
HITCBC 321   9424 if (!tl_cached_impl.get()) 268   9334 if (!tl_cached_impl.get())
322   { 269   {
HITCBC 323   9351 arm_tl_cache_cleanup(); 270   9290 arm_tl_cache_cleanup();
HITCBC 324   9351 tl_cached_impl.set(impl); 271   9290 tl_cached_impl.set(impl);
HITCBC 325   9351 return true; 272   9290 return true;
326   } 273   }
HITCBC 327   73 return false; 274   44 return false;
328   } 275   }
329 - inline waiter_node*  
330 - try_pop_waiter_tl_cache() noexcept  
DCB 331 - 8596 {  
332 - auto* w = tl_cached_waiter.get();  
DCB 333 - 8596 if (w)  
DCB 334 - 8596 {  
335 - tl_cached_waiter.set(nullptr);  
DCB 336 - 6961 return w;  
DCB 337 - 6961 }  
338 - return nullptr;  
DCB 339 - 1635 }  
340 -  
341 - inline bool  
342 - try_push_waiter_tl_cache(waiter_node* w) noexcept  
DCB 343 - 8570 {  
344 - if (!tl_cached_waiter.get())  
DCB 345 - 8570 {  
346 - arm_tl_cache_cleanup();  
DCB 347 - 7181 tl_cached_waiter.set(w);  
DCB 348 - 7181 return true;  
DCB 349 - 7181 }  
350 - return false;  
DCB 351 - 1389 }  
352 -  
353   276  
354   inline void 277   inline void
HITCBC 355   1225 timer_service_invalidate_cache() noexcept 278   1219 timer_service_invalidate_cache() noexcept
356   { 279   {
HITCBC 357   1225 delete tl_cached_impl.get(); 280   1219 delete tl_cached_impl.get();
DCB 358 - 1225  
359 - delete tl_cached_waiter.get();  
DCB 360 - 1225 tl_cached_waiter.set(nullptr);  
HITCBC 361   1225 tl_cached_impl.set(nullptr); 281   1219 tl_cached_impl.set(nullptr);
HITCBC 362   1225 } 282   1219 }
363   283  
364   // timer_service out-of-class member function definitions 284   // timer_service out-of-class member function definitions
365   285  
366   inline void 286   inline void
HITCBC 367   1225 timer_service::shutdown() 287   1219 timer_service::shutdown()
368   { 288   {
HITCBC 369   1225 timer_service_invalidate_cache(); 289   1219 timer_service_invalidate_cache();
HITCBC 370   1225 shutting_down_ = true; 290   1219 shutting_down_ = true;
371   291  
372   // Snapshot impls and detach them from the heap so that 292   // Snapshot impls and detach them from the heap so that
373   // coroutine-owned timer destructors (triggered by h.destroy() 293   // coroutine-owned timer destructors (triggered by h.destroy()
374   // below) cannot re-enter remove_timer_impl() and mutate the 294   // below) cannot re-enter remove_timer_impl() and mutate the
375   // vector during iteration. 295   // vector during iteration.
HITCBC 376   1225 std::vector<timer::implementation*> impls; 296   1219 std::vector<timer::implementation*> impls;
HITCBC 377   1225 impls.reserve(heap_.size()); 297   1219 impls.reserve(heap_.size());
HITCBC 378   1251 for (auto& entry : heap_) 298   1245 for (auto& entry : heap_)
379   { 299   {
HITCBC 380   26 entry.timer_->heap_index_.store( 300   26 entry.timer_->heap_index_.store(
381   (std::numeric_limits<std::size_t>::max)(), 301   (std::numeric_limits<std::size_t>::max)(),
382   std::memory_order_relaxed); 302   std::memory_order_relaxed);
HITCBC 383   26 impls.push_back(entry.timer_); 303   26 impls.push_back(entry.timer_);
384   } 304   }
HITCBC 385   1225 heap_.clear(); 305   1219 heap_.clear();
HITCBC 386   1225 cached_nearest_ns_.store( 306   1219 cached_nearest_ns_.store(
387   (std::numeric_limits<std::int64_t>::max)(), std::memory_order_release); 307   (std::numeric_limits<std::int64_t>::max)(), std::memory_order_release);
388   308  
389   // Cancel waiting timers. Each waiter called work_started() 309   // Cancel waiting timers. Each waiter called work_started()
390   // in implementation::wait(). On IOCP the scheduler shutdown 310   // in implementation::wait(). On IOCP the scheduler shutdown
391   // loop exits when outstanding_work_ reaches zero, so we must 311   // loop exits when outstanding_work_ reaches zero, so we must
392   // call work_finished() here to balance it. On other backends 312   // call work_finished() here to balance it. On other backends
393   // this is harmless. 313   // this is harmless.
HITCBC 394   1251 for (auto* impl : impls) 314   1245 for (auto* impl : impls)
395   { 315   {
HITCBC 396   52 while (auto* w = impl->waiters_.pop_front()) 316   52 while (auto* w = impl->waiters_.pop_front())
397   { 317   {
HITCBC 398 - 26 w->stop_cb_.reset(); 318 + 26 w->reset_stop_cb();
HITCBC 399   26 auto h = std::exchange(w->h_, {}); 319   26 auto h = std::exchange(w->h_, {});
HITCBC 400   26 sched_->work_finished(); 320   26 sched_->work_finished();
  321 + // Destroying the frame also ends the node's storage
HITCBC 401   26 if (h) 322   26 if (h)
DCB 402 - 26 delete w;  
HITCBC 403   26 h.destroy(); 323   26 h.destroy();
HITCBC 404   26 } 324   26 }
HITCBC 405   26 delete impl; 325   26 delete impl;
406   } 326   }
407   327  
408   // Delete free-listed impls 328   // Delete free-listed impls
HITCBC 409   1298 while (free_list_) 329   1263 while (free_list_)
410   { 330   {
HITCBC 411   73 auto* next = free_list_->next_free_; 331   44 auto* next = free_list_->next_free_;
HITCBC 412   73 delete free_list_; 332   44 delete free_list_;
HITCBC 413   73 free_list_ = next; 333   44 free_list_ = next;
414 -  
415 - // Delete free-listed waiters  
416 - while (waiter_free_list_)  
DCB 417 - 2610 {  
418 - auto* next = waiter_free_list_->next_free_;  
DCB 419 - 1385 delete waiter_free_list_;  
DCB 420 - 1385 waiter_free_list_ = next;  
DCB 421 - 1385 }  
422   } 334   }
HITCBC 423   1225 } 335   1219 }
424   336  
425   inline io_object::implementation* 337   inline io_object::implementation*
HITCBC 426   9450 timer_service::construct() 338   9360 timer_service::construct()
427   { 339   {
HITCBC 428   9450 timer::implementation* impl = try_pop_tl_cache(this); 340   9360 timer::implementation* impl = try_pop_tl_cache(this);
HITCBC 429   9450 if (impl) 341   9360 if (impl)
430   { 342   {
HITCBC 431 - 9128 impl->svc_ = this; 343 + 9070 impl->svc_ = this;
  344 + // Reset expiry_ too: a recycled impl must behave like a fresh
  345 + // one, whose default expiry reads as already elapsed
HITGNC   346 + 9070 impl->expiry_ = {};
HITCBC 432   9128 impl->heap_index_.store( 347   9070 impl->heap_index_.store(
433   (std::numeric_limits<std::size_t>::max)(), 348   (std::numeric_limits<std::size_t>::max)(),
434   std::memory_order_relaxed); 349   std::memory_order_relaxed);
HITCBC 435   9128 impl->might_have_pending_waits_.store(false, std::memory_order_relaxed); 350   9070 impl->might_have_pending_waits_.store(false, std::memory_order_relaxed);
HITCBC 436   9128 return impl; 351   9070 return impl;
437   } 352   }
438   353  
HITCBC 439   322 std::lock_guard lock(mutex_); 354   290 std::lock_guard lock(mutex_);
HITCBC 440   322 if (free_list_) 355   290 if (free_list_)
441   { 356   {
MISUBC 442   ✗ impl = free_list_; 357   ✗ impl = free_list_;
MISUBC 443   ✗ free_list_ = impl->next_free_; 358   ✗ free_list_ = impl->next_free_;
MISUBC 444   ✗ impl->next_free_ = nullptr; 359   ✗ impl->next_free_ = nullptr;
MISUBC 445   ✗ impl->svc_ = this; 360   ✗ impl->svc_ = this;
MISUNC   361 + ✗ impl->expiry_ = {};
MISUBC 446   ✗ impl->heap_index_.store( 362   ✗ impl->heap_index_.store(
447   (std::numeric_limits<std::size_t>::max)(), 363   (std::numeric_limits<std::size_t>::max)(),
448   std::memory_order_relaxed); 364   std::memory_order_relaxed);
MISUBC 449   ✗ impl->might_have_pending_waits_.store(false, std::memory_order_relaxed); 365   ✗ impl->might_have_pending_waits_.store(false, std::memory_order_relaxed);
450   } 366   }
451   else 367   else
452   { 368   {
HITCBC 453   322 impl = new timer::implementation(*this); 369   290 impl = new timer::implementation(*this);
454   } 370   }
HITCBC 455   322 return impl; 371   290 return impl;
HITCBC 456   322 } 372   290 }
457   373  
458   inline void 374   inline void
HITCBC 459   9450 timer_service::destroy(io_object::implementation* p) 375   9360 timer_service::destroy(io_object::implementation* p)
460   { 376   {
461   // During shutdown the drain loop owns every impl and deletes 377   // During shutdown the drain loop owns every impl and deletes
462   // them directly. A frame destroyed by that loop can unwind a 378   // them directly. A frame destroyed by that loop can unwind a
463   // handle whose impl was freed in an earlier iteration (a 379   // handle whose impl was freed in an earlier iteration (a
464   // timeout's parent frame owns the timeout timer while 380   // timeout's parent frame owns the timeout timer while
465   // suspended on the inner delay's timer), so bail out before 381   // suspended on the inner delay's timer), so bail out before
466   // even downcasting the pointer. 382   // even downcasting the pointer.
HITCBC 467   9450 if (shutting_down_) 383   9360 if (shutting_down_)
HITCBC 468   26 return; 384   26 return;
HITCBC 469   9424 destroy_impl(static_cast<timer::implementation&>(*p)); 385   9334 destroy_impl(static_cast<timer::implementation&>(*p));
470   } 386   }
471   387  
472   inline void 388   inline void
HITCBC 473   9424 timer_service::destroy_impl(timer::implementation& impl) 389   9334 timer_service::destroy_impl(timer::implementation& impl)
474   { 390   {
475   // During shutdown the impl is owned by the shutdown loop. 391   // During shutdown the impl is owned by the shutdown loop.
476   // Re-entering here (from a coroutine-owned timer destructor 392   // Re-entering here (from a coroutine-owned timer destructor
477   // triggered by h.destroy()) must not modify the heap or 393   // triggered by h.destroy()) must not modify the heap or
478   // recycle the impl — shutdown deletes it directly. 394   // recycle the impl — shutdown deletes it directly.
HITCBC 479   9424 if (shutting_down_) 395   9334 if (shutting_down_)
HITCBC 480   9351 return; 396   9290 return;
481   397  
HITCBC 482   9424 cancel_timer(impl); 398   9334 cancel_timer(impl);
483   399  
HITCBC 484   18848 if (impl.heap_index_.load(std::memory_order_relaxed) != 400   18668 if (impl.heap_index_.load(std::memory_order_relaxed) !=
HITCBC 485   9424 (std::numeric_limits<std::size_t>::max)()) 401   9334 (std::numeric_limits<std::size_t>::max)())
486   { 402   {
MISUBC 487   ✗ std::lock_guard lock(mutex_); 403   ✗ std::lock_guard lock(mutex_);
MISUBC 488   ✗ remove_timer_impl(impl); 404   ✗ remove_timer_impl(impl);
MISUBC 489   ✗ refresh_cached_nearest(); 405   ✗ refresh_cached_nearest();
MISUBC 490   ✗ } 406   ✗ }
491   407  
HITCBC 492   9424 if (try_push_tl_cache(&impl)) 408   9334 if (try_push_tl_cache(&impl))
HITCBC 493   9351 return; 409   9290 return;
494   410  
HITCBC 495   73 std::lock_guard lock(mutex_); 411   44 std::lock_guard lock(mutex_);
HITCBC 496   73 impl.next_free_ = free_list_; 412   44 impl.next_free_ = free_list_;
HITCBC 497   73 free_list_ = &impl; 413   44 free_list_ = &impl;
HITCBC 498   73 } 414   44 }
499 - inline waiter_node*  
500 - timer_service::create_waiter()  
DCB 501 - 8596 {  
502 - if (auto* w = try_pop_waiter_tl_cache())  
DCB 503 - 8596 return w;  
DCB 504 - 6961  
505 - std::lock_guard lock(mutex_);  
DCB 506 - 1635 if (waiter_free_list_)  
DCB 507 - 1635 {  
508 - auto* w = waiter_free_list_;  
DCB 509 - 4 waiter_free_list_ = w->next_free_;  
DCB 510 - 4 w->next_free_ = nullptr;  
DCB 511 - 4 return w;  
DCB 512 - 4 }  
513 -  
514 - return new waiter_node();  
DCB 515 - 1631 }  
DCB 516 - 1635  
517 - inline void  
518 - timer_service::destroy_waiter(waiter_node* w)  
DCB 519 - 8570 {  
520 - if (try_push_waiter_tl_cache(w))  
DCB 521 - 8570 return;  
DCB 522 - 7181  
523 - std::lock_guard lock(mutex_);  
DCB 524 - 1389 w->next_free_ = waiter_free_list_;  
DCB 525 - 1389 waiter_free_list_ = w;  
DCB 526 - 1389 }  
DCB 527 - 1389  
528   415  
529   inline std::size_t 416   inline std::size_t
MISUBC 530   ✗ timer_service::update_timer(timer::implementation& impl, time_point new_time) 417   ✗ timer_service::update_timer(timer::implementation& impl, time_point new_time)
531   { 418   {
532   // Gate on the flag, not waiters_: reading the non-atomic list 419   // Gate on the flag, not waiters_: reading the non-atomic list
533   // here would race a concurrent drain. A false flag is safe to 420   // here would race a concurrent drain. A false flag is safe to
534 - // trust pre-lock because every draining critical section stores 421 + // trust pre-lock: wait() stores it true before publishing, and
535 - // it false as its final touch of the impl. 422 + // it is cleared only under the mutex when the waiter list is
  423 + // empty, so false implies no published waiters.
536   bool in_heap = 424   bool in_heap =
MISUBC 537   ✗ (impl.heap_index_.load(std::memory_order_relaxed) != 425   ✗ (impl.heap_index_.load(std::memory_order_relaxed) !=
MISUBC 538   ✗ (std::numeric_limits<std::size_t>::max)()); 426   ✗ (std::numeric_limits<std::size_t>::max)());
MISUBC 539   ✗ if (!in_heap && 427   ✗ if (!in_heap &&
MISUBC 540   ✗ !impl.might_have_pending_waits_.load(std::memory_order_relaxed)) 428   ✗ !impl.might_have_pending_waits_.load(std::memory_order_relaxed))
MISUBC 541   ✗ return 0; 429   ✗ return 0;
542   430  
MISUBC 543   ✗ bool notify = false; 431   ✗ bool notify = false;
MISUBC 544   ✗ intrusive_list<waiter_node> canceled; 432   ✗ intrusive_list<waiter_node> canceled;
545   433  
546   { 434   {
MISUBC 547   ✗ std::lock_guard lock(mutex_); 435   ✗ std::lock_guard lock(mutex_);
548   436  
MISUBC 549   ✗ while (auto* w = impl.waiters_.pop_front()) 437   ✗ while (auto* w = impl.waiters_.pop_front())
550   { 438   {
MISUBC 551   ✗ w->impl_ = nullptr; 439   ✗ w->impl_ = nullptr;
MISUBC 552   ✗ canceled.push_back(w); 440   ✗ canceled.push_back(w);
MISUBC 553   ✗ } 441   ✗ }
554   442  
MISUBC 555   ✗ std::size_t idx = impl.heap_index_.load(std::memory_order_relaxed); 443   ✗ std::size_t idx = impl.heap_index_.load(std::memory_order_relaxed);
MISUBC 556   ✗ if (idx < heap_.size()) 444   ✗ if (idx < heap_.size())
557   { 445   {
MISUBC 558   ✗ time_point old_time = heap_[idx].time_; 446   ✗ time_point old_time = heap_[idx].time_;
MISUBC 559   ✗ heap_[idx].time_ = new_time; 447   ✗ heap_[idx].time_ = new_time;
560   448  
MISUBC 561   ✗ if (new_time < old_time) 449   ✗ if (new_time < old_time)
MISUBC 562   ✗ up_heap(idx); 450   ✗ up_heap(idx);
563   else 451   else
MISUBC 564   ✗ down_heap(idx); 452   ✗ down_heap(idx);
565   453  
MISUBC 566   ✗ notify = 454   ✗ notify =
MISUBC 567   ✗ (impl.heap_index_.load(std::memory_order_relaxed) == 0); 455   ✗ (impl.heap_index_.load(std::memory_order_relaxed) == 0);
568   } 456   }
569   457  
MISUBC 570   ✗ refresh_cached_nearest(); 458   ✗ refresh_cached_nearest();
MISUBC 571   ✗ } 459   ✗ }
572   460  
MISUBC 573   ✗ std::size_t count = 0; 461   ✗ std::size_t count = 0;
MISUBC 574   ✗ while (auto* w = canceled.pop_front()) 462   ✗ while (auto* w = canceled.pop_front())
575   { 463   {
MISUBC 576 - ✗ w->ec_value_ = make_error_code(capy::error::canceled); 464 + ✗ w->ec_ = make_error_code(capy::error::canceled);
MISUBC 577   ✗ sched_->post(&w->op_); 465   ✗ sched_->post(&w->op_);
MISUBC 578   ✗ ++count; 466   ✗ ++count;
MISUBC 579   ✗ } 467   ✗ }
580   468  
MISUBC 581   ✗ if (notify) 469   ✗ if (notify)
MISUBC 582   ✗ on_earliest_changed_(); 470   ✗ on_earliest_changed_();
583   471  
MISUBC 584   ✗ return count; 472   ✗ return count;
585   } 473   }
586   474  
587   inline void 475   inline void
HITCBC 588   8596 timer_service::insert_waiter(timer::implementation& impl, waiter_node* w) 476   8505 timer_service::insert_waiter(timer::implementation& impl, waiter_node* w)
589   { 477   {
HITCBC 590   8596 bool notify = false; 478   8505 bool notify = false;
HITCBC 591   8596 bool lost_cancel = false; 479   8505 bool lost_cancel = false;
592   { 480   {
HITCBC 593   8596 std::lock_guard lock(mutex_); 481   8505 std::lock_guard lock(mutex_);
594   // Publish: from here the waiter is visible to the fire path and 482   // Publish: from here the waiter is visible to the fire path and
595   // to its own stop callback (impl_ non-null enables cancel_waiter). 483   // to its own stop callback (impl_ non-null enables cancel_waiter).
HITCBC 596   8596 w->impl_ = &impl; 484   8505 w->impl_ = &impl;
HITCBC 597   17192 if (impl.heap_index_.load(std::memory_order_relaxed) == 485   17010 if (impl.heap_index_.load(std::memory_order_relaxed) ==
HITCBC 598   8596 (std::numeric_limits<std::size_t>::max)()) 486   8505 (std::numeric_limits<std::size_t>::max)())
599   { 487   {
HITCBC 600   8596 impl.heap_index_.store(heap_.size(), std::memory_order_relaxed); 488   8505 impl.heap_index_.store(heap_.size(), std::memory_order_relaxed);
HITCBC 601   8596 heap_.push_back({impl.expiry_, &impl}); 489   8505 heap_.push_back({impl.expiry_, &impl});
HITCBC 602   8596 up_heap(heap_.size() - 1); 490   8505 up_heap(heap_.size() - 1);
HITCBC 603   8596 notify = 491   8505 notify =
HITCBC 604   8596 (impl.heap_index_.load(std::memory_order_relaxed) == 0); 492   8505 (impl.heap_index_.load(std::memory_order_relaxed) == 0);
HITCBC 605   8596 refresh_cached_nearest(); 493   8505 refresh_cached_nearest();
606   } 494   }
HITCBC 607   8596 impl.waiters_.push_back(w); 495   8505 impl.waiters_.push_back(w);
608   496  
609   // Lost-cancel re-check: a stop requested after the canceller was 497   // Lost-cancel re-check: a stop requested after the canceller was
610   // armed in wait() but before this publication found impl_ null 498   // armed in wait() but before this publication found impl_ null
611   // and returned a no-op. Observe it now and undo the insertion. 499   // and returned a no-op. Observe it now and undo the insertion.
HITCBC 612 - 8596 if (w->token_.stop_requested()) 500 + 8505 if (w->token_->stop_requested())
613   { 501   {
MISLBC 614   2 w->impl_ = nullptr; 502   ✗ w->impl_ = nullptr;
MISLBC 615   2 impl.waiters_.remove(w); 503   ✗ impl.waiters_.remove(w);
MISLBC 616   2 if (impl.waiters_.empty()) 504   ✗ if (impl.waiters_.empty())
617   { 505   {
MISLBC 618   2 remove_timer_impl(impl); 506   ✗ remove_timer_impl(impl);
MISLBC 619   2 impl.might_have_pending_waits_.store( 507   ✗ impl.might_have_pending_waits_.store(
620   false, std::memory_order_relaxed); 508   false, std::memory_order_relaxed);
621   } 509   }
MISLBC 622   2 refresh_cached_nearest(); 510   ✗ refresh_cached_nearest();
MISLBC 623   2 lost_cancel = true; 511   ✗ lost_cancel = true;
MISLBC 624   2 notify = false; // insertion undone; nearest unchanged 512   ✗ notify = false; // insertion undone; nearest unchanged
625   } 513   }
HITCBC 626   8596 } 514   8505 }
HITCBC 627   8596 if (notify) 515   8505 if (notify)
HITCBC 628   8305 on_earliest_changed_(); 516   8422 on_earliest_changed_();
HITCBC 629   8596 if (lost_cancel) 517   8505 if (lost_cancel)
630   { 518   {
MISLBC 631 - 2 w->ec_value_ = make_error_code(capy::error::canceled); 519 + ✗ w->ec_ = make_error_code(capy::error::canceled);
MISLBC 632   2 sched_->post(&w->op_); 520   ✗ sched_->post(&w->op_);
633   } 521   }
HITCBC 634   8596 } 522   8505 }
635   523  
636   inline std::size_t 524   inline std::size_t
HITCBC 637   9424 timer_service::cancel_timer(timer::implementation& impl) 525   9334 timer_service::cancel_timer(timer::implementation& impl)
638   { 526   {
HITCBC 639   9424 if (!impl.might_have_pending_waits_.load(std::memory_order_relaxed)) 527   9334 if (!impl.might_have_pending_waits_.load(std::memory_order_relaxed))
HITCBC 640   9422 return 0; 528   9332 return 0;
641   529  
642   // No unlocked already-done fast-out here: it would need the 530   // No unlocked already-done fast-out here: it would need the
643   // non-atomic waiters_ (a race with concurrent drains), and an 531   // non-atomic waiters_ (a race with concurrent drains), and an
644   // index-only check is lifetime-unsafe because npos is stored 532   // index-only check is lifetime-unsafe because npos is stored
645   // before the drain finishes touching the impl. A stale-true 533   // before the drain finishes touching the impl. A stale-true
646   // flag is rare with the stateless API; the locked path below 534   // flag is rare with the stateless API; the locked path below
647   // re-validates. 535   // re-validates.
648   536  
HITCBC 649   2 intrusive_list<waiter_node> canceled; 537   2 intrusive_list<waiter_node> canceled;
650   538  
651   { 539   {
HITCBC 652   2 std::lock_guard lock(mutex_); 540   2 std::lock_guard lock(mutex_);
HITCBC 653   2 remove_timer_impl(impl); 541   2 remove_timer_impl(impl);
HITCBC 654   4 while (auto* w = impl.waiters_.pop_front()) 542   4 while (auto* w = impl.waiters_.pop_front())
655   { 543   {
HITCBC 656   2 w->impl_ = nullptr; 544   2 w->impl_ = nullptr;
HITCBC 657   2 canceled.push_back(w); 545   2 canceled.push_back(w);
HITCBC 658   2 } 546   2 }
659   // Store false as the final touch of the impl under the lock so 547   // Store false as the final touch of the impl under the lock so
660   // update_timer's pre-lock false-flag trust holds unqualified. 548   // update_timer's pre-lock false-flag trust holds unqualified.
HITCBC 661   2 impl.might_have_pending_waits_.store(false, std::memory_order_relaxed); 549   2 impl.might_have_pending_waits_.store(false, std::memory_order_relaxed);
HITCBC 662   2 refresh_cached_nearest(); 550   2 refresh_cached_nearest();
HITCBC 663   2 } 551   2 }
664   552  
HITCBC 665   2 std::size_t count = 0; 553   2 std::size_t count = 0;
HITCBC 666   4 while (auto* w = canceled.pop_front()) 554   4 while (auto* w = canceled.pop_front())
667   { 555   {
HITCBC 668 - 2 w->ec_value_ = make_error_code(capy::error::canceled); 556 + 2 w->ec_ = make_error_code(capy::error::canceled);
HITCBC 669   2 sched_->post(&w->op_); 557   2 sched_->post(&w->op_);
HITCBC 670   2 ++count; 558   2 ++count;
HITCBC 671   2 } 559   2 }
672   560  
HITCBC 673   2 return count; 561   2 return count;
674   } 562   }
675   563  
676   inline void 564   inline void
HITCBC 677   1415 timer_service::cancel_waiter(waiter_node* w) 565   1386 timer_service::cancel_waiter(waiter_node* w)
678   { 566   {
679   { 567   {
HITCBC 680   1415 std::lock_guard lock(mutex_); 568   1386 std::lock_guard lock(mutex_);
681 - // Already removed by cancel_timer or process_expired 569 + // Already removed by another drain: cancel_timer,
  570 + // cancel_one_waiter, update_timer, process_expired, or
  571 + // insert_waiter's lost-cancel recheck
HITCBC 682   1415 if (!w->impl_) 572   1386 if (!w->impl_)
MISLBC 683   5 return; 573   ✗ return;
HITCBC 684   1410 auto* impl = w->impl_; 574   1386 auto* impl = w->impl_;
HITCBC 685   1410 w->impl_ = nullptr; 575   1386 w->impl_ = nullptr;
HITCBC 686   1410 impl->waiters_.remove(w); 576   1386 impl->waiters_.remove(w);
HITCBC 687   1410 if (impl->waiters_.empty()) 577   1386 if (impl->waiters_.empty())
688   { 578   {
HITCBC 689   1410 remove_timer_impl(*impl); 579   1386 remove_timer_impl(*impl);
HITCBC 690   1410 impl->might_have_pending_waits_.store( 580   1386 impl->might_have_pending_waits_.store(
691   false, std::memory_order_relaxed); 581   false, std::memory_order_relaxed);
692   } 582   }
HITCBC 693   1410 refresh_cached_nearest(); 583   1386 refresh_cached_nearest();
HITCBC 694   1415 } 584   1386 }
695   585  
HITCBC 696 - 1410 w->ec_value_ = make_error_code(capy::error::canceled); 586 + 1386 w->ec_ = make_error_code(capy::error::canceled);
HITCBC 697   1410 sched_->post(&w->op_); 587   1386 sched_->post(&w->op_);
698   } 588   }
699   589  
700   inline std::size_t 590   inline std::size_t
MISUBC 701   ✗ timer_service::cancel_one_waiter(timer::implementation& impl) 591   ✗ timer_service::cancel_one_waiter(timer::implementation& impl)
702   { 592   {
MISUBC 703   ✗ if (!impl.might_have_pending_waits_.load(std::memory_order_relaxed)) 593   ✗ if (!impl.might_have_pending_waits_.load(std::memory_order_relaxed))
MISUBC 704   ✗ return 0; 594   ✗ return 0;
705   595  
MISUBC 706   ✗ waiter_node* w = nullptr; 596   ✗ waiter_node* w = nullptr;
707   597  
708   { 598   {
MISUBC 709   ✗ std::lock_guard lock(mutex_); 599   ✗ std::lock_guard lock(mutex_);
MISUBC 710   ✗ w = impl.waiters_.pop_front(); 600   ✗ w = impl.waiters_.pop_front();
MISUBC 711   ✗ if (!w) 601   ✗ if (!w)
MISUBC 712   ✗ return 0; 602   ✗ return 0;
MISUBC 713   ✗ w->impl_ = nullptr; 603   ✗ w->impl_ = nullptr;
MISUBC 714   ✗ if (impl.waiters_.empty()) 604   ✗ if (impl.waiters_.empty())
715   { 605   {
MISUBC 716   ✗ remove_timer_impl(impl); 606   ✗ remove_timer_impl(impl);
MISUBC 717   ✗ impl.might_have_pending_waits_.store( 607   ✗ impl.might_have_pending_waits_.store(
718   false, std::memory_order_relaxed); 608   false, std::memory_order_relaxed);
719   } 609   }
MISUBC 720   ✗ refresh_cached_nearest(); 610   ✗ refresh_cached_nearest();
MISUBC 721   ✗ } 611   ✗ }
722   612  
MISUBC 723 - ✗ w->ec_value_ = make_error_code(capy::error::canceled); 613 + ✗ w->ec_ = make_error_code(capy::error::canceled);
MISUBC 724   ✗ sched_->post(&w->op_); 614   ✗ sched_->post(&w->op_);
MISUBC 725   ✗ return 1; 615   ✗ return 1;
726   } 616   }
727   617  
728   inline std::size_t 618   inline std::size_t
HITCBC 729   248729 timer_service::process_expired() 619   240947 timer_service::process_expired()
730   { 620   {
HITCBC 731   248729 intrusive_list<waiter_node> expired; 621   240947 intrusive_list<waiter_node> expired;
732   622  
733   { 623   {
HITCBC 734   248729 std::lock_guard lock(mutex_); 624   240947 std::lock_guard lock(mutex_);
HITCBC 735   248729 auto now = clock_type::now(); 625   240947 auto now = clock_type::now();
736   626  
HITCBC 737   255885 while (!heap_.empty() && heap_[0].time_ <= now) 627   248038 while (!heap_.empty() && heap_[0].time_ <= now)
738   { 628   {
HITCBC 739   7156 timer::implementation* t = heap_[0].timer_; 629   7091 timer::implementation* t = heap_[0].timer_;
HITCBC 740   7156 remove_timer_impl(*t); 630   7091 remove_timer_impl(*t);
HITCBC 741   14312 while (auto* w = t->waiters_.pop_front()) 631   14182 while (auto* w = t->waiters_.pop_front())
742   { 632   {
HITCBC 743 - 7156 w->impl_ = nullptr; 633 + 7091 w->impl_ = nullptr;
HITCBC 744 - 7156 w->ec_value_ = {}; 634 + 7091 w->ec_ = {};
HITCBC 745   7156 expired.push_back(w); 635   7091 expired.push_back(w);
HITCBC 746   7156 } 636   7091 }
HITCBC 747   7156 t->might_have_pending_waits_.store( 637   7091 t->might_have_pending_waits_.store(
748   false, std::memory_order_relaxed); 638   false, std::memory_order_relaxed);
749   } 639   }
750   640  
HITCBC 751   248729 refresh_cached_nearest(); 641   240947 refresh_cached_nearest();
HITCBC 752   248729 } 642   240947 }
753   643  
HITCBC 754   248729 std::size_t count = 0; 644   240947 std::size_t count = 0;
HITCBC 755   255885 while (auto* w = expired.pop_front()) 645   248038 while (auto* w = expired.pop_front())
756   { 646   {
HITCBC 757   7156 sched_->post(&w->op_); 647   7091 sched_->post(&w->op_);
HITCBC 758   7156 ++count; 648   7091 ++count;
HITCBC 759   7156 } 649   7091 }
760   650  
HITCBC 761   248729 return count; 651   240947 return count;
762   } 652   }
763   653  
764   inline void 654   inline void
HITCBC 765   8570 timer_service::remove_timer_impl(timer::implementation& impl) 655   8479 timer_service::remove_timer_impl(timer::implementation& impl)
766   { 656   {
HITCBC 767   8570 std::size_t index = impl.heap_index_.load(std::memory_order_relaxed); 657   8479 std::size_t index = impl.heap_index_.load(std::memory_order_relaxed);
HITCBC 768   8570 if (index >= heap_.size()) 658   8479 if (index >= heap_.size())
MISUBC 769   ✗ return; // Not in heap 659   ✗ return; // Not in heap
770   660  
HITCBC 771   8570 if (index == heap_.size() - 1) 661   8479 if (index == heap_.size() - 1)
772   { 662   {
773   // Last element, just pop 663   // Last element, just pop
HITCBC 774   1608 impl.heap_index_.store( 664   1622 impl.heap_index_.store(
775   (std::numeric_limits<std::size_t>::max)(), 665   (std::numeric_limits<std::size_t>::max)(),
776   std::memory_order_relaxed); 666   std::memory_order_relaxed);
HITCBC 777   1608 heap_.pop_back(); 667   1622 heap_.pop_back();
778   } 668   }
779   else 669   else
780   { 670   {
781   // Swap with last and reheapify 671   // Swap with last and reheapify
HITCBC 782   6962 swap_heap(index, heap_.size() - 1); 672   6857 swap_heap(index, heap_.size() - 1);
HITCBC 783   6962 impl.heap_index_.store( 673   6857 impl.heap_index_.store(
784   (std::numeric_limits<std::size_t>::max)(), 674   (std::numeric_limits<std::size_t>::max)(),
785   std::memory_order_relaxed); 675   std::memory_order_relaxed);
HITCBC 786   6962 heap_.pop_back(); 676   6857 heap_.pop_back();
787   677  
HITCBC 788   6962 if (index > 0 && heap_[index].time_ < heap_[(index - 1) / 2].time_) 678   6857 if (index > 0 && heap_[index].time_ < heap_[(index - 1) / 2].time_)
MISUBC 789   ✗ up_heap(index); 679   ✗ up_heap(index);
790   else 680   else
HITCBC 791   6962 down_heap(index); 681   6857 down_heap(index);
792   } 682   }
793   } 683   }
794   684  
795   inline void 685   inline void
HITCBC 796   8596 timer_service::up_heap(std::size_t index) 686   8505 timer_service::up_heap(std::size_t index)
797   { 687   {
HITCBC 798   15515 while (index > 0) 688   15338 while (index > 0)
799   { 689   {
HITCBC 800   7208 std::size_t parent = (index - 1) / 2; 690   6916 std::size_t parent = (index - 1) / 2;
HITCBC 801   7208 if (!(heap_[index].time_ < heap_[parent].time_)) 691   6916 if (!(heap_[index].time_ < heap_[parent].time_))
HITCBC 802   289 break; 692   83 break;
HITCBC 803   6919 swap_heap(index, parent); 693   6833 swap_heap(index, parent);
HITCBC 804   6919 index = parent; 694   6833 index = parent;
805   } 695   }
HITCBC 806   8596 } 696   8505 }
807   697  
808   inline void 698   inline void
HITCBC 809   6962 timer_service::down_heap(std::size_t index) 699   6857 timer_service::down_heap(std::size_t index)
810   { 700   {
HITCBC 811   6962 std::size_t child = index * 2 + 1; 701   6857 std::size_t child = index * 2 + 1;
HITCBC 812   7008 while (child < heap_.size()) 702   6859 while (child < heap_.size())
813   { 703   {
HITCBC 814   53 std::size_t min_child = (child + 1 == heap_.size() || 704   4 std::size_t min_child = (child + 1 == heap_.size() ||
MISLBC 815   46 heap_[child].time_ < heap_[child + 1].time_) 705   ✗ heap_[child].time_ < heap_[child + 1].time_)
HITCBC 816   99 ? child 706   4 ? child
HITCBC 817   53 : child + 1; 707   4 : child + 1;
818   708  
HITCBC 819   53 if (heap_[index].time_ < heap_[min_child].time_) 709   4 if (heap_[index].time_ < heap_[min_child].time_)
HITCBC 820   7 break; 710   2 break;
821   711  
HITCBC 822   46 swap_heap(index, min_child); 712   2 swap_heap(index, min_child);
HITCBC 823   46 index = min_child; 713   2 index = min_child;
HITCBC 824   46 child = index * 2 + 1; 714   2 child = index * 2 + 1;
825   } 715   }
HITCBC 826   6962 } 716   6857 }
827   717  
828   inline void 718   inline void
HITCBC 829   13927 timer_service::swap_heap(std::size_t i1, std::size_t i2) 719   13692 timer_service::swap_heap(std::size_t i1, std::size_t i2)
830   { 720   {
HITCBC 831   13927 heap_entry tmp = heap_[i1]; 721   13692 heap_entry tmp = heap_[i1];
HITCBC 832   13927 heap_[i1] = heap_[i2]; 722   13692 heap_[i1] = heap_[i2];
HITCBC 833   13927 heap_[i2] = tmp; 723   13692 heap_[i2] = tmp;
HITCBC 834   13927 heap_[i1].timer_->heap_index_.store(i1, std::memory_order_relaxed); 724   13692 heap_[i1].timer_->heap_index_.store(i1, std::memory_order_relaxed);
HITCBC 835   13927 heap_[i2].timer_->heap_index_.store(i2, std::memory_order_relaxed); 725   13692 heap_[i2].timer_->heap_index_.store(i2, std::memory_order_relaxed);
HITCBC 836   13927 } 726   13692 }
837   727  
838 - // waiter_node out-of-class member function definitions 728 + // waiter_node's completion_op and canceller members are defined in
839 - 729 + // timer.cpp alongside implementation::wait(), for the same reason
840 - inline void 730 + // wait() lives there (see below).
DCB 841 - 1415 waiter_node::canceller::operator()() const  
842 - {  
DCB 843 - 1415 waiter_->svc_->cancel_waiter(waiter_);  
DCB 844 - 1415 }  
845 -  
846 - inline void  
DUB 847 - ✗ waiter_node::completion_op::do_complete(  
848 - [[maybe_unused]] void* owner,  
849 - scheduler_op* base,  
850 - std::uint32_t,  
851 - std::uint32_t)  
852 - {  
853 - // owner is always non-null here. The destroy path (owner == nullptr)  
854 - // is unreachable because completion_op overrides destroy() directly,  
855 - // bypassing scheduler_op::destroy() which would call func_(nullptr, ...).  
DUB 856 - ✗ BOOST_COROSIO_ASSERT(owner);  
DUB 857 - ✗ static_cast<completion_op*>(base)->operator()();  
DUB 858 - ✗ }  
859 -  
860 - inline void  
DCB 861 - 8570 waiter_node::completion_op::operator()()  
862 - {  
DCB 863 - 8570 auto* w = waiter_;  
DCB 864 - 8570 w->stop_cb_.reset();  
DCB 865 - 8570 if (w->ec_out_)  
DCB 866 - 8570 *w->ec_out_ = w->ec_value_;  
867 -  
DCB 868 - 8570 auto* cont = w->cont_;  
DCB 869 - 8570 auto d = w->d_;  
DCB 870 - 8570 auto* svc = w->svc_;  
DCB 871 - 8570 auto& sched = svc->get_scheduler();  
872 -  
DCB 873 - 8570 svc->destroy_waiter(w);  
874 -  
DCB 875 - 8570 d.post(*cont);  
DCB 876 - 8570 sched.work_finished();  
DCB 877 - 8570 }  
878 -  
879 - // GCC 14 false-positive: inlining ~optional<stop_callback> through  
880 - // delete loses track that stop_cb_ was already .reset() above.  
881 - #if defined(__GNUC__) && !defined(__clang__)  
882 - #pragma GCC diagnostic push  
883 - #pragma GCC diagnostic ignored "-Wmaybe-uninitialized"  
884 - #endif  
885 - inline void  
DUB 886 - ✗ waiter_node::completion_op::destroy()  
887 - {  
888 - // Called during scheduler shutdown drain when this completion_op is  
889 - // in the scheduler's ready queue (posted by cancel_timer() or  
890 - // process_expired()). Balances the work_started() from  
891 - // implementation::wait(). The scheduler drain loop separately  
892 - // balances the work_started() from post(). On IOCP both decrements  
893 - // are required for outstanding_work_ to reach zero; on other  
894 - // backends this is harmless.  
895 - //  
896 - // This override also prevents scheduler_op::destroy() from calling  
897 - // do_complete(nullptr, ...). See also: timer_service::shutdown()  
898 - // which drains waiters still in the timer heap (the other path).  
DUB 899 - ✗ auto* w = waiter_;  
DUB 900 - ✗ w->stop_cb_.reset();  
DUB 901 - ✗ auto h = std::exchange(w->h_, {});  
DUB 902 - ✗ auto& sched = w->svc_->get_scheduler();  
DUB 903 - ✗ delete w;  
DUB 904 - ✗ sched.work_finished();  
DUB 905 - ✗ if (h)  
DUB 906 - ✗ h.destroy();  
DUB 907 - ✗ }  
908 - #if defined(__GNUC__) && !defined(__clang__)  
909 - #pragma GCC diagnostic pop  
910 - #endif  
911   731  
912   // timer::implementation::wait() is defined in timer.cpp, not here. 732   // timer::implementation::wait() is defined in timer.cpp, not here.
913   // It must be a non-inline definition in a translation unit that is 733   // It must be a non-inline definition in a translation unit that is
914   // always pulled into the link whenever detail::timer is used (every 734   // always pulled into the link whenever detail::timer is used (every
915   // consumer needs timer's constructors from that same object file). 735   // consumer needs timer's constructors from that same object file).
916   // An inline definition in this header would only be emitted in 736   // An inline definition in this header would only be emitted in
917   // translation units that happen to also include this header, which 737   // translation units that happen to also include this header, which
918   // is not guaranteed for every caller of wait_awaitable::await_suspend 738   // is not guaranteed for every caller of wait_awaitable::await_suspend
919   // in timer.hpp (e.g. code that only reaches timer.hpp through 739   // in timer.hpp (e.g. code that only reaches timer.hpp through
920   // delay.hpp, without transitively including a scheduler header). 740   // delay.hpp, without transitively including a scheduler header).
921   741  
922   // Free functions 742   // Free functions
923   743  
924   inline std::size_t 744   inline std::size_t
MISUBC 925   ✗ timer_service_update_expiry(timer::implementation& impl) 745   ✗ timer_service_update_expiry(timer::implementation& impl)
926   { 746   {
MISUBC 927   ✗ return impl.svc_->update_timer(impl, impl.expiry_); 747   ✗ return impl.svc_->update_timer(impl, impl.expiry_);
928   } 748   }
929   749  
930   inline std::size_t 750   inline std::size_t
MISUBC 931   ✗ timer_service_cancel(timer::implementation& impl) noexcept 751   ✗ timer_service_cancel(timer::implementation& impl) noexcept
932   { 752   {
MISUBC 933   ✗ return impl.svc_->cancel_timer(impl); 753   ✗ return impl.svc_->cancel_timer(impl);
934   } 754   }
935   755  
936   inline std::size_t 756   inline std::size_t
MISUBC 937   ✗ timer_service_cancel_one(timer::implementation& impl) noexcept 757   ✗ timer_service_cancel_one(timer::implementation& impl) noexcept
938   { 758   {
MISUBC 939   ✗ return impl.svc_->cancel_one_waiter(impl); 759   ✗ return impl.svc_->cancel_one_waiter(impl);
940   } 760   }
941   761  
942   inline timer_service& 762   inline timer_service&
HITCBC 943   1225 get_timer_service(capy::execution_context& ctx, scheduler& sched) 763   1219 get_timer_service(capy::execution_context& ctx, scheduler& sched)
944   { 764   {
HITCBC 945   1225 return ctx.make_service<timer_service>(sched); 765   1219 return ctx.make_service<timer_service>(sched);
946   } 766   }
947   767  
948   } // namespace boost::corosio::detail 768   } // namespace boost::corosio::detail
949   769  
950   #endif 770   #endif