. /include/boost/corosio /native /detail /select /select_scheduler.hpp
99.44% Lines (178/179) 100.00% Functions (13/13)
[ ] 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   2724 ensure_write_registered(int, reactor_descriptor_state*) const noexcept 118   2727 ensure_write_registered(int, reactor_descriptor_state*) const noexcept
119   { 119   {
HITCBC 120   2724 return {}; 120   2727 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   272 [[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   272 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   416 void notify_reparked() const override 148   411 void notify_reparked() const override
149   { 149   {
HITCBC 150   416 interrupt_reactor(); 150   411 interrupt_reactor();
HITCBC 151   416 } 151   411 }
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 - 1958 inline select_scheduler::select_scheduler(capy::execution_context&, int) 167 + 1953 inline select_scheduler::select_scheduler(capy::execution_context& ctx, int)
HITCBC 168   1958 : pipe_fds_{-1, -1} 168   1953 : pipe_fds_{-1, -1}
HITCBC 169   1958 , max_fd_(-1) 169   1953 , max_fd_(-1)
170   { 170   {
HITCBC 171   1958 if (::pipe(pipe_fds_) < 0) 171   1953 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   5862 for (int i = 0; i < 2; ++i) 174   5847 for (int i = 0; i < 2; ++i)
175   { 175   {
HITCBC 176   3911 int flags = ::fcntl(pipe_fds_[i], F_GETFL, 0); 176   3901 int flags = ::fcntl(pipe_fds_[i], F_GETFL, 0);
HITCBC 177   3911 if (flags == -1) 177   3901 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   3909 if (::fcntl(pipe_fds_[i], F_SETFL, flags | O_NONBLOCK) == -1) 184   3899 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   3907 if (::fcntl(pipe_fds_[i], F_SETFD, FD_CLOEXEC) == -1) 191   3897 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  
HITGNC   200 + 1946 timer_svc_ = &get_timer_service(ctx, *this);
HITGNC   201 + 1946 timer_svc_->set_on_earliest_changed(
HITGNC   202 + 4981 timer_service::callback(this, [](void* p) {
HITGNC   203 + 3035 static_cast<select_scheduler*>(p)->interrupt_reactor();
HITGNC   204 + 3035 }));
  205 +
HITCBC 200   1951 completed_ops_.push(&task_op_); 206   1946 completed_ops_.push(&task_op_);
HITCBC 201   1972 } 207   1967 }
202   208  
HITCBC 203   3902 inline select_scheduler::~select_scheduler() 209   3892 inline select_scheduler::~select_scheduler()
204   { 210   {
HITCBC 205   1951 if (pipe_fds_[0] >= 0) 211   1946 if (pipe_fds_[0] >= 0)
HITCBC 206   1951 ::close(pipe_fds_[0]); 212   1946 ::close(pipe_fds_[0]);
HITCBC 207   1951 if (pipe_fds_[1] >= 0) 213   1946 if (pipe_fds_[1] >= 0)
HITCBC 208   1951 ::close(pipe_fds_[1]); 214   1946 ::close(pipe_fds_[1]);
HITCBC 209   3902 } 215   3892 }
210   216  
211   inline void 217   inline void
HITCBC 212   1951 select_scheduler::shutdown() 218   1946 select_scheduler::shutdown()
213   { 219   {
HITCBC 214   1951 shutdown_drain(); 220   1946 shutdown_drain();
215   221  
HITCBC 216   1951 if (pipe_fds_[1] >= 0) 222   1946 if (pipe_fds_[1] >= 0)
HITCBC 217   1951 interrupt_reactor(); 223   1946 interrupt_reactor();
HITCBC 218   1951 } 224   1946 }
219   225  
220   inline std::error_code 226   inline std::error_code
HITCBC 221   9759 select_scheduler::register_descriptor( 227   9761 select_scheduler::register_descriptor(
222   int fd, reactor_descriptor_state* desc) const 228   int fd, reactor_descriptor_state* desc) const
223   { 229   {
HITCBC 224   9759 if (fd < 0 || fd >= FD_SETSIZE) 230   9761 if (fd < 0 || fd >= FD_SETSIZE)
HITCBC 225   1 return make_err(EMFILE); 231   1 return make_err(EMFILE);
226   232  
227   // A completion queued by an earlier registration of the same state 233   // A completion queued by an earlier registration of the same state
228   // may still be running, so the fields it reads change under its 234   // may still be running, so the fields it reads change under its
229   // mutex, and the mutex mode is only written when it differs. 235   // mutex, and the mutex mode is only written when it differs.
HITCBC 230   9758 if (desc->mutex.enabled() != reactor_io_locking_) 236   9760 if (desc->mutex.enabled() != reactor_io_locking_)
HITCBC 231   3 desc->mutex.set_enabled(reactor_io_locking_); 237   3 desc->mutex.set_enabled(reactor_io_locking_);
232   { 238   {
HITCBC 233   9758 conditionally_enabled_mutex::scoped_lock lock(desc->mutex); 239   9760 conditionally_enabled_mutex::scoped_lock lock(desc->mutex);
HITCBC 234   9758 desc->registered_events = reactor_event_read | reactor_event_write; 240   9760 desc->registered_events = reactor_event_read | reactor_event_write;
HITCBC 235   9758 desc->unpollable = false; // the state is reused across adoptions 241   9760 desc->unpollable = false; // the state is reused across adoptions
HITCBC 236   9758 desc->fd = fd; 242   9760 desc->fd = fd;
HITCBC 237   9758 desc->scheduler_ = this; 243   9760 desc->scheduler_ = this;
HITCBC 238   9758 desc->ready_events_.store(0, std::memory_order_release); 244   9760 desc->ready_events_.store(0, std::memory_order_release);
239   // object_ref_ is not touched: a queued invocation from an 245   // object_ref_ is not touched: a queued invocation from an
240   // earlier registration still owns the reference in it. 246   // earlier registration still owns the reference in it.
HITCBC 241   9758 desc->read_ready = false; 247   9760 desc->read_ready = false;
HITCBC 242   9758 desc->write_ready = false; 248   9760 desc->write_ready = false;
HITCBC 243   9758 } 249   9760 }
244   250  
245   { 251   {
HITCBC 246   9758 mutex_type::scoped_lock lock(mutex_); 252   9760 mutex_type::scoped_lock lock(mutex_);
247   try 253   try
248   { 254   {
HITCBC 249   9758 registered_descs_[fd] = desc; 255   9760 registered_descs_[fd] = desc;
250   } 256   }
HITCBC 251   1 catch (std::bad_alloc const&) 257   1 catch (std::bad_alloc const&)
252   { 258   {
HITCBC 253   1 return make_err(ENOMEM); 259   1 return make_err(ENOMEM);
HITCBC 254   1 } 260   1 }
HITCBC 255   9757 if (fd > max_fd_) 261   9759 if (fd > max_fd_)
HITCBC 256   9318 max_fd_ = fd; 262   9301 max_fd_ = fd;
HITCBC 257   9758 } 263   9760 }
258   264  
HITCBC 259   9757 interrupt_reactor(); 265   9759 interrupt_reactor();
HITCBC 260   9757 return {}; 266   9759 return {};
261   } 267   }
262   268  
263   inline void 269   inline void
HITCBC 264   9485 select_scheduler::deregister_descriptor(int fd) const 270   9488 select_scheduler::deregister_descriptor(int fd) const
265   { 271   {
HITCBC 266   9485 mutex_type::scoped_lock lock(mutex_); 272   9488 mutex_type::scoped_lock lock(mutex_);
267   273  
HITCBC 268   9485 auto it = registered_descs_.find(fd); 274   9488 auto it = registered_descs_.find(fd);
HITCBC 269   9485 if (it == registered_descs_.end()) 275   9488 if (it == registered_descs_.end())
MISUBC 270   ✗ return; 276   ✗ return;
271   277  
HITCBC 272   9485 registered_descs_.erase(it); 278   9488 registered_descs_.erase(it);
273   279  
HITCBC 274   9485 if (fd == max_fd_) 280   9488 if (fd == max_fd_)
275   { 281   {
HITCBC 276   7627 max_fd_ = pipe_fds_[0]; 282   7627 max_fd_ = pipe_fds_[0];
HITCBC 277   12623 for (auto& [registered_fd, state] : registered_descs_) 283   12635 for (auto& [registered_fd, state] : registered_descs_)
278   { 284   {
HITCBC 279   4996 if (registered_fd > max_fd_) 285   5008 if (registered_fd > max_fd_)
HITCBC 280   4807 max_fd_ = registered_fd; 286   4813 max_fd_ = registered_fd;
281   } 287   }
282   } 288   }
HITCBC 283   9485 } 289   9488 }
284   290  
285   inline void 291   inline void
HITCBC 286   6724 select_scheduler::notify_reactor() const 292   6738 select_scheduler::notify_reactor() const
287   { 293   {
HITCBC 288   6724 interrupt_reactor(); 294   6738 interrupt_reactor();
HITCBC 289   6724 } 295   6738 }
290   296  
291   inline void 297   inline void
HITCBC 292   25173 select_scheduler::interrupt_reactor() const 298   24970 select_scheduler::interrupt_reactor() const
293   { 299   {
HITCBC 294   25173 char byte = 1; 300   24970 char byte = 1;
HITCBC 295   25173 [[maybe_unused]] auto r = ::write(pipe_fds_[1], &byte, 1); 301   24970 [[maybe_unused]] auto r = ::write(pipe_fds_[1], &byte, 1);
HITCBC 296   25173 } 302   24970 }
297   303  
298   inline long 304   inline long
HITCBC 299   13450 select_scheduler::calculate_timeout(long requested_timeout_us) const 305   18014 select_scheduler::calculate_timeout(long requested_timeout_us) const
300   { 306   {
HITCBC 301   13450 if (requested_timeout_us == 0) 307   18014 if (requested_timeout_us == 0)
302   return 0; // LCOV_EXCL_LINE run_task passes 0 via task_interrupted_, never through this argument 308   return 0; // LCOV_EXCL_LINE run_task passes 0 via task_interrupted_, never through this argument
303   309  
HITCBC 304 - 13450 auto* tsvc = timers(); 310 + 18014 auto nearest = timer_svc_->nearest_expiry();
DCB 305 - 13450 if (!tsvc)  
DCB 306 - 6857 return requested_timeout_us;  
DCB 307 - 6593 auto nearest = tsvc->nearest_expiry();  
HITCBC 308   6593 if (nearest == timer_service::time_point::max()) 311   18014 if (nearest == timer_service::time_point::max())
HITCBC 309   139 return requested_timeout_us; 312   11555 return requested_timeout_us;
310   313  
HITCBC 311   6454 auto now = std::chrono::steady_clock::now(); 314   6459 auto now = std::chrono::steady_clock::now();
HITCBC 312   6454 if (nearest <= now) 315   6459 if (nearest <= now)
HITCBC 313   87 return 0; 316   107 return 0;
314   317  
315   auto timer_timeout_us = 318   auto timer_timeout_us =
HITCBC 316   6367 std::chrono::duration_cast<std::chrono::microseconds>(nearest - now) 319   6352 std::chrono::duration_cast<std::chrono::microseconds>(nearest - now)
HITCBC 317   6367 .count(); 320   6352 .count();
318   321  
HITCBC 319   6367 constexpr auto long_max = 322   6352 constexpr auto long_max =
320   static_cast<long long>((std::numeric_limits<long>::max)()); 323   static_cast<long long>((std::numeric_limits<long>::max)());
321   auto capped_timer_us = 324   auto capped_timer_us =
HITCBC 322   6367 (std::min)((std::max)(static_cast<long long>(timer_timeout_us), 325   6352 (std::min)((std::max)(static_cast<long long>(timer_timeout_us),
HITCBC 323   6367 static_cast<long long>(0)), 326   6352 static_cast<long long>(0)),
HITCBC 324   6367 long_max); 327   6352 long_max);
325   328  
HITCBC 326   6367 if (requested_timeout_us < 0) 329   6352 if (requested_timeout_us < 0)
HITCBC 327   6361 return static_cast<long>(capped_timer_us); 330   6346 return static_cast<long>(capped_timer_us);
328   331  
329   return static_cast<long>( 332   return static_cast<long>(
HITCBC 330   6 (std::min)(static_cast<long long>(requested_timeout_us), 333   6 (std::min)(static_cast<long long>(requested_timeout_us),
HITCBC 331   6 capped_timer_us)); 334   6 capped_timer_us));
332   } 335   }
333   336  
334   inline void 337   inline void
HITCBC 335   140346 select_scheduler::run_task(lock_type& lock, context_type& ctx, long timeout_us) 338   145115 select_scheduler::run_task(lock_type& lock, context_type& ctx, long timeout_us)
336   { 339   {
337   long effective_timeout_us = 340   long effective_timeout_us =
HITCBC 338   140346 task_interrupted_ ? 0 : calculate_timeout(timeout_us); 341   145115 task_interrupted_ ? 0 : calculate_timeout(timeout_us);
339   342  
340   // Snapshot registered descriptors while holding lock. 343   // Snapshot registered descriptors while holding lock.
341   // Record which directions each fd needs monitored to avoid a hot 344   // Record which directions each fd needs monitored to avoid a hot
342   // loop: select is level-triggered, so a writable socket (nearly 345   // loop: select is level-triggered, so a writable socket (nearly
343   // always writable) or an always-readable fd (/dev/zero, a pipe at 346   // always writable) or an always-readable fd (/dev/zero, a pipe at
344   // EOF) would return select() immediately every iteration if 347   // EOF) would return select() immediately every iteration if
345   // unconditionally added. Membership in both sets is opt-in: a 348   // unconditionally added. Membership in both sets is opt-in: a
346   // parked op or wait in a direction opts that direction in. The 349   // 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: 350   // exceptional set is opt-in too, for any parked op or wait:
348   // Darwin reports a character device (/dev/zero, /dev/null) as 351   // Darwin reports a character device (/dev/zero, /dev/null) as
349   // exceptional on every call. 352   // exceptional on every call.
350   struct fd_entry 353   struct fd_entry
351   { 354   {
352   int fd; 355   int fd;
353   reactor_descriptor_state* desc; 356   reactor_descriptor_state* desc;
354   std::uint32_t want; 357   std::uint32_t want;
355   }; 358   };
356   fd_entry snapshot[FD_SETSIZE]; 359   fd_entry snapshot[FD_SETSIZE];
HITCBC 357   140346 int snapshot_count = 0; 360   145115 int snapshot_count = 0;
358   361  
359   // The caller may have released the lock to hand queued work to a 362   // The caller may have released the lock to hand queued work to a
360   // peer, and deregister_descriptor() erases from this map under it. 363   // peer, and deregister_descriptor() erases from this map under it.
HITCBC 361   140346 if (!lock.owns_lock()) 364   145115 if (!lock.owns_lock())
HITCBC 362   126893 lock.lock(); 365   127098 lock.lock();
363   366  
HITCBC 364   393416 for (auto& [fd, desc] : registered_descs_) 367   399363 for (auto& [fd, desc] : registered_descs_)
365   { 368   {
HITCBC 366   253070 if (snapshot_count < FD_SETSIZE) 369   254248 if (snapshot_count < FD_SETSIZE)
367   { 370   {
HITCBC 368   253070 conditionally_enabled_mutex::scoped_lock desc_lock(desc->mutex); 371   254248 conditionally_enabled_mutex::scoped_lock desc_lock(desc->mutex);
HITCBC 369   253070 snapshot[snapshot_count].fd = fd; 372   254248 snapshot[snapshot_count].fd = fd;
HITCBC 370   253070 snapshot[snapshot_count].desc = desc; 373   254248 snapshot[snapshot_count].desc = desc;
HITCBC 371   253070 snapshot[snapshot_count].want = 374   254248 snapshot[snapshot_count].want =
HITCBC 372   253070 ((desc->read_op || desc->wait_read_op) ? reactor_event_read 375   254248 ((desc->read_op || desc->wait_read_op) ? reactor_event_read
HITCBC 373   253070 : 0) | 376   254248 : 0) |
HITCBC 374   252696 ((desc->write_op || desc->connect_op || desc->wait_write_op) 377   253868 ((desc->write_op || desc->connect_op || desc->wait_write_op)
HITCBC 375   505766 ? reactor_event_write 378   508116 ? reactor_event_write
HITCBC 376   253070 : 0) | 379   254248 : 0) |
HITCBC 377   253070 (desc->wait_error_op ? reactor_event_error : 0); 380   254248 (desc->wait_error_op ? reactor_event_error : 0);
HITCBC 378   253070 ++snapshot_count; 381   254248 ++snapshot_count;
HITCBC 379   253070 } 382   254248 }
380   } 383   }
381   384  
HITCBC 382   140346 if (lock.owns_lock()) 385   145115 if (lock.owns_lock())
HITCBC 383   140345 lock.unlock(); 386   145114 lock.unlock();
384   387  
HITCBC 385   140346 task_cleanup on_exit{this, &lock, ctx}; 388   145115 task_cleanup on_exit{this, &lock, ctx};
386   389  
387   fd_set read_fds, write_fds, except_fds; 390   fd_set read_fds, write_fds, except_fds;
HITCBC 388   2385882 FD_ZERO(&read_fds); 391   2466955 FD_ZERO(&read_fds);
HITCBC 389   2385882 FD_ZERO(&write_fds); 392   2466955 FD_ZERO(&write_fds);
HITCBC 390   2385882 FD_ZERO(&except_fds); 393   2466955 FD_ZERO(&except_fds);
391   394  
HITCBC 392   140346 FD_SET(pipe_fds_[0], &read_fds); 395   145115 FD_SET(pipe_fds_[0], &read_fds);
HITCBC 393   140346 int nfds = pipe_fds_[0]; 396   145115 int nfds = pipe_fds_[0];
394   397  
HITCBC 395   393416 for (int i = 0; i < snapshot_count; ++i) 398   399363 for (int i = 0; i < snapshot_count; ++i)
396   { 399   {
HITCBC 397   253070 int fd = snapshot[i].fd; 400   254248 int fd = snapshot[i].fd;
HITCBC 398   253070 if (snapshot[i].want & reactor_event_read) 401   254248 if (snapshot[i].want & reactor_event_read)
HITCBC 399   112460 FD_SET(fd, &read_fds); 402   119420 FD_SET(fd, &read_fds);
HITCBC 400   253070 if (snapshot[i].want & reactor_event_write) 403   254248 if (snapshot[i].want & reactor_event_write)
HITCBC 401   2988 FD_SET(fd, &write_fds); 404   3001 FD_SET(fd, &write_fds);
HITCBC 402   253070 if (snapshot[i].want != 0) 405   254248 if (snapshot[i].want != 0)
HITCBC 403   115643 FD_SET(fd, &except_fds); 406   122711 FD_SET(fd, &except_fds);
HITCBC 404   253070 if (fd > nfds) 407   254248 if (fd > nfds)
HITCBC 405   138307 nfds = fd; 408   142463 nfds = fd;
406   } 409   }
407   410  
408   struct timeval tv; 411   struct timeval tv;
HITCBC 409   140346 struct timeval* tv_ptr = nullptr; 412   145115 struct timeval* tv_ptr = nullptr;
HITCBC 410   140346 if (effective_timeout_us >= 0) 413   145115 if (effective_timeout_us >= 0)
411   { 414   {
HITCBC 412   134019 tv.tv_sec = effective_timeout_us / 1000000; 415   134160 tv.tv_sec = effective_timeout_us / 1000000;
HITCBC 413   134019 tv.tv_usec = effective_timeout_us % 1000000; 416   134160 tv.tv_usec = effective_timeout_us % 1000000;
HITCBC 414   134019 tv_ptr = &tv; 417   134160 tv_ptr = &tv;
415   } 418   }
416   419  
HITCBC 417   140346 int ready = ::select(nfds + 1, &read_fds, &write_fds, &except_fds, tv_ptr); 420   145115 int ready = ::select(nfds + 1, &read_fds, &write_fds, &except_fds, tv_ptr);
418   421  
419   // EINTR: signal interrupted select(), just retry. 422   // EINTR: signal interrupted select(), just retry.
420   // EBADF: an fd was closed between snapshot and select(); retry 423   // EBADF: an fd was closed between snapshot and select(); retry
421   // with a fresh snapshot from registered_descs_. 424   // with a fresh snapshot from registered_descs_.
422   // Both fall through with no ready descriptors rather than 425   // Both fall through with no ready descriptors rather than
423   // returning: the caller handed this function an owned lock that 426   // returning: the caller handed this function an owned lock that
424   // only the epilogue below re-acquires. 427   // only the epilogue below re-acquires.
HITCBC 425   140346 if (ready < 0) 428   145115 if (ready < 0)
426   { 429   {
HITCBC 427   3 if (errno != EINTR && errno != EBADF) 430   3 if (errno != EINTR && errno != EBADF)
HITCBC 428   1 detail::throw_system_error(make_err(errno), "select"); 431   1 detail::throw_system_error(make_err(errno), "select");
HITCBC 429   2 ready = 0; 432   2 ready = 0;
430   } 433   }
431   434  
432   // Process timers outside the lock 435   // Process timers outside the lock
HITCBC 433 - 140345 if (auto* tsvc = timers()) 436 + 145114 timer_svc_->process_expired();
DCB 434 - 28740 tsvc->process_expired();  
435   437  
HITCBC 436   140345 ready_queue local_ops; 438   145114 ready_queue local_ops;
437   439  
HITCBC 438   140345 if (ready > 0) 440   145114 if (ready > 0)
439   { 441   {
HITCBC 440   13663 if (FD_ISSET(pipe_fds_[0], &read_fds)) 442   18240 if (FD_ISSET(pipe_fds_[0], &read_fds))
441   { 443   {
442   char buf[256]; 444   char buf[256];
HITCBC 443   22365 while (::read(pipe_fds_[0], buf, sizeof(buf)) > 0) 445   22070 while (::read(pipe_fds_[0], buf, sizeof(buf)) > 0)
444   { 446   {
445   } 447   }
446   } 448   }
447   449  
HITCBC 448   96928 for (int i = 0; i < snapshot_count; ++i) 450   104465 for (int i = 0; i < snapshot_count; ++i)
449   { 451   {
HITCBC 450   83265 int fd = snapshot[i].fd; 452   86225 int fd = snapshot[i].fd;
HITCBC 451   83265 reactor_descriptor_state* desc = snapshot[i].desc; 453   86225 reactor_descriptor_state* desc = snapshot[i].desc;
452   454  
HITCBC 453   83265 std::uint32_t flags = 0; 455   86225 std::uint32_t flags = 0;
HITCBC 454   83265 if (FD_ISSET(fd, &read_fds)) 456   86225 if (FD_ISSET(fd, &read_fds))
HITCBC 455   5200 flags |= reactor_event_read; 457   9923 flags |= reactor_event_read;
HITCBC 456   83265 if (FD_ISSET(fd, &write_fds)) 458   86225 if (FD_ISSET(fd, &write_fds))
HITCBC 457   2704 flags |= reactor_event_write; 459   2707 flags |= reactor_event_write;
HITCBC 458   83265 if (FD_ISSET(fd, &except_fds)) 460   86225 if (FD_ISSET(fd, &except_fds))
HITCBC 459   10 flags |= reactor_event_error; 461   16 flags |= reactor_event_error;
460   462  
HITCBC 461   83265 if (flags == 0) 463   86225 if (flags == 0)
HITCBC 462   75359 continue; 464   73593 continue;
463   465  
HITCBC 464   7906 desc->add_ready_events(flags); 466   12632 desc->add_ready_events(flags);
465   467  
HITCBC 466   7906 bool expected = false; 468   12632 bool expected = false;
HITCBC 467   7906 if (desc->is_enqueued_.compare_exchange_strong( 469   12632 if (desc->is_enqueued_.compare_exchange_strong(
468   expected, true, std::memory_order_release, 470   expected, true, std::memory_order_release,
469   std::memory_order_relaxed)) 471   std::memory_order_relaxed))
470   { 472   {
HITCBC 471   6178 local_ops.push(desc); 473   6190 local_ops.push(desc);
472   } 474   }
473   } 475   }
474   } 476   }
475   477  
HITCBC 476   140345 lock.lock(); 478   145114 lock.lock();
477   479  
HITCBC 478   140345 completed_ops_.splice(local_ops); 480   145114 completed_ops_.splice(local_ops);
HITCBC 479   140346 } 481   145115 }
480   482  
481   } // namespace boost::corosio::detail 483   } // namespace boost::corosio::detail
482   484  
483   #endif // BOOST_COROSIO_HAS_SELECT 485   #endif // BOOST_COROSIO_HAS_SELECT
484   486  
485   #endif // BOOST_COROSIO_NATIVE_DETAIL_SELECT_SELECT_SCHEDULER_HPP 487   #endif // BOOST_COROSIO_NATIVE_DETAIL_SELECT_SELECT_SCHEDULER_HPP