. /include/boost/corosio /native /detail /select /select_scheduler.hpp
99.44% Lines (177/178) 100.00% Functions (12/12)
[ ] prev/next file n p prev/next uncovered line
TLA Baseline Branch
Line Hits Code Line Hits Code
1   // 1   //
2   // Copyright (c) 2026 Steve Gerbino 2   // Copyright (c) 2026 Steve Gerbino
3   // Copyright (c) 2026 Michael Vandeberg 3   // Copyright (c) 2026 Michael Vandeberg
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_NATIVE_DETAIL_SELECT_SELECT_SCHEDULER_HPP 11   #ifndef BOOST_COROSIO_NATIVE_DETAIL_SELECT_SELECT_SCHEDULER_HPP
12   #define BOOST_COROSIO_NATIVE_DETAIL_SELECT_SELECT_SCHEDULER_HPP 12   #define BOOST_COROSIO_NATIVE_DETAIL_SELECT_SELECT_SCHEDULER_HPP
13   13  
14   #include <boost/corosio/detail/platform.hpp> 14   #include <boost/corosio/detail/platform.hpp>
15   15  
16   #if BOOST_COROSIO_HAS_SELECT 16   #if BOOST_COROSIO_HAS_SELECT
17   17  
18   #include <boost/corosio/detail/config.hpp> 18   #include <boost/corosio/detail/config.hpp>
19   #include <boost/capy/ex/execution_context.hpp> 19   #include <boost/capy/ex/execution_context.hpp>
20   20  
21   #include <boost/corosio/native/detail/reactor/reactor_scheduler.hpp> 21   #include <boost/corosio/native/detail/reactor/reactor_scheduler.hpp>
22   #include <boost/corosio/native/detail/reactor/reactor_signal_pipe.hpp> 22   #include <boost/corosio/native/detail/reactor/reactor_signal_pipe.hpp>
23   23  
24   #include <boost/corosio/native/detail/select/select_traits.hpp> 24   #include <boost/corosio/native/detail/select/select_traits.hpp>
25   #include <boost/corosio/detail/timer_service.hpp> 25   #include <boost/corosio/detail/timer_service.hpp>
26   #include <boost/corosio/native/detail/make_err.hpp> 26   #include <boost/corosio/native/detail/make_err.hpp>
27   27  
28   #include <boost/corosio/detail/except.hpp> 28   #include <boost/corosio/detail/except.hpp>
29   29  
30   #include <sys/select.h> 30   #include <sys/select.h>
31   #include <unistd.h> 31   #include <unistd.h>
32   #include <errno.h> 32   #include <errno.h>
33   #include <fcntl.h> 33   #include <fcntl.h>
34   34  
35   #include <atomic> 35   #include <atomic>
36   #include <chrono> 36   #include <chrono>
37   #include <cstdint> 37   #include <cstdint>
38   #include <limits> 38   #include <limits>
39   #include <mutex> 39   #include <mutex>
40   #include <new> 40   #include <new>
41   #include <unordered_map> 41   #include <unordered_map>
42   42  
43   namespace boost::corosio::detail { 43   namespace boost::corosio::detail {
44   44  
45   struct select_op; 45   struct select_op;
46   46  
47   /** POSIX scheduler using select() for I/O multiplexing. 47   /** POSIX scheduler using select() for I/O multiplexing.
48   48  
49   This scheduler implements the scheduler interface using the POSIX select() 49   This scheduler implements the scheduler interface using the POSIX select()
50   call for I/O event notification. It inherits the shared reactor threading 50   call for I/O event notification. It inherits the shared reactor threading
51   model from reactor_scheduler: signal state machine, inline completion 51   model from reactor_scheduler: signal state machine, inline completion
52   budget, work counting, and the do_one event loop. 52   budget, work counting, and the do_one event loop.
53   53  
54   The design mirrors epoll_scheduler for behavioral consistency: 54   The design mirrors epoll_scheduler for behavioral consistency:
55   - Same single-reactor thread coordination model 55   - Same single-reactor thread coordination model
56   - Same deferred I/O pattern (reactor marks ready; workers do I/O) 56   - Same deferred I/O pattern (reactor marks ready; workers do I/O)
57   - Same timer integration pattern 57   - Same timer integration pattern
58   58  
59   Known Limitations: 59   Known Limitations:
60   - FD_SETSIZE (~1024) limits maximum concurrent connections 60   - FD_SETSIZE (~1024) limits maximum concurrent connections
61   - O(n) scanning: rebuilds fd_sets each iteration 61   - O(n) scanning: rebuilds fd_sets each iteration
62   - Level-triggered only (no edge-triggered mode) 62   - Level-triggered only (no edge-triggered mode)
63   63  
64   @par Thread Safety 64   @par Thread Safety
65   All public member functions are thread-safe. 65   All public member functions are thread-safe.
66   */ 66   */
67   class BOOST_COROSIO_DECL select_scheduler final : public reactor_scheduler 67   class BOOST_COROSIO_DECL select_scheduler final : public reactor_scheduler
68   { 68   {
69   public: 69   public:
70   /** Construct the scheduler. 70   /** Construct the scheduler.
71   71  
72   Creates a self-pipe for reactor interruption. 72   Creates a self-pipe for reactor interruption.
73   73  
74   @param ctx Reference to the owning execution_context. 74   @param ctx Reference to the owning execution_context.
75   @param concurrency_hint Hint for expected thread count (unused). 75   @param concurrency_hint Hint for expected thread count (unused).
76   */ 76   */
77   select_scheduler(capy::execution_context& ctx, int concurrency_hint = -1); 77   select_scheduler(capy::execution_context& ctx, int concurrency_hint = -1);
78   78  
79   /// Destroy the scheduler. 79   /// Destroy the scheduler.
80   ~select_scheduler() override; 80   ~select_scheduler() override;
81   81  
82   select_scheduler(select_scheduler const&) = delete; 82   select_scheduler(select_scheduler const&) = delete;
83   select_scheduler& operator=(select_scheduler const&) = delete; 83   select_scheduler& operator=(select_scheduler const&) = delete;
84   84  
85   /// Shut down the scheduler, draining pending operations. 85   /// Shut down the scheduler, draining pending operations.
86   void shutdown() override; 86   void shutdown() override;
87   87  
88   /** Return the maximum file descriptor value supported. 88   /** Return the maximum file descriptor value supported.
89   89  
90   Returns FD_SETSIZE - 1, the maximum fd value that can be 90   Returns FD_SETSIZE - 1, the maximum fd value that can be
91   monitored by select(). Operations with fd >= FD_SETSIZE 91   monitored by select(). Operations with fd >= FD_SETSIZE
92   will fail with EINVAL. 92   will fail with EINVAL.
93   93  
94   @return The maximum supported file descriptor value. 94   @return The maximum supported file descriptor value.
95   */ 95   */
96   static constexpr int max_fd() noexcept 96   static constexpr int max_fd() noexcept
97   { 97   {
98   return FD_SETSIZE - 1; 98   return FD_SETSIZE - 1;
99   } 99   }
100   100  
101   /** Register a descriptor for persistent monitoring. 101   /** Register a descriptor for persistent monitoring.
102   102  
103   The fd is added to the registered_descs_ map and will be 103   The fd is added to the registered_descs_ map and will be
104   included in subsequent select() calls. The reactor is 104   included in subsequent select() calls. The reactor is
105   interrupted so a blocked select() rebuilds its fd_sets. 105   interrupted so a blocked select() rebuilds its fd_sets.
106   106  
107   @param fd The file descriptor to register. 107   @param fd The file descriptor to register.
108   @param desc Pointer to descriptor state for this fd. 108   @param desc Pointer to descriptor state for this fd.
109   109  
110   @return The error if the fd cannot be tracked, otherwise a 110   @return The error if the fd cannot be tracked, otherwise a
111   default constructed error code. 111   default constructed error code.
112   */ 112   */
113   std::error_code 113   std::error_code
114   register_descriptor(int fd, reactor_descriptor_state* desc) const; 114   register_descriptor(int fd, reactor_descriptor_state* desc) const;
115   115  
116   /// No-op: write readiness is watched from registration on. 116   /// No-op: write readiness is watched from registration on.
117   std::error_code 117   std::error_code
HITCBC 118   2676 ensure_write_registered(int, reactor_descriptor_state*) const noexcept 118   2709 ensure_write_registered(int, reactor_descriptor_state*) const noexcept
119   { 119   {
HITCBC 120   2676 return {}; 120   2709 return {};
121   } 121   }
122   122  
123   /** Deregister a persistently registered descriptor. 123   /** Deregister a persistently registered descriptor.
124   124  
125   @param fd The file descriptor to deregister. 125   @param fd The file descriptor to deregister.
126   */ 126   */
127   void deregister_descriptor(int fd) const; 127   void deregister_descriptor(int fd) const;
128   128  
129   /** Interrupt the reactor so it rebuilds its fd_sets. 129   /** Interrupt the reactor so it rebuilds its fd_sets.
130   130  
131   Called when a write, connect, or write-wait op is registered 131   Called when a write, connect, or write-wait op is registered
132   after the reactor's snapshot was taken. Without this, 132   after the reactor's snapshot was taken. Without this,
133   select() may block not watching for writability on the fd. 133   select() may block not watching for writability on the fd.
134   */ 134   */
135   void notify_reactor() const; 135   void notify_reactor() const;
136   136  
137   /// Watch the read end of the POSIX signal self-pipe (see scheduler.hpp). 137   /// Watch the read end of the POSIX signal self-pipe (see scheduler.hpp).
HITCBC 138   273 [[nodiscard]] std::error_code register_signal_reader(int read_fd) override 138   273 [[nodiscard]] std::error_code register_signal_reader(int read_fd) override
139   { 139   {
HITCBC 140   273 return register_descriptor(read_fd, signal_pipe_reader_.arm()); 140   273 return register_descriptor(read_fd, signal_pipe_reader_.arm());
141   } 141   }
142   142  
143   private: 143   private:
144   void run_task(lock_type& lock, context_type& ctx, long timeout_us) override; 144   void run_task(lock_type& lock, context_type& ctx, long timeout_us) override;
145   void interrupt_reactor() const override; 145   void interrupt_reactor() const override;
146   146  
147   /// Rebuild the poll set a blocked select() was given. 147   /// Rebuild the poll set a blocked select() was given.
HITCBC 148   375 void notify_reparked() const override 148   467 void notify_reparked() const override
149   { 149   {
HITCBC 150   375 interrupt_reactor(); 150   467 interrupt_reactor();
HITCBC 151   375 } 151   467 }
152   long calculate_timeout(long requested_timeout_us) const; 152   long calculate_timeout(long requested_timeout_us) const;
153   153  
154   // Watches the global signal self-pipe's read end (armed lazily by 154   // Watches the global signal self-pipe's read end (armed lazily by
155   // register_signal_reader on the first signal registration). 155   // register_signal_reader on the first signal registration).
156   reactor_signal_pipe_reader signal_pipe_reader_; 156   reactor_signal_pipe_reader signal_pipe_reader_;
157   157  
158   // Self-pipe for interrupting select() 158   // Self-pipe for interrupting select()
159   int pipe_fds_[2]; // [0]=read, [1]=write 159   int pipe_fds_[2]; // [0]=read, [1]=write
160   160  
161   // Per-fd tracking for fd_set building 161   // Per-fd tracking for fd_set building
162   mutable std::unordered_map<int, reactor_descriptor_state*> 162   mutable std::unordered_map<int, reactor_descriptor_state*>
163   registered_descs_; 163   registered_descs_;
164   mutable int max_fd_ = -1; 164   mutable int max_fd_ = -1;
165   }; 165   };
166   166  
HITCBC 167 - 1954 inline select_scheduler::select_scheduler(capy::execution_context& ctx, int) 167 + 1958 inline select_scheduler::select_scheduler(capy::execution_context&, int)
HITCBC 168   1954 : pipe_fds_{-1, -1} 168   1958 : pipe_fds_{-1, -1}
HITCBC 169   1954 , max_fd_(-1) 169   1958 , max_fd_(-1)
170   { 170   {
HITCBC 171   1954 if (::pipe(pipe_fds_) < 0) 171   1958 if (::pipe(pipe_fds_) < 0)
HITCBC 172   1 detail::throw_system_error(make_err(errno), "pipe"); 172   1 detail::throw_system_error(make_err(errno), "pipe");
173   173  
HITCBC 174   5850 for (int i = 0; i < 2; ++i) 174   5862 for (int i = 0; i < 2; ++i)
175   { 175   {
HITCBC 176   3903 int flags = ::fcntl(pipe_fds_[i], F_GETFL, 0); 176   3911 int flags = ::fcntl(pipe_fds_[i], F_GETFL, 0);
HITCBC 177   3903 if (flags == -1) 177   3911 if (flags == -1)
178   { 178   {
HITCBC 179   2 int errn = errno; 179   2 int errn = errno;
HITCBC 180   2 ::close(pipe_fds_[0]); 180   2 ::close(pipe_fds_[0]);
HITCBC 181   2 ::close(pipe_fds_[1]); 181   2 ::close(pipe_fds_[1]);
HITCBC 182   2 detail::throw_system_error(make_err(errn), "fcntl F_GETFL"); 182   2 detail::throw_system_error(make_err(errn), "fcntl F_GETFL");
183   } 183   }
HITCBC 184   3901 if (::fcntl(pipe_fds_[i], F_SETFL, flags | O_NONBLOCK) == -1) 184   3909 if (::fcntl(pipe_fds_[i], F_SETFL, flags | O_NONBLOCK) == -1)
185   { 185   {
HITCBC 186   2 int errn = errno; 186   2 int errn = errno;
HITCBC 187   2 ::close(pipe_fds_[0]); 187   2 ::close(pipe_fds_[0]);
HITCBC 188   2 ::close(pipe_fds_[1]); 188   2 ::close(pipe_fds_[1]);
HITCBC 189   2 detail::throw_system_error(make_err(errn), "fcntl F_SETFL"); 189   2 detail::throw_system_error(make_err(errn), "fcntl F_SETFL");
190   } 190   }
HITCBC 191   3899 if (::fcntl(pipe_fds_[i], F_SETFD, FD_CLOEXEC) == -1) 191   3907 if (::fcntl(pipe_fds_[i], F_SETFD, FD_CLOEXEC) == -1)
192   { 192   {
HITCBC 193   2 int errn = errno; 193   2 int errn = errno;
HITCBC 194   2 ::close(pipe_fds_[0]); 194   2 ::close(pipe_fds_[0]);
HITCBC 195   2 ::close(pipe_fds_[1]); 195   2 ::close(pipe_fds_[1]);
HITCBC 196   2 detail::throw_system_error(make_err(errn), "fcntl F_SETFD"); 196   2 detail::throw_system_error(make_err(errn), "fcntl F_SETFD");
197   } 197   }
198   } 198   }
199 - timer_svc_ = &get_timer_service(ctx, *this);  
DCB 200 - 1947 timer_svc_->set_on_earliest_changed(  
DCB 201 - 1947 timer_service::callback(this, [](void* p) {  
DCB 202 - 5138 static_cast<select_scheduler*>(p)->interrupt_reactor();  
DCB 203 - 3191 }));  
DCB 204 - 3191  
205   199  
HITCBC 206   1947 completed_ops_.push(&task_op_); 200   1951 completed_ops_.push(&task_op_);
HITCBC 207   1968 } 201   1972 }
208   202  
HITCBC 209   3894 inline select_scheduler::~select_scheduler() 203   3902 inline select_scheduler::~select_scheduler()
210   { 204   {
HITCBC 211   1947 if (pipe_fds_[0] >= 0) 205   1951 if (pipe_fds_[0] >= 0)
HITCBC 212   1947 ::close(pipe_fds_[0]); 206   1951 ::close(pipe_fds_[0]);
HITCBC 213   1947 if (pipe_fds_[1] >= 0) 207   1951 if (pipe_fds_[1] >= 0)
HITCBC 214   1947 ::close(pipe_fds_[1]); 208   1951 ::close(pipe_fds_[1]);
HITCBC 215   3894 } 209   3902 }
216   210  
217   inline void 211   inline void
HITCBC 218   1947 select_scheduler::shutdown() 212   1951 select_scheduler::shutdown()
219   { 213   {
HITCBC 220   1947 shutdown_drain(); 214   1951 shutdown_drain();
221   215  
HITCBC 222   1947 if (pipe_fds_[1] >= 0) 216   1951 if (pipe_fds_[1] >= 0)
HITCBC 223   1947 interrupt_reactor(); 217   1951 interrupt_reactor();
HITCBC 224   1947 } 218   1951 }
225   219  
226   inline std::error_code 220   inline std::error_code
HITCBC 227   9793 select_scheduler::register_descriptor( 221   9846 select_scheduler::register_descriptor(
228   int fd, reactor_descriptor_state* desc) const 222   int fd, reactor_descriptor_state* desc) const
229   { 223   {
HITCBC 230   9793 if (fd < 0 || fd >= FD_SETSIZE) 224   9846 if (fd < 0 || fd >= FD_SETSIZE)
HITCBC 231   1 return make_err(EMFILE); 225   1 return make_err(EMFILE);
232   226  
233   // A completion queued by an earlier registration of the same state 227   // A completion queued by an earlier registration of the same state
234   // may still be running, so the fields it reads change under its 228   // may still be running, so the fields it reads change under its
235   // mutex, and the mutex mode is only written when it differs. 229   // mutex, and the mutex mode is only written when it differs.
HITCBC 236   9792 if (desc->mutex.enabled() != reactor_io_locking_) 230   9845 if (desc->mutex.enabled() != reactor_io_locking_)
HITCBC 237   3 desc->mutex.set_enabled(reactor_io_locking_); 231   3 desc->mutex.set_enabled(reactor_io_locking_);
238   { 232   {
HITCBC 239   9792 conditionally_enabled_mutex::scoped_lock lock(desc->mutex); 233   9845 conditionally_enabled_mutex::scoped_lock lock(desc->mutex);
HITCBC 240   9792 desc->registered_events = reactor_event_read | reactor_event_write; 234   9845 desc->registered_events = reactor_event_read | reactor_event_write;
HITCBC 241   9792 desc->unpollable = false; // the state is reused across adoptions 235   9845 desc->unpollable = false; // the state is reused across adoptions
HITCBC 242   9792 desc->fd = fd; 236   9845 desc->fd = fd;
HITCBC 243   9792 desc->scheduler_ = this; 237   9845 desc->scheduler_ = this;
HITCBC 244   9792 desc->ready_events_.store(0, std::memory_order_release); 238   9845 desc->ready_events_.store(0, std::memory_order_release);
245   // object_ref_ is not touched: a queued invocation from an 239   // object_ref_ is not touched: a queued invocation from an
246   // earlier registration still owns the reference in it. 240   // earlier registration still owns the reference in it.
HITCBC 247   9792 desc->read_ready = false; 241   9845 desc->read_ready = false;
HITCBC 248   9792 desc->write_ready = false; 242   9845 desc->write_ready = false;
HITCBC 249   9792 } 243   9845 }
250   244  
251   { 245   {
HITCBC 252   9792 mutex_type::scoped_lock lock(mutex_); 246   9845 mutex_type::scoped_lock lock(mutex_);
253   try 247   try
254   { 248   {
HITCBC 255   9792 registered_descs_[fd] = desc; 249   9845 registered_descs_[fd] = desc;
256   } 250   }
HITCBC 257   1 catch (std::bad_alloc const&) 251   1 catch (std::bad_alloc const&)
258   { 252   {
HITCBC 259   1 return make_err(ENOMEM); 253   1 return make_err(ENOMEM);
HITCBC 260   1 } 254   1 }
HITCBC 261   9791 if (fd > max_fd_) 255   9844 if (fd > max_fd_)
HITCBC 262   9298 max_fd_ = fd; 256   9363 max_fd_ = fd;
HITCBC 263   9792 } 257   9845 }
264   258  
HITCBC 265   9791 interrupt_reactor(); 259   9844 interrupt_reactor();
HITCBC 266   9791 return {}; 260   9844 return {};
267   } 261   }
268   262  
269   inline void 263   inline void
HITCBC 270   9519 select_scheduler::deregister_descriptor(int fd) const 264   9572 select_scheduler::deregister_descriptor(int fd) const
271   { 265   {
HITCBC 272   9519 mutex_type::scoped_lock lock(mutex_); 266   9572 mutex_type::scoped_lock lock(mutex_);
273   267  
HITCBC 274   9519 auto it = registered_descs_.find(fd); 268   9572 auto it = registered_descs_.find(fd);
HITCBC 275   9519 if (it == registered_descs_.end()) 269   9572 if (it == registered_descs_.end())
MISUBC 276   ✗ return; 270   ✗ return;
277   271  
HITCBC 278   9519 registered_descs_.erase(it); 272   9572 registered_descs_.erase(it);
279   273  
HITCBC 280   9519 if (fd == max_fd_) 274   9572 if (fd == max_fd_)
281   { 275   {
HITCBC 282   7558 max_fd_ = pipe_fds_[0]; 276   7658 max_fd_ = pipe_fds_[0];
HITCBC 283   12588 for (auto& [registered_fd, state] : registered_descs_) 277   12714 for (auto& [registered_fd, state] : registered_descs_)
284   { 278   {
HITCBC 285   5030 if (registered_fd > max_fd_) 279   5056 if (registered_fd > max_fd_)
HITCBC 286   4773 max_fd_ = registered_fd; 280   4823 max_fd_ = registered_fd;
287   } 281   }
288   } 282   }
HITCBC 289   9519 } 283   9572 }
290   284  
291   inline void 285   inline void
HITCBC 292   6630 select_scheduler::notify_reactor() const 286   6764 select_scheduler::notify_reactor() const
293   { 287   {
HITCBC 294   6630 interrupt_reactor(); 288   6764 interrupt_reactor();
HITCBC 295   6630 } 289   6764 }
296   290  
297   inline void 291   inline void
HITCBC 298   25052 select_scheduler::interrupt_reactor() const 292   25515 select_scheduler::interrupt_reactor() const
299   { 293   {
HITCBC 300   25052 char byte = 1; 294   25515 char byte = 1;
HITCBC 301   25052 [[maybe_unused]] auto r = ::write(pipe_fds_[1], &byte, 1); 295   25515 [[maybe_unused]] auto r = ::write(pipe_fds_[1], &byte, 1);
HITCBC 302   25052 } 296   25515 }
303   297  
304   inline long 298   inline long
HITCBC 305   18129 select_scheduler::calculate_timeout(long requested_timeout_us) const 299   17846 select_scheduler::calculate_timeout(long requested_timeout_us) const
306   { 300   {
HITCBC 307   18129 if (requested_timeout_us == 0) 301   17846 if (requested_timeout_us == 0)
308   return 0; // LCOV_EXCL_LINE run_task passes 0 via task_interrupted_, never through this argument 302   return 0; // LCOV_EXCL_LINE run_task passes 0 via task_interrupted_, never through this argument
309   303  
HITCBC 310 - 18129 auto nearest = timer_svc_->nearest_expiry(); 304 + 17846 auto* tsvc = timers();
HITGNC   305 + 17846 if (!tsvc)
HITGNC   306 + 11374 return requested_timeout_us;
HITGNC   307 + 6472 auto nearest = tsvc->nearest_expiry();
HITCBC 311   18129 if (nearest == timer_service::time_point::max()) 308   6472 if (nearest == timer_service::time_point::max())
HITCBC 312   11870 return requested_timeout_us; 309   139 return requested_timeout_us;
313   310  
HITCBC 314   6259 auto now = std::chrono::steady_clock::now(); 311   6333 auto now = std::chrono::steady_clock::now();
HITCBC 315   6259 if (nearest <= now) 312   6333 if (nearest <= now)
HITCBC 316   145 return 0; 313   155 return 0;
317   314  
318   auto timer_timeout_us = 315   auto timer_timeout_us =
HITCBC 319   6114 std::chrono::duration_cast<std::chrono::microseconds>(nearest - now) 316   6178 std::chrono::duration_cast<std::chrono::microseconds>(nearest - now)
HITCBC 320   6114 .count(); 317   6178 .count();
321   318  
HITCBC 322   6114 constexpr auto long_max = 319   6178 constexpr auto long_max =
323   static_cast<long long>((std::numeric_limits<long>::max)()); 320   static_cast<long long>((std::numeric_limits<long>::max)());
324   auto capped_timer_us = 321   auto capped_timer_us =
HITCBC 325   6114 (std::min)((std::max)(static_cast<long long>(timer_timeout_us), 322   6178 (std::min)((std::max)(static_cast<long long>(timer_timeout_us),
HITCBC 326   6114 static_cast<long long>(0)), 323   6178 static_cast<long long>(0)),
HITCBC 327   6114 long_max); 324   6178 long_max);
328   325  
HITCBC 329   6114 if (requested_timeout_us < 0) 326   6178 if (requested_timeout_us < 0)
HITCBC 330   6108 return static_cast<long>(capped_timer_us); 327   6172 return static_cast<long>(capped_timer_us);
331   328  
332   return static_cast<long>( 329   return static_cast<long>(
HITCBC 333   6 (std::min)(static_cast<long long>(requested_timeout_us), 330   6 (std::min)(static_cast<long long>(requested_timeout_us),
HITCBC 334   6 capped_timer_us)); 331   6 capped_timer_us));
335   } 332   }
336   333  
337   inline void 334   inline void
HITCBC 338   148101 select_scheduler::run_task(lock_type& lock, context_type& ctx, long timeout_us) 335   144069 select_scheduler::run_task(lock_type& lock, context_type& ctx, long timeout_us)
339   { 336   {
340   long effective_timeout_us = 337   long effective_timeout_us =
HITCBC 341   148101 task_interrupted_ ? 0 : calculate_timeout(timeout_us); 338   144069 task_interrupted_ ? 0 : calculate_timeout(timeout_us);
342   339  
343   // Snapshot registered descriptors while holding lock. 340   // Snapshot registered descriptors while holding lock.
344   // Record which directions each fd needs monitored to avoid a hot 341   // Record which directions each fd needs monitored to avoid a hot
345   // loop: select is level-triggered, so a writable socket (nearly 342   // loop: select is level-triggered, so a writable socket (nearly
346   // always writable) or an always-readable fd (/dev/zero, a pipe at 343   // always writable) or an always-readable fd (/dev/zero, a pipe at
347   // EOF) would return select() immediately every iteration if 344   // EOF) would return select() immediately every iteration if
348   // unconditionally added. Membership in both sets is opt-in: a 345   // unconditionally added. Membership in both sets is opt-in: a
349   // parked op or wait in a direction opts that direction in. The 346   // parked op or wait in a direction opts that direction in. The
350   // exceptional set is opt-in too, for any parked op or wait: 347   // exceptional set is opt-in too, for any parked op or wait:
351   // Darwin reports a character device (/dev/zero, /dev/null) as 348   // Darwin reports a character device (/dev/zero, /dev/null) as
352   // exceptional on every call. 349   // exceptional on every call.
353   struct fd_entry 350   struct fd_entry
354   { 351   {
355   int fd; 352   int fd;
356   reactor_descriptor_state* desc; 353   reactor_descriptor_state* desc;
357   std::uint32_t want; 354   std::uint32_t want;
358   }; 355   };
359   fd_entry snapshot[FD_SETSIZE]; 356   fd_entry snapshot[FD_SETSIZE];
HITCBC 360   148101 int snapshot_count = 0; 357   144069 int snapshot_count = 0;
361   358  
362   // The caller may have released the lock to hand queued work to a 359   // The caller may have released the lock to hand queued work to a
363   // peer, and deregister_descriptor() erases from this map under it. 360   // peer, and deregister_descriptor() erases from this map under it.
HITCBC 364   148101 if (!lock.owns_lock()) 361   144069 if (!lock.owns_lock())
HITCBC 365   129969 lock.lock(); 362   126220 lock.lock();
366   363  
HITCBC 367   403751 for (auto& [fd, desc] : registered_descs_) 364   398341 for (auto& [fd, desc] : registered_descs_)
368   { 365   {
HITCBC 369   255650 if (snapshot_count < FD_SETSIZE) 366   254272 if (snapshot_count < FD_SETSIZE)
370   { 367   {
HITCBC 371   255650 conditionally_enabled_mutex::scoped_lock desc_lock(desc->mutex); 368   254272 conditionally_enabled_mutex::scoped_lock desc_lock(desc->mutex);
HITCBC 372   255650 snapshot[snapshot_count].fd = fd; 369   254272 snapshot[snapshot_count].fd = fd;
HITCBC 373   255650 snapshot[snapshot_count].desc = desc; 370   254272 snapshot[snapshot_count].desc = desc;
HITCBC 374   255650 snapshot[snapshot_count].want = 371   254272 snapshot[snapshot_count].want =
HITCBC 375   255650 ((desc->read_op || desc->wait_read_op) ? reactor_event_read 372   254272 ((desc->read_op || desc->wait_read_op) ? reactor_event_read
HITCBC 376   255650 : 0) | 373   254272 : 0) |
HITCBC 377   255270 ((desc->write_op || desc->connect_op || desc->wait_write_op) 374   253898 ((desc->write_op || desc->connect_op || desc->wait_write_op)
HITCBC 378   510920 ? reactor_event_write 375   508170 ? reactor_event_write
HITCBC 379   255650 : 0) | 376   254272 : 0) |
HITCBC 380   255650 (desc->wait_error_op ? reactor_event_error : 0); 377   254272 (desc->wait_error_op ? reactor_event_error : 0);
HITCBC 381   255650 ++snapshot_count; 378   254272 ++snapshot_count;
HITCBC 382   255650 } 379   254272 }
383   } 380   }
384   381  
HITCBC 385   148101 if (lock.owns_lock()) 382   144069 if (lock.owns_lock())
HITCBC 386   148100 lock.unlock(); 383   144068 lock.unlock();
387   384  
HITCBC 388   148101 task_cleanup on_exit{this, &lock, ctx}; 385   144069 task_cleanup on_exit{this, &lock, ctx};
389   386  
390   fd_set read_fds, write_fds, except_fds; 387   fd_set read_fds, write_fds, except_fds;
HITCBC 391   2517717 FD_ZERO(&read_fds); 388   2449173 FD_ZERO(&read_fds);
HITCBC 392   2517717 FD_ZERO(&write_fds); 389   2449173 FD_ZERO(&write_fds);
HITCBC 393   2517717 FD_ZERO(&except_fds); 390   2449173 FD_ZERO(&except_fds);
394   391  
HITCBC 395   148101 FD_SET(pipe_fds_[0], &read_fds); 392   144069 FD_SET(pipe_fds_[0], &read_fds);
HITCBC 396   148101 int nfds = pipe_fds_[0]; 393   144069 int nfds = pipe_fds_[0];
397   394  
HITCBC 398   403751 for (int i = 0; i < snapshot_count; ++i) 395   398341 for (int i = 0; i < snapshot_count; ++i)
399   { 396   {
HITCBC 400   255650 int fd = snapshot[i].fd; 397   254272 int fd = snapshot[i].fd;
HITCBC 401   255650 if (snapshot[i].want & reactor_event_read) 398   254272 if (snapshot[i].want & reactor_event_read)
HITCBC 402   119144 FD_SET(fd, &read_fds); 399   117409 FD_SET(fd, &read_fds);
HITCBC 403   255650 if (snapshot[i].want & reactor_event_write) 400   254272 if (snapshot[i].want & reactor_event_write)
HITCBC 404   2950 FD_SET(fd, &write_fds); 401   2973 FD_SET(fd, &write_fds);
HITCBC 405   255650 if (snapshot[i].want != 0) 402   254272 if (snapshot[i].want != 0)
HITCBC 406   122384 FD_SET(fd, &except_fds); 403   120577 FD_SET(fd, &except_fds);
HITCBC 407   255650 if (fd > nfds) 404   254272 if (fd > nfds)
HITCBC 408   145938 nfds = fd; 405   142229 nfds = fd;
409   } 406   }
410   407  
411   struct timeval tv; 408   struct timeval tv;
HITCBC 412   148101 struct timeval* tv_ptr = nullptr; 409   144069 struct timeval* tv_ptr = nullptr;
HITCBC 413   148101 if (effective_timeout_us >= 0) 410   144069 if (effective_timeout_us >= 0)
414   { 411   {
HITCBC 415   136812 tv.tv_sec = effective_timeout_us / 1000000; 412   133204 tv.tv_sec = effective_timeout_us / 1000000;
HITCBC 416   136812 tv.tv_usec = effective_timeout_us % 1000000; 413   133204 tv.tv_usec = effective_timeout_us % 1000000;
HITCBC 417   136812 tv_ptr = &tv; 414   133204 tv_ptr = &tv;
418   } 415   }
419   416  
HITCBC 420   148101 int ready = ::select(nfds + 1, &read_fds, &write_fds, &except_fds, tv_ptr); 417   144069 int ready = ::select(nfds + 1, &read_fds, &write_fds, &except_fds, tv_ptr);
421   418  
422   // EINTR: signal interrupted select(), just retry. 419   // EINTR: signal interrupted select(), just retry.
423   // EBADF: an fd was closed between snapshot and select(); retry 420   // EBADF: an fd was closed between snapshot and select(); retry
424   // with a fresh snapshot from registered_descs_. 421   // with a fresh snapshot from registered_descs_.
425   // Both fall through with no ready descriptors rather than 422   // Both fall through with no ready descriptors rather than
426   // returning: the caller handed this function an owned lock that 423   // returning: the caller handed this function an owned lock that
427   // only the epilogue below re-acquires. 424   // only the epilogue below re-acquires.
HITCBC 428   148101 if (ready < 0) 425   144069 if (ready < 0)
429   { 426   {
HITCBC 430   3 if (errno != EINTR && errno != EBADF) 427   3 if (errno != EINTR && errno != EBADF)
HITCBC 431   1 detail::throw_system_error(make_err(errno), "select"); 428   1 detail::throw_system_error(make_err(errno), "select");
HITCBC 432   2 ready = 0; 429   2 ready = 0;
433   } 430   }
434   431  
435   // Process timers outside the lock 432   // Process timers outside the lock
HITCBC 436 - 148100 timer_svc_->process_expired(); 433 + 144068 if (auto* tsvc = timers())
HITGNC   434 + 27882 tsvc->process_expired();
437   435  
HITCBC 438   148100 ready_queue local_ops; 436   144068 ready_queue local_ops;
439   437  
HITCBC 440   148100 if (ready > 0) 438   144068 if (ready > 0)
441   { 439   {
HITCBC 442   18447 if (FD_ISSET(pipe_fds_[0], &read_fds)) 440   18344 if (FD_ISSET(pipe_fds_[0], &read_fds))
443   { 441   {
444   char buf[256]; 442   char buf[256];
HITCBC 445   21978 while (::read(pipe_fds_[0], buf, sizeof(buf)) > 0) 443   22660 while (::read(pipe_fds_[0], buf, sizeof(buf)) > 0)
446   { 444   {
447   } 445   }
448   } 446   }
449   447  
HITCBC 450   104799 for (int i = 0; i < snapshot_count; ++i) 448   105795 for (int i = 0; i < snapshot_count; ++i)
451   { 449   {
HITCBC 452   86352 int fd = snapshot[i].fd; 450   87451 int fd = snapshot[i].fd;
HITCBC 453   86352 reactor_descriptor_state* desc = snapshot[i].desc; 451   87451 reactor_descriptor_state* desc = snapshot[i].desc;
454   452  
HITCBC 455   86352 std::uint32_t flags = 0; 453   87451 std::uint32_t flags = 0;
HITCBC 456   86352 if (FD_ISSET(fd, &read_fds)) 454   87451 if (FD_ISSET(fd, &read_fds))
HITCBC 457   10114 flags |= reactor_event_read; 455   9725 flags |= reactor_event_read;
HITCBC 458   86352 if (FD_ISSET(fd, &write_fds)) 456   87451 if (FD_ISSET(fd, &write_fds))
HITCBC 459   2656 flags |= reactor_event_write; 457   2689 flags |= reactor_event_write;
HITCBC 460   86352 if (FD_ISSET(fd, &except_fds)) 458   87451 if (FD_ISSET(fd, &except_fds))
HITCBC 461   10 flags |= reactor_event_error; 459   10 flags |= reactor_event_error;
462   460  
HITCBC 463   86352 if (flags == 0) 461   87451 if (flags == 0)
HITCBC 464   73580 continue; 462   75035 continue;
465   463  
HITCBC 466   12772 desc->add_ready_events(flags); 464   12416 desc->add_ready_events(flags);
467   465  
HITCBC 468   12772 bool expected = false; 466   12416 bool expected = false;
HITCBC 469   12772 if (desc->is_enqueued_.compare_exchange_strong( 467   12416 if (desc->is_enqueued_.compare_exchange_strong(
470   expected, true, std::memory_order_release, 468   expected, true, std::memory_order_release,
471   std::memory_order_relaxed)) 469   std::memory_order_relaxed))
472   { 470   {
HITCBC 473   6047 local_ops.push(desc); 471   6215 local_ops.push(desc);
474   } 472   }
475   } 473   }
476   } 474   }
477   475  
HITCBC 478   148100 lock.lock(); 476   144068 lock.lock();
479   477  
HITCBC 480   148100 completed_ops_.splice(local_ops); 478   144068 completed_ops_.splice(local_ops);
HITCBC 481   148101 } 479   144069 }
482   480  
483   } // namespace boost::corosio::detail 481   } // namespace boost::corosio::detail
484   482  
485   #endif // BOOST_COROSIO_HAS_SELECT 483   #endif // BOOST_COROSIO_HAS_SELECT
486   484  
487   #endif // BOOST_COROSIO_NATIVE_DETAIL_SELECT_SELECT_SCHEDULER_HPP 485   #endif // BOOST_COROSIO_NATIVE_DETAIL_SELECT_SELECT_SCHEDULER_HPP