. /include/boost/corosio /native /detail /select /select_scheduler.hpp
99.41% Lines (168/169) 100.00% Functions (11/11)
[ ] 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   #include <boost/corosio/native/detail/posix/posix_resolver_service.hpp> 27   #include <boost/corosio/native/detail/posix/posix_resolver_service.hpp>
28   #include <boost/corosio/native/detail/posix/posix_signal_service.hpp> 28   #include <boost/corosio/native/detail/posix/posix_signal_service.hpp>
29   #include <boost/corosio/native/detail/posix/posix_stream_file_service.hpp> 29   #include <boost/corosio/native/detail/posix/posix_stream_file_service.hpp>
30   #include <boost/corosio/native/detail/posix/posix_random_access_file_service.hpp> 30   #include <boost/corosio/native/detail/posix/posix_random_access_file_service.hpp>
31   31  
32   #include <boost/corosio/detail/except.hpp> 32   #include <boost/corosio/detail/except.hpp>
33   33  
34   #include <sys/select.h> 34   #include <sys/select.h>
35   #include <unistd.h> 35   #include <unistd.h>
36   #include <errno.h> 36   #include <errno.h>
37   #include <fcntl.h> 37   #include <fcntl.h>
38   38  
39   #include <atomic> 39   #include <atomic>
40   #include <chrono> 40   #include <chrono>
41   #include <cstdint> 41   #include <cstdint>
42   #include <limits> 42   #include <limits>
43   #include <mutex> 43   #include <mutex>
44   #include <new> 44   #include <new>
45   #include <unordered_map> 45   #include <unordered_map>
46   46  
47   namespace boost::corosio::detail { 47   namespace boost::corosio::detail {
48   48  
49   struct select_op; 49   struct select_op;
50   50  
51   /** POSIX scheduler using select() for I/O multiplexing. 51   /** POSIX scheduler using select() for I/O multiplexing.
52   52  
53   This scheduler implements the scheduler interface using the POSIX select() 53   This scheduler implements the scheduler interface using the POSIX select()
54   call for I/O event notification. It inherits the shared reactor threading 54   call for I/O event notification. It inherits the shared reactor threading
55   model from reactor_scheduler: signal state machine, inline completion 55   model from reactor_scheduler: signal state machine, inline completion
56   budget, work counting, and the do_one event loop. 56   budget, work counting, and the do_one event loop.
57   57  
58   The design mirrors epoll_scheduler for behavioral consistency: 58   The design mirrors epoll_scheduler for behavioral consistency:
59   - Same single-reactor thread coordination model 59   - Same single-reactor thread coordination model
60   - Same deferred I/O pattern (reactor marks ready; workers do I/O) 60   - Same deferred I/O pattern (reactor marks ready; workers do I/O)
61   - Same timer integration pattern 61   - Same timer integration pattern
62   62  
63   Known Limitations: 63   Known Limitations:
64   - FD_SETSIZE (~1024) limits maximum concurrent connections 64   - FD_SETSIZE (~1024) limits maximum concurrent connections
65   - O(n) scanning: rebuilds fd_sets each iteration 65   - O(n) scanning: rebuilds fd_sets each iteration
66   - Level-triggered only (no edge-triggered mode) 66   - Level-triggered only (no edge-triggered mode)
67   67  
68   @par Thread Safety 68   @par Thread Safety
69   All public member functions are thread-safe. 69   All public member functions are thread-safe.
70   */ 70   */
71   class BOOST_COROSIO_DECL select_scheduler final : public reactor_scheduler 71   class BOOST_COROSIO_DECL select_scheduler final : public reactor_scheduler
72   { 72   {
73   public: 73   public:
74   /** Construct the scheduler. 74   /** Construct the scheduler.
75   75  
76   Creates a self-pipe for reactor interruption. 76   Creates a self-pipe for reactor interruption.
77   77  
78   @param ctx Reference to the owning execution_context. 78   @param ctx Reference to the owning execution_context.
79   @param concurrency_hint Hint for expected thread count (unused). 79   @param concurrency_hint Hint for expected thread count (unused).
80   */ 80   */
81   select_scheduler(capy::execution_context& ctx, int concurrency_hint = -1); 81   select_scheduler(capy::execution_context& ctx, int concurrency_hint = -1);
82   82  
83   /// Destroy the scheduler. 83   /// Destroy the scheduler.
84   ~select_scheduler() override; 84   ~select_scheduler() override;
85   85  
86   select_scheduler(select_scheduler const&) = delete; 86   select_scheduler(select_scheduler const&) = delete;
87   select_scheduler& operator=(select_scheduler const&) = delete; 87   select_scheduler& operator=(select_scheduler const&) = delete;
88   88  
89   /// Shut down the scheduler, draining pending operations. 89   /// Shut down the scheduler, draining pending operations.
90   void shutdown() override; 90   void shutdown() override;
91   91  
92   /** Return the maximum file descriptor value supported. 92   /** Return the maximum file descriptor value supported.
93   93  
94   Returns FD_SETSIZE - 1, the maximum fd value that can be 94   Returns FD_SETSIZE - 1, the maximum fd value that can be
95   monitored by select(). Operations with fd >= FD_SETSIZE 95   monitored by select(). Operations with fd >= FD_SETSIZE
96   will fail with EINVAL. 96   will fail with EINVAL.
97   97  
98   @return The maximum supported file descriptor value. 98   @return The maximum supported file descriptor value.
99   */ 99   */
100   static constexpr int max_fd() noexcept 100   static constexpr int max_fd() noexcept
101   { 101   {
102   return FD_SETSIZE - 1; 102   return FD_SETSIZE - 1;
103   } 103   }
104   104  
105   /** Register a descriptor for persistent monitoring. 105   /** Register a descriptor for persistent monitoring.
106   106  
107   The fd is added to the registered_descs_ map and will be 107   The fd is added to the registered_descs_ map and will be
108   included in subsequent select() calls. The reactor is 108   included in subsequent select() calls. The reactor is
109   interrupted so a blocked select() rebuilds its fd_sets. 109   interrupted so a blocked select() rebuilds its fd_sets.
110   110  
111   @param fd The file descriptor to register. 111   @param fd The file descriptor to register.
112   @param desc Pointer to descriptor state for this fd. 112   @param desc Pointer to descriptor state for this fd.
113   113  
114   @return The error if the fd cannot be tracked, otherwise a 114   @return The error if the fd cannot be tracked, otherwise a
115   default constructed error code. 115   default constructed error code.
116   */ 116   */
117   std::error_code 117   std::error_code
118   register_descriptor(int fd, reactor_descriptor_state* desc) const; 118   register_descriptor(int fd, reactor_descriptor_state* desc) const;
119   119  
120   /** Deregister a persistently registered descriptor. 120   /** Deregister a persistently registered descriptor.
121   121  
122   @param fd The file descriptor to deregister. 122   @param fd The file descriptor to deregister.
123   */ 123   */
124   void deregister_descriptor(int fd) const; 124   void deregister_descriptor(int fd) const;
125   125  
126   /** Interrupt the reactor so it rebuilds its fd_sets. 126   /** Interrupt the reactor so it rebuilds its fd_sets.
127   127  
128   Called when a write, connect, or write-wait op is registered 128   Called when a write, connect, or write-wait op is registered
129   after the reactor's snapshot was taken. Without this, 129   after the reactor's snapshot was taken. Without this,
130   select() may block not watching for writability on the fd. 130   select() may block not watching for writability on the fd.
131   */ 131   */
132   void notify_reactor() const; 132   void notify_reactor() const;
133   133  
134   /// Watch the read end of the POSIX signal self-pipe (see scheduler.hpp). 134   /// Watch the read end of the POSIX signal self-pipe (see scheduler.hpp).
ECB 135 - 55 [[nodiscard]] std::error_code register_signal_reader(int read_fd) override 135 + [[nodiscard]] std::error_code
HITGNC   136 + 55 register_signal_reader(int read_fd) override
136   { 137   {
HITCBC 137   55 return register_descriptor(read_fd, signal_pipe_reader_.arm()); 138   55 return register_descriptor(read_fd, signal_pipe_reader_.arm());
138   } 139   }
139   140  
140   private: 141   private:
141 - void run_task(lock_type& lock, context_type& ctx, long timeout_us) override; 142 + void
  143 + run_task(lock_type& lock, context_type& ctx,
  144 + long timeout_us) override;
142   void interrupt_reactor() const override; 145   void interrupt_reactor() const override;
143   long calculate_timeout(long requested_timeout_us) const; 146   long calculate_timeout(long requested_timeout_us) const;
144   147  
145   // Watches the global signal self-pipe's read end (armed lazily by 148   // Watches the global signal self-pipe's read end (armed lazily by
146   // register_signal_reader on the first signal registration). 149   // register_signal_reader on the first signal registration).
147   reactor_signal_pipe_reader signal_pipe_reader_; 150   reactor_signal_pipe_reader signal_pipe_reader_;
148   151  
149   // Self-pipe for interrupting select() 152   // Self-pipe for interrupting select()
150   int pipe_fds_[2]; // [0]=read, [1]=write 153   int pipe_fds_[2]; // [0]=read, [1]=write
151   154  
152   // Per-fd tracking for fd_set building 155   // Per-fd tracking for fd_set building
153 - mutable std::unordered_map<int, reactor_descriptor_state*> 156 + mutable std::unordered_map<int, reactor_descriptor_state*> registered_descs_;
154 - registered_descs_;  
155   mutable int max_fd_ = -1; 157   mutable int max_fd_ = -1;
156   }; 158   };
157   159  
HITCBC 158   886 inline select_scheduler::select_scheduler(capy::execution_context& ctx, int) 160   886 inline select_scheduler::select_scheduler(capy::execution_context& ctx, int)
HITCBC 159   886 : pipe_fds_{-1, -1} 161   886 : pipe_fds_{-1, -1}
HITCBC 160   886 , max_fd_(-1) 162   886 , max_fd_(-1)
161   { 163   {
HITCBC 162   886 if (::pipe(pipe_fds_) < 0) 164   886 if (::pipe(pipe_fds_) < 0)
HITCBC 163   1 detail::throw_system_error(make_err(errno), "pipe"); 165   1 detail::throw_system_error(make_err(errno), "pipe");
164   166  
HITCBC 165   2646 for (int i = 0; i < 2; ++i) 167   2646 for (int i = 0; i < 2; ++i)
166   { 168   {
HITCBC 167   1767 int flags = ::fcntl(pipe_fds_[i], F_GETFL, 0); 169   1767 int flags = ::fcntl(pipe_fds_[i], F_GETFL, 0);
HITCBC 168   1767 if (flags == -1) 170   1767 if (flags == -1)
169   { 171   {
HITCBC 170   2 int errn = errno; 172   2 int errn = errno;
HITCBC 171   2 ::close(pipe_fds_[0]); 173   2 ::close(pipe_fds_[0]);
HITCBC 172   2 ::close(pipe_fds_[1]); 174   2 ::close(pipe_fds_[1]);
HITCBC 173   2 detail::throw_system_error(make_err(errn), "fcntl F_GETFL"); 175   2 detail::throw_system_error(make_err(errn), "fcntl F_GETFL");
174   } 176   }
HITCBC 175   1765 if (::fcntl(pipe_fds_[i], F_SETFL, flags | O_NONBLOCK) == -1) 177   1765 if (::fcntl(pipe_fds_[i], F_SETFL, flags | O_NONBLOCK) == -1)
176   { 178   {
HITCBC 177   2 int errn = errno; 179   2 int errn = errno;
HITCBC 178   2 ::close(pipe_fds_[0]); 180   2 ::close(pipe_fds_[0]);
HITCBC 179   2 ::close(pipe_fds_[1]); 181   2 ::close(pipe_fds_[1]);
HITCBC 180   2 detail::throw_system_error(make_err(errn), "fcntl F_SETFL"); 182   2 detail::throw_system_error(make_err(errn), "fcntl F_SETFL");
181   } 183   }
HITCBC 182   1763 if (::fcntl(pipe_fds_[i], F_SETFD, FD_CLOEXEC) == -1) 184   1763 if (::fcntl(pipe_fds_[i], F_SETFD, FD_CLOEXEC) == -1)
183   { 185   {
HITCBC 184   2 int errn = errno; 186   2 int errn = errno;
HITCBC 185   2 ::close(pipe_fds_[0]); 187   2 ::close(pipe_fds_[0]);
HITCBC 186   2 ::close(pipe_fds_[1]); 188   2 ::close(pipe_fds_[1]);
HITCBC 187   2 detail::throw_system_error(make_err(errn), "fcntl F_SETFD"); 189   2 detail::throw_system_error(make_err(errn), "fcntl F_SETFD");
188   } 190   }
189   } 191   }
190   192  
HITCBC 191   879 timer_svc_ = &get_timer_service(ctx, *this); 193   879 timer_svc_ = &get_timer_service(ctx, *this);
HITCBC 192   879 timer_svc_->set_on_earliest_changed( 194   879 timer_svc_->set_on_earliest_changed(
HITCBC 193   3674 timer_service::callback(this, [](void* p) { 195   3572 timer_service::callback(this, [](void* p) {
HITCBC 194   2795 static_cast<select_scheduler*>(p)->interrupt_reactor(); 196   2693 static_cast<select_scheduler*>(p)->interrupt_reactor();
HITCBC 195   2795 })); 197   2693 }));
196   198  
HITCBC 197   879 get_resolver_service(ctx, *this); 199   879 get_resolver_service(ctx, *this);
HITCBC 198   879 get_signal_service(ctx, *this); 200   879 get_signal_service(ctx, *this);
HITCBC 199   879 get_stream_file_service(ctx, *this); 201   879 get_stream_file_service(ctx, *this);
HITCBC 200   879 get_random_access_file_service(ctx, *this); 202   879 get_random_access_file_service(ctx, *this);
201   203  
HITCBC 202   879 completed_ops_.push(&task_op_); 204   879 completed_ops_.push(&task_op_);
HITCBC 203   900 } 205   900 }
204   206  
HITCBC 205   1758 inline select_scheduler::~select_scheduler() 207   1758 inline select_scheduler::~select_scheduler()
206   { 208   {
HITCBC 207   879 if (pipe_fds_[0] >= 0) 209   879 if (pipe_fds_[0] >= 0)
HITCBC 208   879 ::close(pipe_fds_[0]); 210   879 ::close(pipe_fds_[0]);
HITCBC 209   879 if (pipe_fds_[1] >= 0) 211   879 if (pipe_fds_[1] >= 0)
HITCBC 210   879 ::close(pipe_fds_[1]); 212   879 ::close(pipe_fds_[1]);
HITCBC 211   1758 } 213   1758 }
212   214  
213   inline void 215   inline void
HITCBC 214   879 select_scheduler::shutdown() 216   879 select_scheduler::shutdown()
215   { 217   {
HITCBC 216   879 shutdown_drain(); 218   879 shutdown_drain();
217   219  
HITCBC 218   879 if (pipe_fds_[1] >= 0) 220   879 if (pipe_fds_[1] >= 0)
HITCBC 219   879 interrupt_reactor(); 221   879 interrupt_reactor();
HITCBC 220   879 } 222   879 }
221   223  
222   inline std::error_code 224   inline std::error_code
HITCBC 223   4823 select_scheduler::register_descriptor( 225   4646 select_scheduler::register_descriptor(
224   int fd, reactor_descriptor_state* desc) const 226   int fd, reactor_descriptor_state* desc) const
225   { 227   {
HITCBC 226   4823 if (fd < 0 || fd >= FD_SETSIZE) 228   4646 if (fd < 0 || fd >= FD_SETSIZE)
HITCBC 227   1 return make_err(EMFILE); 229   1 return make_err(EMFILE);
228   230  
HITCBC 229   4822 desc->registered_events = reactor_event_read | reactor_event_write; 231   4645 desc->registered_events = reactor_event_read | reactor_event_write;
HITCBC 230   4822 desc->fd = fd; 232   4645 desc->fd = fd;
HITCBC 231   4822 desc->scheduler_ = this; 233   4645 desc->scheduler_ = this;
HITCBC 232   4822 desc->mutex.set_enabled(reactor_io_locking_); 234   4645 desc->mutex.set_enabled(reactor_io_locking_);
HITCBC 233   4822 desc->ready_events_.store(0, std::memory_order_relaxed); 235   4645 desc->ready_events_.store(0, std::memory_order_relaxed);
234   236  
235   { 237   {
HITCBC 236   4822 conditionally_enabled_mutex::scoped_lock lock(desc->mutex); 238   4645 conditionally_enabled_mutex::scoped_lock lock(desc->mutex);
HITCBC 237   4822 desc->impl_ref_.reset(); 239   4645 desc->impl_ref_.reset();
HITCBC 238   4822 desc->read_ready = false; 240   4645 desc->read_ready = false;
HITCBC 239   4822 desc->write_ready = false; 241   4645 desc->write_ready = false;
HITCBC 240   4822 } 242   4645 }
241   243  
242   { 244   {
HITCBC 243   4822 mutex_type::scoped_lock lock(mutex_); 245   4645 mutex_type::scoped_lock lock(mutex_);
244   try 246   try
245   { 247   {
HITCBC 246   4822 registered_descs_[fd] = desc; 248   4645 registered_descs_[fd] = desc;
247   } 249   }
HITCBC 248   1 catch (std::bad_alloc const&) 250   1 catch (std::bad_alloc const&)
249   { 251   {
HITCBC 250   1 return make_err(ENOMEM); 252   1 return make_err(ENOMEM);
HITCBC 251   1 } 253   1 }
HITCBC 252   4821 if (fd > max_fd_) 254   4644 if (fd > max_fd_)
HITCBC 253   4767 max_fd_ = fd; 255   4590 max_fd_ = fd;
HITCBC 254   4822 } 256   4645 }
255   257  
HITCBC 256   4821 interrupt_reactor(); 258   4644 interrupt_reactor();
HITCBC 257   4821 return {}; 259   4644 return {};
258   } 260   }
259   261  
260   inline void 262   inline void
HITCBC 261   4767 select_scheduler::deregister_descriptor(int fd) const 263   4590 select_scheduler::deregister_descriptor(int fd) const
262   { 264   {
HITCBC 263   4767 mutex_type::scoped_lock lock(mutex_); 265   4590 mutex_type::scoped_lock lock(mutex_);
264   266  
HITCBC 265   4767 auto it = registered_descs_.find(fd); 267   4590 auto it = registered_descs_.find(fd);
HITCBC 266   4767 if (it == registered_descs_.end()) 268   4590 if (it == registered_descs_.end())
MISUBC 267   ✗ return; 269   ✗ return;
268   270  
HITCBC 269   4767 registered_descs_.erase(it); 271   4590 registered_descs_.erase(it);
270   272  
HITCBC 271   4767 if (fd == max_fd_) 273   4590 if (fd == max_fd_)
272   { 274   {
HITCBC 273   4431 max_fd_ = pipe_fds_[0]; 275   4253 max_fd_ = pipe_fds_[0];
HITCBC 274   8460 for (auto& [registered_fd, state] : registered_descs_) 276   8104 for (auto& [registered_fd, state] : registered_descs_)
275   { 277   {
HITCBC 276   4029 if (registered_fd > max_fd_) 278   3851 if (registered_fd > max_fd_)
HITCBC 277   3936 max_fd_ = registered_fd; 279   3758 max_fd_ = registered_fd;
278   } 280   }
279   } 281   }
HITCBC 280   4767 } 282   4590 }
281   283  
282   inline void 284   inline void
HITCBC 283   2143 select_scheduler::notify_reactor() const 285   2055 select_scheduler::notify_reactor() const
284   { 286   {
HITCBC 285   2143 interrupt_reactor(); 287   2055 interrupt_reactor();
HITCBC 286   2143 } 288   2055 }
287   289  
288   inline void 290   inline void
HITCBC 289   11445 select_scheduler::interrupt_reactor() const 291   11080 select_scheduler::interrupt_reactor() const
290   { 292   {
HITCBC 291   11445 char byte = 1; 293   11080 char byte = 1;
HITCBC 292   11445 [[maybe_unused]] auto r = ::write(pipe_fds_[1], &byte, 1); 294   11080 [[maybe_unused]] auto r = ::write(pipe_fds_[1], &byte, 1);
HITCBC 293   11445 } 295   11080 }
294   296  
295   inline long 297   inline long
HITCBC 296   267620 select_scheduler::calculate_timeout(long requested_timeout_us) const 298   253656 select_scheduler::calculate_timeout(long requested_timeout_us) const
297   { 299   {
HITCBC 298   267620 if (requested_timeout_us == 0) 300   253656 if (requested_timeout_us == 0)
299   return 0; // LCOV_EXCL_LINE run_task passes 0 via task_interrupted_, never through this argument 301   return 0; // LCOV_EXCL_LINE run_task passes 0 via task_interrupted_, never through this argument
300   302  
HITCBC 301   267620 auto nearest = timer_svc_->nearest_expiry(); 303   253656 auto nearest = timer_svc_->nearest_expiry();
HITCBC 302   267620 if (nearest == timer_service::time_point::max()) 304   253656 if (nearest == timer_service::time_point::max())
HITCBC 303   741 return requested_timeout_us; 305   743 return requested_timeout_us;
304   306  
HITCBC 305   266879 auto now = std::chrono::steady_clock::now(); 307   252913 auto now = std::chrono::steady_clock::now();
HITCBC 306   266879 if (nearest <= now) 308   252913 if (nearest <= now)
HITCBC 307   683 return 0; 309   713 return 0;
308   310  
309   auto timer_timeout_us = 311   auto timer_timeout_us =
HITCBC 310   266196 std::chrono::duration_cast<std::chrono::microseconds>(nearest - now) 312   252200 std::chrono::duration_cast<std::chrono::microseconds>(nearest - now)
HITCBC 311   266196 .count(); 313   252200 .count();
312   314  
HITCBC 313   266196 constexpr auto long_max = 315   252200 constexpr auto long_max =
314   static_cast<long long>((std::numeric_limits<long>::max)()); 316   static_cast<long long>((std::numeric_limits<long>::max)());
315   auto capped_timer_us = 317   auto capped_timer_us =
HITCBC 316   266196 (std::min)((std::max)(static_cast<long long>(timer_timeout_us), 318   252200 (std::min)((std::max)(static_cast<long long>(timer_timeout_us),
HITCBC 317   266196 static_cast<long long>(0)), 319   252200 static_cast<long long>(0)),
HITCBC 318   266196 long_max); 320   252200 long_max);
319   321  
HITCBC 320   266196 if (requested_timeout_us < 0) 322   252200 if (requested_timeout_us < 0)
HITCBC 321   266194 return static_cast<long>(capped_timer_us); 323   252198 return static_cast<long>(capped_timer_us);
322   324  
323   return static_cast<long>( 325   return static_cast<long>(
HITCBC 324   2 (std::min)(static_cast<long long>(requested_timeout_us), 326   2 (std::min)(static_cast<long long>(requested_timeout_us),
HITCBC 325   2 capped_timer_us)); 327   2 capped_timer_us));
326   } 328   }
327   329  
328   inline void 330   inline void
HITCBC 329 - 291596 select_scheduler::run_task(lock_type& lock, context_type& ctx, long timeout_us) 331 + 278623 select_scheduler::run_task(
  332 + lock_type& lock, context_type& ctx, long timeout_us)
330   { 333   {
331   long effective_timeout_us = 334   long effective_timeout_us =
HITCBC 332   291596 task_interrupted_ ? 0 : calculate_timeout(timeout_us); 335   278623 task_interrupted_ ? 0 : calculate_timeout(timeout_us);
333   336  
334   // Snapshot registered descriptors while holding lock. 337   // Snapshot registered descriptors while holding lock.
335   // Record which fds need write monitoring to avoid a hot loop: 338   // Record which fds need write monitoring to avoid a hot loop:
336   // select is level-triggered so writable sockets (nearly always 339   // select is level-triggered so writable sockets (nearly always
337   // writable) would cause select() to return immediately every 340   // writable) would cause select() to return immediately every
338   // iteration if unconditionally added to write_fds. Membership 341   // iteration if unconditionally added to write_fds. Membership
339   // stays opt-in: a parked write wait opts in the same way a 342   // stays opt-in: a parked write wait opts in the same way a
340   // parked write or connect op does. 343   // parked write or connect op does.
341   struct fd_entry 344   struct fd_entry
342   { 345   {
343   int fd; 346   int fd;
344   reactor_descriptor_state* desc; 347   reactor_descriptor_state* desc;
345   bool needs_write; 348   bool needs_write;
346   }; 349   };
347   fd_entry snapshot[FD_SETSIZE]; 350   fd_entry snapshot[FD_SETSIZE];
HITCBC 348   291596 int snapshot_count = 0; 351   278623 int snapshot_count = 0;
349   352  
HITCBC 350   760329 for (auto& [fd, desc] : registered_descs_) 353   724455 for (auto& [fd, desc] : registered_descs_)
351   { 354   {
HITCBC 352   468733 if (snapshot_count < FD_SETSIZE) 355   445832 if (snapshot_count < FD_SETSIZE)
353   { 356   {
HITCBC 354   468733 conditionally_enabled_mutex::scoped_lock desc_lock(desc->mutex); 357   445832 conditionally_enabled_mutex::scoped_lock desc_lock(desc->mutex);
HITCBC 355   468733 snapshot[snapshot_count].fd = fd; 358   445832 snapshot[snapshot_count].fd = fd;
HITCBC 356   468733 snapshot[snapshot_count].desc = desc; 359   445832 snapshot[snapshot_count].desc = desc;
HITCBC 357   468733 snapshot[snapshot_count].needs_write = 360   445832 snapshot[snapshot_count].needs_write =
HITCBC 358 - 468733 (desc->write_op || desc->connect_op || desc->wait_write_op); 361 + 878709 (desc->write_op || desc->connect_op ||
HITGNC   362 + 432877 desc->wait_write_op);
HITCBC 359   468733 ++snapshot_count; 363   445832 ++snapshot_count;
HITCBC 360   468733 } 364   445832 }
361   } 365   }
362   366  
HITCBC 363   291596 if (lock.owns_lock()) 367   278623 if (lock.owns_lock())
HITCBC 364   267621 lock.unlock(); 368   253657 lock.unlock();
365   369  
HITCBC 366   291596 task_cleanup on_exit{this, &lock, ctx}; 370   278623 task_cleanup on_exit{this, &lock, ctx};
367   371  
368   fd_set read_fds, write_fds, except_fds; 372   fd_set read_fds, write_fds, except_fds;
HITCBC 369   4957132 FD_ZERO(&read_fds); 373   4736591 FD_ZERO(&read_fds);
HITCBC 370   4957132 FD_ZERO(&write_fds); 374   4736591 FD_ZERO(&write_fds);
HITCBC 371   4957132 FD_ZERO(&except_fds); 375   4736591 FD_ZERO(&except_fds);
372   376  
HITCBC 373   291596 FD_SET(pipe_fds_[0], &read_fds); 377   278623 FD_SET(pipe_fds_[0], &read_fds);
HITCBC 374   291596 int nfds = pipe_fds_[0]; 378   278623 int nfds = pipe_fds_[0];
375   379  
HITCBC 376   760329 for (int i = 0; i < snapshot_count; ++i) 380   724455 for (int i = 0; i < snapshot_count; ++i)
377   { 381   {
HITCBC 378   468733 int fd = snapshot[i].fd; 382   445832 int fd = snapshot[i].fd;
HITCBC 379   468733 FD_SET(fd, &read_fds); 383   445832 FD_SET(fd, &read_fds);
HITCBC 380   468733 if (snapshot[i].needs_write) 384   445832 if (snapshot[i].needs_write)
HITCBC 381   12010 FD_SET(fd, &write_fds); 385   12961 FD_SET(fd, &write_fds);
HITCBC 382   468733 FD_SET(fd, &except_fds); 386   445832 FD_SET(fd, &except_fds);
HITCBC 383   468733 if (fd > nfds) 387   445832 if (fd > nfds)
HITCBC 384   291200 nfds = fd; 388   278228 nfds = fd;
385   } 389   }
386   390  
387   struct timeval tv; 391   struct timeval tv;
HITCBC 388   291596 struct timeval* tv_ptr = nullptr; 392   278623 struct timeval* tv_ptr = nullptr;
HITCBC 389   291596 if (effective_timeout_us >= 0) 393   278623 if (effective_timeout_us >= 0)
390   { 394   {
HITCBC 391   290874 tv.tv_sec = effective_timeout_us / 1000000; 395   277900 tv.tv_sec = effective_timeout_us / 1000000;
HITCBC 392   290874 tv.tv_usec = effective_timeout_us % 1000000; 396   277900 tv.tv_usec = effective_timeout_us % 1000000;
HITCBC 393   290874 tv_ptr = &tv; 397   277900 tv_ptr = &tv;
394   } 398   }
395   399  
HITCBC 396   291596 int ready = ::select(nfds + 1, &read_fds, &write_fds, &except_fds, tv_ptr); 400   278623 int ready = ::select(nfds + 1, &read_fds, &write_fds, &except_fds, tv_ptr);
397   401  
398   // EINTR: signal interrupted select(), just retry. 402   // EINTR: signal interrupted select(), just retry.
399   // EBADF: an fd was closed between snapshot and select(); retry 403   // EBADF: an fd was closed between snapshot and select(); retry
400   // with a fresh snapshot from registered_descs_. 404   // with a fresh snapshot from registered_descs_.
401   // Both fall through with no ready descriptors rather than 405   // Both fall through with no ready descriptors rather than
402   // returning: the caller handed this function an owned lock that 406   // returning: the caller handed this function an owned lock that
403   // only the epilogue below re-acquires. 407   // only the epilogue below re-acquires.
HITCBC 404   291596 if (ready < 0) 408   278623 if (ready < 0)
405   { 409   {
HITCBC 406   3 if (errno != EINTR && errno != EBADF) 410   3 if (errno != EINTR && errno != EBADF)
HITCBC 407   1 detail::throw_system_error(make_err(errno), "select"); 411   1 detail::throw_system_error(make_err(errno), "select");
HITCBC 408   2 ready = 0; 412   2 ready = 0;
409   } 413   }
410   414  
411   // Process timers outside the lock 415   // Process timers outside the lock
HITCBC 412   291595 timer_svc_->process_expired(); 416   278622 timer_svc_->process_expired();
413   417  
HITCBC 414   291595 ready_queue local_ops; 418   278622 ready_queue local_ops;
415   419  
HITCBC 416   291595 if (ready > 0) 420   278622 if (ready > 0)
417   { 421   {
HITCBC 418   275405 if (FD_ISSET(pipe_fds_[0], &read_fds)) 422   262184 if (FD_ISSET(pipe_fds_[0], &read_fds))
419   { 423   {
420   char buf[256]; 424   char buf[256];
HITCBC 421   9990 while (::read(pipe_fds_[0], buf, sizeof(buf)) > 0) 425   9638 while (::read(pipe_fds_[0], buf, sizeof(buf)) > 0)
422   { 426   {
423   } 427   }
424   } 428   }
425   429  
HITCBC 426   695598 for (int i = 0; i < snapshot_count; ++i) 430   660894 for (int i = 0; i < snapshot_count; ++i)
427   { 431   {
HITCBC 428 - 420193 int fd = snapshot[i].fd; 432 + 398710 int fd = snapshot[i].fd;
HITCBC 429   420193 reactor_descriptor_state* desc = snapshot[i].desc; 433   398710 reactor_descriptor_state* desc = snapshot[i].desc;
430   434  
HITCBC 431   420193 std::uint32_t flags = 0; 435   398710 std::uint32_t flags = 0;
HITCBC 432   420193 if (FD_ISSET(fd, &read_fds)) 436   398710 if (FD_ISSET(fd, &read_fds))
HITCBC 433   275198 flags |= reactor_event_read; 437   261857 flags |= reactor_event_read;
HITCBC 434   420193 if (FD_ISSET(fd, &write_fds)) 438   398710 if (FD_ISSET(fd, &write_fds))
HITCBC 435   2135 flags |= reactor_event_write; 439   2047 flags |= reactor_event_write;
HITCBC 436   420193 if (FD_ISSET(fd, &except_fds)) 440   398710 if (FD_ISSET(fd, &except_fds))
HITCBC 437   16 flags |= reactor_event_error; 441   16 flags |= reactor_event_error;
438   442  
HITCBC 439   420193 if (flags == 0) 443   398710 if (flags == 0)
HITCBC 440   142868 continue; 444   134815 continue;
441   445  
HITCBC 442   277325 desc->add_ready_events(flags); 446   263895 desc->add_ready_events(flags);
443   447  
HITCBC 444   277325 bool expected = false; 448   263895 bool expected = false;
HITCBC 445   277325 if (desc->is_enqueued_.compare_exchange_strong( 449   263895 if (desc->is_enqueued_.compare_exchange_strong(
446   expected, true, std::memory_order_release, 450   expected, true, std::memory_order_release,
447   std::memory_order_relaxed)) 451   std::memory_order_relaxed))
448   { 452   {
HITCBC 449   277325 local_ops.push(desc); 453   263895 local_ops.push(desc);
450   } 454   }
451   } 455   }
452   } 456   }
453   457  
HITCBC 454   291595 lock.lock(); 458   278622 lock.lock();
455   459  
HITCBC 456   291595 completed_ops_.splice(local_ops); 460   278622 completed_ops_.splice(local_ops);
HITCBC 457   291596 } 461   278623 }
458   462  
459   } // namespace boost::corosio::detail 463   } // namespace boost::corosio::detail
460   464  
461   #endif // BOOST_COROSIO_HAS_SELECT 465   #endif // BOOST_COROSIO_HAS_SELECT
462   466  
463   #endif // BOOST_COROSIO_NATIVE_DETAIL_SELECT_SELECT_SCHEDULER_HPP 467   #endif // BOOST_COROSIO_NATIVE_DETAIL_SELECT_SELECT_SCHEDULER_HPP