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