. /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   2806 ensure_write_registered(int, reactor_descriptor_state*) const noexcept 118   2754 ensure_write_registered(int, reactor_descriptor_state*) const noexcept
119   { 119   {
HITCBC 120   2806 return {}; 120   2754 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   274 [[nodiscard]] std::error_code register_signal_reader(int read_fd) override 138   274 [[nodiscard]] std::error_code register_signal_reader(int read_fd) override
139   { 139   {
HITCBC 140   274 return register_descriptor(read_fd, signal_pipe_reader_.arm()); 140   274 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   482 void notify_reparked() const override 148   505 void notify_reparked() const override
149   { 149   {
HITCBC 150   482 interrupt_reactor(); 150   505 interrupt_reactor();
HITCBC 151   482 } 151   505 }
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   1993 inline select_scheduler::select_scheduler(capy::execution_context&, int) 167   1993 inline select_scheduler::select_scheduler(capy::execution_context&, int)
HITCBC 168   1993 : pipe_fds_{-1, -1} 168   1993 : pipe_fds_{-1, -1}
HITCBC 169   1993 , max_fd_(-1) 169   1993 , max_fd_(-1)
170   { 170   {
HITCBC 171   1993 if (::pipe(pipe_fds_) < 0) 171   1993 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   5967 for (int i = 0; i < 2; ++i) 174   5967 for (int i = 0; i < 2; ++i)
175   { 175   {
HITCBC 176   3981 int flags = ::fcntl(pipe_fds_[i], F_GETFL, 0); 176   3981 int flags = ::fcntl(pipe_fds_[i], F_GETFL, 0);
HITCBC 177   3981 if (flags == -1) 177   3981 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   3979 if (::fcntl(pipe_fds_[i], F_SETFL, flags | O_NONBLOCK) == -1) 184   3979 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   3977 if (::fcntl(pipe_fds_[i], F_SETFD, FD_CLOEXEC) == -1) 191   3977 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   199  
HITCBC 200   1986 completed_ops_.push(&task_op_); 200   1986 completed_ops_.push(&task_op_);
HITCBC 201   2007 } 201   2007 }
202   202  
HITCBC 203   3972 inline select_scheduler::~select_scheduler() 203   3972 inline select_scheduler::~select_scheduler()
204   { 204   {
HITCBC 205   1986 if (pipe_fds_[0] >= 0) 205   1986 if (pipe_fds_[0] >= 0)
HITCBC 206   1986 ::close(pipe_fds_[0]); 206   1986 ::close(pipe_fds_[0]);
HITCBC 207   1986 if (pipe_fds_[1] >= 0) 207   1986 if (pipe_fds_[1] >= 0)
HITCBC 208   1986 ::close(pipe_fds_[1]); 208   1986 ::close(pipe_fds_[1]);
HITCBC 209   3972 } 209   3972 }
210   210  
211   inline void 211   inline void
HITCBC 212   1986 select_scheduler::shutdown() 212   1986 select_scheduler::shutdown()
213   { 213   {
HITCBC 214   1986 shutdown_drain(); 214   1986 shutdown_drain();
215   215  
HITCBC 216   1986 if (pipe_fds_[1] >= 0) 216   1986 if (pipe_fds_[1] >= 0)
HITCBC 217   1986 interrupt_reactor(); 217   1986 interrupt_reactor();
HITCBC 218   1986 } 218   1986 }
219   219  
220   inline std::error_code 220   inline std::error_code
HITCBC 221   10077 select_scheduler::register_descriptor( 221   10073 select_scheduler::register_descriptor(
222   int fd, reactor_descriptor_state* desc) const 222   int fd, reactor_descriptor_state* desc) const
223   { 223   {
HITCBC 224   10077 if (fd < 0 || fd >= FD_SETSIZE) 224   10073 if (fd < 0 || fd >= FD_SETSIZE)
HITCBC 225   1 return make_err(EMFILE); 225   1 return make_err(EMFILE);
226   226  
227   // A completion queued by an earlier registration of the same state 227   // A completion queued by an earlier registration of the same state
228   // 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
229   // mutex, and the mutex mode is only written when it differs. 229   // mutex, and the mutex mode is only written when it differs.
HITCBC 230   10076 if (desc->mutex.enabled() != reactor_io_locking_) 230   10072 if (desc->mutex.enabled() != reactor_io_locking_)
HITCBC 231   3 desc->mutex.set_enabled(reactor_io_locking_); 231   3 desc->mutex.set_enabled(reactor_io_locking_);
232   { 232   {
HITCBC 233   10076 conditionally_enabled_mutex::scoped_lock lock(desc->mutex); 233   10072 conditionally_enabled_mutex::scoped_lock lock(desc->mutex);
HITCBC 234   10076 desc->registered_events = reactor_event_read | reactor_event_write; 234   10072 desc->registered_events = reactor_event_read | reactor_event_write;
HITCBC 235   10076 desc->unpollable = false; // the state is reused across adoptions 235   10072 desc->unpollable = false; // the state is reused across adoptions
HITCBC 236   10076 desc->fd = fd; 236   10072 desc->fd = fd;
HITCBC 237   10076 desc->scheduler_ = this; 237   10072 desc->scheduler_ = this;
HITCBC 238   10076 desc->ready_events_.store(0, std::memory_order_release); 238   10072 desc->ready_events_.store(0, std::memory_order_release);
239   // object_ref_ is not touched: a queued invocation from an 239   // object_ref_ is not touched: a queued invocation from an
240   // earlier registration still owns the reference in it. 240   // earlier registration still owns the reference in it.
HITCBC 241   10076 desc->read_ready = false; 241   10072 desc->read_ready = false;
HITCBC 242   10076 desc->write_ready = false; 242   10072 desc->write_ready = false;
HITCBC 243   10076 } 243   10072 }
244   244  
245   { 245   {
HITCBC 246   10076 mutex_type::scoped_lock lock(mutex_); 246   10072 mutex_type::scoped_lock lock(mutex_);
247   try 247   try
248   { 248   {
HITCBC 249   10076 registered_descs_[fd] = desc; 249   10072 registered_descs_[fd] = desc;
250   } 250   }
HITCBC 251   1 catch (std::bad_alloc const&) 251   1 catch (std::bad_alloc const&)
252   { 252   {
HITCBC 253   1 return make_err(ENOMEM); 253   1 return make_err(ENOMEM);
HITCBC 254   1 } 254   1 }
HITCBC 255   10075 if (fd > max_fd_) 255   10071 if (fd > max_fd_)
HITCBC 256   9533 max_fd_ = fd; 256   9511 max_fd_ = fd;
HITCBC 257   10076 } 257   10072 }
258   258  
HITCBC 259   10075 interrupt_reactor(); 259   10071 interrupt_reactor();
HITCBC 260   10075 return {}; 260   10071 return {};
261   } 261   }
262   262  
263   inline void 263   inline void
HITCBC 264   9802 select_scheduler::deregister_descriptor(int fd) const 264   9798 select_scheduler::deregister_descriptor(int fd) const
265   { 265   {
HITCBC 266   9802 mutex_type::scoped_lock lock(mutex_); 266   9798 mutex_type::scoped_lock lock(mutex_);
267   267  
HITCBC 268   9802 auto it = registered_descs_.find(fd); 268   9798 auto it = registered_descs_.find(fd);
HITCBC 269   9802 if (it == registered_descs_.end()) 269   9798 if (it == registered_descs_.end())
MISUBC 270   ✗ return; 270   ✗ return;
271   271  
HITCBC 272   9802 registered_descs_.erase(it); 272   9798 registered_descs_.erase(it);
273   273  
HITCBC 274   9802 if (fd == max_fd_) 274   9798 if (fd == max_fd_)
275   { 275   {
HITCBC 276   7937 max_fd_ = pipe_fds_[0]; 276   7869 max_fd_ = pipe_fds_[0];
HITCBC 277   13313 for (auto& [registered_fd, state] : registered_descs_) 277   13213 for (auto& [registered_fd, state] : registered_descs_)
278   { 278   {
HITCBC 279   5376 if (registered_fd > max_fd_) 279   5344 if (registered_fd > max_fd_)
HITCBC 280   5103 max_fd_ = registered_fd; 280   5053 max_fd_ = registered_fd;
281   } 281   }
282   } 282   }
HITCBC 283   9802 } 283   9798 }
284   284  
285   inline void 285   inline void
HITCBC 286   6910 select_scheduler::notify_reactor() const 286   6809 select_scheduler::notify_reactor() const
287   { 287   {
HITCBC 288   6910 interrupt_reactor(); 288   6809 interrupt_reactor();
HITCBC 289   6910 } 289   6809 }
290   290  
291   inline void 291   inline void
HITCBC 292   25951 select_scheduler::interrupt_reactor() const 292   25780 select_scheduler::interrupt_reactor() const
293   { 293   {
HITCBC 294   25951 char byte = 1; 294   25780 char byte = 1;
HITCBC 295   25951 [[maybe_unused]] auto r = ::write(pipe_fds_[1], &byte, 1); 295   25780 [[maybe_unused]] auto r = ::write(pipe_fds_[1], &byte, 1);
HITCBC 296   25951 } 296   25780 }
297   297  
298   inline long 298   inline long
HITCBC 299   12981 select_scheduler::calculate_timeout(long requested_timeout_us) const 299   26373 select_scheduler::calculate_timeout(long requested_timeout_us) const
300   { 300   {
HITCBC 301   12981 if (requested_timeout_us == 0) 301   26373 if (requested_timeout_us == 0)
302   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
303   303  
HITCBC 304   12981 auto* tsvc = timers(); 304   26373 auto* tsvc = timers();
HITCBC 305   12981 if (!tsvc) 305   26373 if (!tsvc)
HITCBC 306   6141 return requested_timeout_us; 306   19714 return requested_timeout_us;
HITCBC 307   6840 auto nearest = tsvc->nearest_expiry(); 307   6659 auto nearest = tsvc->nearest_expiry();
HITCBC 308   6840 if (nearest == timer_service::time_point::max()) 308   6659 if (nearest == timer_service::time_point::max())
HITCBC 309   136 return requested_timeout_us; 309   137 return requested_timeout_us;
310   310  
HITCBC 311   6704 auto now = std::chrono::steady_clock::now(); 311   6522 auto now = std::chrono::steady_clock::now();
HITCBC 312   6704 if (nearest <= now) 312   6522 if (nearest <= now)
HITCBC 313   51 return 0; 313   89 return 0;
314   314  
315   auto timer_timeout_us = 315   auto timer_timeout_us =
HITCBC 316   6653 std::chrono::duration_cast<std::chrono::microseconds>(nearest - now) 316   6433 std::chrono::duration_cast<std::chrono::microseconds>(nearest - now)
HITCBC 317   6653 .count(); 317   6433 .count();
318   318  
HITCBC 319   6653 constexpr auto long_max = 319   6433 constexpr auto long_max =
320   static_cast<long long>((std::numeric_limits<long>::max)()); 320   static_cast<long long>((std::numeric_limits<long>::max)());
321   auto capped_timer_us = 321   auto capped_timer_us =
HITCBC 322   6653 (std::min)((std::max)(static_cast<long long>(timer_timeout_us), 322   6433 (std::min)((std::max)(static_cast<long long>(timer_timeout_us),
HITCBC 323   6653 static_cast<long long>(0)), 323   6433 static_cast<long long>(0)),
HITCBC 324   6653 long_max); 324   6433 long_max);
325   325  
HITCBC 326   6653 if (requested_timeout_us < 0) 326   6433 if (requested_timeout_us < 0)
HITCBC 327   6647 return static_cast<long>(capped_timer_us); 327   6427 return static_cast<long>(capped_timer_us);
328   328  
329   return static_cast<long>( 329   return static_cast<long>(
HITCBC 330   6 (std::min)(static_cast<long long>(requested_timeout_us), 330   6 (std::min)(static_cast<long long>(requested_timeout_us),
HITCBC 331   6 capped_timer_us)); 331   6 capped_timer_us));
332   } 332   }
333   333  
334   inline void 334   inline void
HITCBC 335   140644 select_scheduler::run_task(lock_type& lock, context_type& ctx, long timeout_us) 335   152410 select_scheduler::run_task(lock_type& lock, context_type& ctx, long timeout_us)
336   { 336   {
337   long effective_timeout_us = 337   long effective_timeout_us =
HITCBC 338   140644 task_interrupted_ ? 0 : calculate_timeout(timeout_us); 338   152410 task_interrupted_ ? 0 : calculate_timeout(timeout_us);
339   339  
340   // Snapshot registered descriptors while holding lock. 340   // Snapshot registered descriptors while holding lock.
341   // Record which directions each fd needs monitored to avoid a hot 341   // Record which directions each fd needs monitored to avoid a hot
342   // loop: select is level-triggered, so a writable socket (nearly 342   // loop: select is level-triggered, so a writable socket (nearly
343   // 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
344   // EOF) would return select() immediately every iteration if 344   // EOF) would return select() immediately every iteration if
345   // unconditionally added. Membership in both sets is opt-in: a 345   // unconditionally added. Membership in both sets is opt-in: a
346   // 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
347   // 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:
348   // Darwin reports a character device (/dev/zero, /dev/null) as 348   // Darwin reports a character device (/dev/zero, /dev/null) as
349   // exceptional on every call. 349   // exceptional on every call.
350   struct fd_entry 350   struct fd_entry
351   { 351   {
352   int fd; 352   int fd;
353   reactor_descriptor_state* desc; 353   reactor_descriptor_state* desc;
354   std::uint32_t want; 354   std::uint32_t want;
355   }; 355   };
356   fd_entry snapshot[FD_SETSIZE]; 356   fd_entry snapshot[FD_SETSIZE];
HITCBC 357   140644 int snapshot_count = 0; 357   152410 int snapshot_count = 0;
358   358  
359   // 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
360   // peer, and deregister_descriptor() erases from this map under it. 360   // peer, and deregister_descriptor() erases from this map under it.
HITCBC 361   140644 if (!lock.owns_lock()) 361   152410 if (!lock.owns_lock())
HITCBC 362   127660 lock.lock(); 362   126034 lock.lock();
363   363  
HITCBC 364   392174 for (auto& [fd, desc] : registered_descs_) 364   411676 for (auto& [fd, desc] : registered_descs_)
365   { 365   {
HITCBC 366   251530 if (snapshot_count < FD_SETSIZE) 366   259266 if (snapshot_count < FD_SETSIZE)
367   { 367   {
HITCBC 368   251530 conditionally_enabled_mutex::scoped_lock desc_lock(desc->mutex); 368   259266 conditionally_enabled_mutex::scoped_lock desc_lock(desc->mutex);
HITCBC 369   251530 snapshot[snapshot_count].fd = fd; 369   259266 snapshot[snapshot_count].fd = fd;
HITCBC 370   251530 snapshot[snapshot_count].desc = desc; 370   259266 snapshot[snapshot_count].desc = desc;
HITCBC 371   251530 snapshot[snapshot_count].want = 371   259266 snapshot[snapshot_count].want =
HITCBC 372   251530 ((desc->read_op || desc->wait_read_op) ? reactor_event_read 372   259266 ((desc->read_op || desc->wait_read_op) ? reactor_event_read
HITCBC 373   251530 : 0) | 373   259266 : 0) |
HITCBC 374   251156 ((desc->write_op || desc->connect_op || desc->wait_write_op) 374   258892 ((desc->write_op || desc->connect_op || desc->wait_write_op)
HITCBC 375   502686 ? reactor_event_write 375   518158 ? reactor_event_write
HITCBC 376   251530 : 0) | 376   259266 : 0) |
HITCBC 377   251530 (desc->wait_error_op ? reactor_event_error : 0); 377   259266 (desc->wait_error_op ? reactor_event_error : 0);
HITCBC 378   251530 ++snapshot_count; 378   259266 ++snapshot_count;
HITCBC 379   251530 } 379   259266 }
380   } 380   }
381   381  
HITCBC 382   140644 if (lock.owns_lock()) 382   152410 if (lock.owns_lock())
HITCBC 383   140643 lock.unlock(); 383   152409 lock.unlock();
384   384  
HITCBC 385   140644 task_cleanup on_exit{this, &lock, ctx}; 385   152410 task_cleanup on_exit{this, &lock, ctx};
386   386  
387   fd_set read_fds, write_fds, except_fds; 387   fd_set read_fds, write_fds, except_fds;
HITCBC 388   2390948 FD_ZERO(&read_fds); 388   2590970 FD_ZERO(&read_fds);
HITCBC 389   2390948 FD_ZERO(&write_fds); 389   2590970 FD_ZERO(&write_fds);
HITCBC 390   2390948 FD_ZERO(&except_fds); 390   2590970 FD_ZERO(&except_fds);
391   391  
HITCBC 392   140644 FD_SET(pipe_fds_[0], &read_fds); 392   152410 FD_SET(pipe_fds_[0], &read_fds);
HITCBC 393   140644 int nfds = pipe_fds_[0]; 393   152410 int nfds = pipe_fds_[0];
394   394  
HITCBC 395   392174 for (int i = 0; i < snapshot_count; ++i) 395   411676 for (int i = 0; i < snapshot_count; ++i)
396   { 396   {
HITCBC 397   251530 int fd = snapshot[i].fd; 397   259266 int fd = snapshot[i].fd;
HITCBC 398   251530 if (snapshot[i].want & reactor_event_read) 398   259266 if (snapshot[i].want & reactor_event_read)
HITCBC 399   112198 FD_SET(fd, &read_fds); 399   125582 FD_SET(fd, &read_fds);
HITCBC 400   251530 if (snapshot[i].want & reactor_event_write) 400   259266 if (snapshot[i].want & reactor_event_write)
HITCBC 401   3070 FD_SET(fd, &write_fds); 401   3018 FD_SET(fd, &write_fds);
HITCBC 402   251530 if (snapshot[i].want != 0) 402   259266 if (snapshot[i].want != 0)
HITCBC 403   115463 FD_SET(fd, &except_fds); 403   128795 FD_SET(fd, &except_fds);
HITCBC 404   251530 if (fd > nfds) 404   259266 if (fd > nfds)
HITCBC 405   138620 nfds = fd; 405   150387 nfds = fd;
406   } 406   }
407   407  
408   struct timeval tv; 408   struct timeval tv;
HITCBC 409   140644 struct timeval* tv_ptr = nullptr; 409   152410 struct timeval* tv_ptr = nullptr;
HITCBC 410   140644 if (effective_timeout_us >= 0) 410   152410 if (effective_timeout_us >= 0)
411   { 411   {
HITCBC 412   134959 tv.tv_sec = effective_timeout_us / 1000000; 412   133227 tv.tv_sec = effective_timeout_us / 1000000;
HITCBC 413   134959 tv.tv_usec = effective_timeout_us % 1000000; 413   133227 tv.tv_usec = effective_timeout_us % 1000000;
HITCBC 414   134959 tv_ptr = &tv; 414   133227 tv_ptr = &tv;
415   } 415   }
416   416  
HITCBC 417   140644 int ready = ::select(nfds + 1, &read_fds, &write_fds, &except_fds, tv_ptr); 417   152410 int ready = ::select(nfds + 1, &read_fds, &write_fds, &except_fds, tv_ptr);
418   418  
419   // EINTR: signal interrupted select(), just retry. 419   // EINTR: signal interrupted select(), just retry.
420   // EBADF: an fd was closed between snapshot and select(); retry 420   // EBADF: an fd was closed between snapshot and select(); retry
421   // with a fresh snapshot from registered_descs_. 421   // with a fresh snapshot from registered_descs_.
422   // Both fall through with no ready descriptors rather than 422   // Both fall through with no ready descriptors rather than
423   // returning: the caller handed this function an owned lock that 423   // returning: the caller handed this function an owned lock that
424   // only the epilogue below re-acquires. 424   // only the epilogue below re-acquires.
HITCBC 425   140644 if (ready < 0) 425   152410 if (ready < 0)
426   { 426   {
HITCBC 427   3 if (errno != EINTR && errno != EBADF) 427   4 if (errno != EINTR && errno != EBADF)
HITCBC 428   1 detail::throw_system_error(make_err(errno), "select"); 428   1 detail::throw_system_error(make_err(errno), "select");
HITCBC 429   2 ready = 0; 429   3 ready = 0;
430   } 430   }
431   431  
432   // Process timers outside the lock 432   // Process timers outside the lock
HITCBC 433   140643 if (auto* tsvc = timers()) 433   152409 if (auto* tsvc = timers())
HITCBC 434   29582 tsvc->process_expired(); 434   28051 tsvc->process_expired();
435   435  
HITCBC 436   140643 ready_queue local_ops; 436   152409 ready_queue local_ops;
437   437  
HITCBC 438   140643 if (ready > 0) 438   152409 if (ready > 0)
439   { 439   {
HITCBC 440   13288 if (FD_ISSET(pipe_fds_[0], &read_fds)) 440   26837 if (FD_ISSET(pipe_fds_[0], &read_fds))
441   { 441   {
442   char buf[256]; 442   char buf[256];
HITCBC 443   22681 while (::read(pipe_fds_[0], buf, sizeof(buf)) > 0) 443   22616 while (::read(pipe_fds_[0], buf, sizeof(buf)) > 0)
444   { 444   {
445   } 445   }
446   } 446   }
447   447  
HITCBC 448   94363 for (int i = 0; i < snapshot_count; ++i) 448   121373 for (int i = 0; i < snapshot_count; ++i)
449   { 449   {
HITCBC 450   81075 int fd = snapshot[i].fd; 450   94536 int fd = snapshot[i].fd;
HITCBC 451   81075 reactor_descriptor_state* desc = snapshot[i].desc; 451   94536 reactor_descriptor_state* desc = snapshot[i].desc;
452   452  
HITCBC 453   81075 std::uint32_t flags = 0; 453   94536 std::uint32_t flags = 0;
HITCBC 454   81075 if (FD_ISSET(fd, &read_fds)) 454   94536 if (FD_ISSET(fd, &read_fds))
HITCBC 455   4870 flags |= reactor_event_read; 455   18341 flags |= reactor_event_read;
HITCBC 456   81075 if (FD_ISSET(fd, &write_fds)) 456   94536 if (FD_ISSET(fd, &write_fds))
HITCBC 457   2786 flags |= reactor_event_write; 457   2734 flags |= reactor_event_write;
HITCBC 458   81075 if (FD_ISSET(fd, &except_fds)) 458   94536 if (FD_ISSET(fd, &except_fds))
HITCBC 459   14 flags |= reactor_event_error; 459   15 flags |= reactor_event_error;
460   460  
HITCBC 461   81075 if (flags == 0) 461   94536 if (flags == 0)
HITCBC 462   73417 continue; 462   73459 continue;
463   463  
HITCBC 464   7658 desc->add_ready_events(flags); 464   21077 desc->add_ready_events(flags);
465   465  
HITCBC 466   7658 bool expected = false; 466   21077 bool expected = false;
HITCBC 467   7658 if (desc->is_enqueued_.compare_exchange_strong( 467   21077 if (desc->is_enqueued_.compare_exchange_strong(
468   expected, true, std::memory_order_release, 468   expected, true, std::memory_order_release,
469   std::memory_order_relaxed)) 469   std::memory_order_relaxed))
470   { 470   {
HITCBC 471   6428 local_ops.push(desc); 471   6354 local_ops.push(desc);
472   } 472   }
473   } 473   }
474   } 474   }
475   475  
HITCBC 476   140643 lock.lock(); 476   152409 lock.lock();
477   477  
HITCBC 478   140643 completed_ops_.splice(local_ops); 478   152409 completed_ops_.splice(local_ops);
HITCBC 479   140644 } 479   152410 }
480   480  
481   } // namespace boost::corosio::detail 481   } // namespace boost::corosio::detail
482   482  
483   #endif // BOOST_COROSIO_HAS_SELECT 483   #endif // BOOST_COROSIO_HAS_SELECT
484   484  
485   #endif // BOOST_COROSIO_NATIVE_DETAIL_SELECT_SELECT_SCHEDULER_HPP 485   #endif // BOOST_COROSIO_NATIVE_DETAIL_SELECT_SELECT_SCHEDULER_HPP