. /include/boost/corosio /native /detail /select /select_scheduler.hpp
84.88% Lines (146/172) 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 <unordered_map> 45   #include <unordered_map>
45   46  
46   namespace boost::corosio::detail { 47   namespace boost::corosio::detail {
47   48  
48   struct select_op; 49   struct select_op;
49   50  
50   /** POSIX scheduler using select() for I/O multiplexing. 51   /** POSIX scheduler using select() for I/O multiplexing.
51   52  
52   This scheduler implements the scheduler interface using the POSIX select() 53   This scheduler implements the scheduler interface using the POSIX select()
53   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
54   model from reactor_scheduler: signal state machine, inline completion 55   model from reactor_scheduler: signal state machine, inline completion
55   budget, work counting, and the do_one event loop. 56   budget, work counting, and the do_one event loop.
56   57  
57   The design mirrors epoll_scheduler for behavioral consistency: 58   The design mirrors epoll_scheduler for behavioral consistency:
58   - Same single-reactor thread coordination model 59   - Same single-reactor thread coordination model
59   - 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)
60   - Same timer integration pattern 61   - Same timer integration pattern
61   62  
62   Known Limitations: 63   Known Limitations:
63   - FD_SETSIZE (~1024) limits maximum concurrent connections 64   - FD_SETSIZE (~1024) limits maximum concurrent connections
64   - O(n) scanning: rebuilds fd_sets each iteration 65   - O(n) scanning: rebuilds fd_sets each iteration
65   - Level-triggered only (no edge-triggered mode) 66   - Level-triggered only (no edge-triggered mode)
66   67  
67   @par Thread Safety 68   @par Thread Safety
68   All public member functions are thread-safe. 69   All public member functions are thread-safe.
69   */ 70   */
70   class BOOST_COROSIO_DECL select_scheduler final : public reactor_scheduler 71   class BOOST_COROSIO_DECL select_scheduler final : public reactor_scheduler
71   { 72   {
72   public: 73   public:
73   /** Construct the scheduler. 74   /** Construct the scheduler.
74   75  
75   Creates a self-pipe for reactor interruption. 76   Creates a self-pipe for reactor interruption.
76   77  
77   @param ctx Reference to the owning execution_context. 78   @param ctx Reference to the owning execution_context.
78   @param concurrency_hint Hint for expected thread count (unused). 79   @param concurrency_hint Hint for expected thread count (unused).
79   */ 80   */
80   select_scheduler(capy::execution_context& ctx, int concurrency_hint = -1); 81   select_scheduler(capy::execution_context& ctx, int concurrency_hint = -1);
81   82  
82   /// Destroy the scheduler. 83   /// Destroy the scheduler.
83   ~select_scheduler() override; 84   ~select_scheduler() override;
84   85  
85   select_scheduler(select_scheduler const&) = delete; 86   select_scheduler(select_scheduler const&) = delete;
86   select_scheduler& operator=(select_scheduler const&) = delete; 87   select_scheduler& operator=(select_scheduler const&) = delete;
87   88  
88   /// Shut down the scheduler, draining pending operations. 89   /// Shut down the scheduler, draining pending operations.
89   void shutdown() override; 90   void shutdown() override;
90   91  
91   /** Return the maximum file descriptor value supported. 92   /** Return the maximum file descriptor value supported.
92   93  
93   Returns FD_SETSIZE - 1, the maximum fd value that can be 94   Returns FD_SETSIZE - 1, the maximum fd value that can be
94   monitored by select(). Operations with fd >= FD_SETSIZE 95   monitored by select(). Operations with fd >= FD_SETSIZE
95   will fail with EINVAL. 96   will fail with EINVAL.
96   97  
97   @return The maximum supported file descriptor value. 98   @return The maximum supported file descriptor value.
98   */ 99   */
99   static constexpr int max_fd() noexcept 100   static constexpr int max_fd() noexcept
100   { 101   {
101   return FD_SETSIZE - 1; 102   return FD_SETSIZE - 1;
102   } 103   }
103   104  
104   /** Register a descriptor for persistent monitoring. 105   /** Register a descriptor for persistent monitoring.
105   106  
106   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
107   included in subsequent select() calls. The reactor is 108   included in subsequent select() calls. The reactor is
108   interrupted so a blocked select() rebuilds its fd_sets. 109   interrupted so a blocked select() rebuilds its fd_sets.
109   110  
110   @param fd The file descriptor to register. 111   @param fd The file descriptor to register.
111   @param desc Pointer to descriptor state for this fd. 112   @param desc Pointer to descriptor state for this fd.
  113 +
  114 + @return The error if the fd cannot be tracked, otherwise a
  115 + default constructed error code.
112   */ 116   */
113 - void register_descriptor(int fd, reactor_descriptor_state* desc) const; 117 + std::error_code
  118 + register_descriptor(int fd, reactor_descriptor_state* desc) const;
114   119  
115   /** Deregister a persistently registered descriptor. 120   /** Deregister a persistently registered descriptor.
116   121  
117   @param fd The file descriptor to deregister. 122   @param fd The file descriptor to deregister.
118   */ 123   */
119   void deregister_descriptor(int fd) const; 124   void deregister_descriptor(int fd) const;
120   125  
121   /** Interrupt the reactor so it rebuilds its fd_sets. 126   /** Interrupt the reactor so it rebuilds its fd_sets.
122   127  
123 - Called when a write or connect op is registered after 128 + Called when a write, connect, or write-wait op is registered
124 - the reactor's snapshot was taken. Without this, select() 129 + after the reactor's snapshot was taken. Without this,
125 - may block not watching for writability on the fd. 130 + select() may block not watching for writability on the fd.
126   */ 131   */
127   void notify_reactor() const; 132   void notify_reactor() const;
128   133  
129   /// 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).
HITCBC 130   41 void register_signal_reader(int read_fd) override 135   41 void register_signal_reader(int read_fd) override
131   { 136   {
HITCBC 132 - 41 register_descriptor(read_fd, signal_pipe_reader_.arm()); 137 + 41 if (auto ec = register_descriptor(read_fd, signal_pipe_reader_.arm()))
MISUNC   138 + ✗ detail::throw_system_error(ec, "select: register");
HITCBC 133   41 } 139   41 }
134   140  
135   private: 141   private:
136   void 142   void
137   run_task(lock_type& lock, context_type* ctx, 143   run_task(lock_type& lock, context_type* ctx,
138   long timeout_us) override; 144   long timeout_us) override;
139   void interrupt_reactor() const override; 145   void interrupt_reactor() const override;
140   long calculate_timeout(long requested_timeout_us) const; 146   long calculate_timeout(long requested_timeout_us) const;
141   147  
142   // 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
143   // register_signal_reader on the first signal registration). 149   // register_signal_reader on the first signal registration).
144   reactor_signal_pipe_reader signal_pipe_reader_; 150   reactor_signal_pipe_reader signal_pipe_reader_;
145   151  
146   // Self-pipe for interrupting select() 152   // Self-pipe for interrupting select()
147   int pipe_fds_[2]; // [0]=read, [1]=write 153   int pipe_fds_[2]; // [0]=read, [1]=write
148   154  
149   // Per-fd tracking for fd_set building 155   // Per-fd tracking for fd_set building
150   mutable std::unordered_map<int, reactor_descriptor_state*> registered_descs_; 156   mutable std::unordered_map<int, reactor_descriptor_state*> registered_descs_;
151   mutable int max_fd_ = -1; 157   mutable int max_fd_ = -1;
152   }; 158   };
153   159  
HITCBC 154   625 inline select_scheduler::select_scheduler(capy::execution_context& ctx, int) 160   666 inline select_scheduler::select_scheduler(capy::execution_context& ctx, int)
HITCBC 155   625 : pipe_fds_{-1, -1} 161   666 : pipe_fds_{-1, -1}
HITCBC 156   625 , max_fd_(-1) 162   666 , max_fd_(-1)
157   { 163   {
HITCBC 158   625 if (::pipe(pipe_fds_) < 0) 164   666 if (::pipe(pipe_fds_) < 0)
MISUBC 159   ✗ detail::throw_system_error(make_err(errno), "pipe"); 165   ✗ detail::throw_system_error(make_err(errno), "pipe");
160   166  
HITCBC 161   1875 for (int i = 0; i < 2; ++i) 167   1998 for (int i = 0; i < 2; ++i)
162   { 168   {
HITCBC 163   1250 int flags = ::fcntl(pipe_fds_[i], F_GETFL, 0); 169   1332 int flags = ::fcntl(pipe_fds_[i], F_GETFL, 0);
HITCBC 164   1250 if (flags == -1) 170   1332 if (flags == -1)
165   { 171   {
MISUBC 166   ✗ int errn = errno; 172   ✗ int errn = errno;
MISUBC 167   ✗ ::close(pipe_fds_[0]); 173   ✗ ::close(pipe_fds_[0]);
MISUBC 168   ✗ ::close(pipe_fds_[1]); 174   ✗ ::close(pipe_fds_[1]);
MISUBC 169   ✗ detail::throw_system_error(make_err(errn), "fcntl F_GETFL"); 175   ✗ detail::throw_system_error(make_err(errn), "fcntl F_GETFL");
170   } 176   }
HITCBC 171   1250 if (::fcntl(pipe_fds_[i], F_SETFL, flags | O_NONBLOCK) == -1) 177   1332 if (::fcntl(pipe_fds_[i], F_SETFL, flags | O_NONBLOCK) == -1)
172   { 178   {
MISUBC 173   ✗ int errn = errno; 179   ✗ int errn = errno;
MISUBC 174   ✗ ::close(pipe_fds_[0]); 180   ✗ ::close(pipe_fds_[0]);
MISUBC 175   ✗ ::close(pipe_fds_[1]); 181   ✗ ::close(pipe_fds_[1]);
MISUBC 176   ✗ detail::throw_system_error(make_err(errn), "fcntl F_SETFL"); 182   ✗ detail::throw_system_error(make_err(errn), "fcntl F_SETFL");
177   } 183   }
HITCBC 178   1250 if (::fcntl(pipe_fds_[i], F_SETFD, FD_CLOEXEC) == -1) 184   1332 if (::fcntl(pipe_fds_[i], F_SETFD, FD_CLOEXEC) == -1)
179   { 185   {
MISUBC 180   ✗ int errn = errno; 186   ✗ int errn = errno;
MISUBC 181   ✗ ::close(pipe_fds_[0]); 187   ✗ ::close(pipe_fds_[0]);
MISUBC 182   ✗ ::close(pipe_fds_[1]); 188   ✗ ::close(pipe_fds_[1]);
MISUBC 183   ✗ detail::throw_system_error(make_err(errn), "fcntl F_SETFD"); 189   ✗ detail::throw_system_error(make_err(errn), "fcntl F_SETFD");
184   } 190   }
185   } 191   }
186   192  
HITCBC 187   625 timer_svc_ = &get_timer_service(ctx, *this); 193   666 timer_svc_ = &get_timer_service(ctx, *this);
HITCBC 188   625 timer_svc_->set_on_earliest_changed( 194   666 timer_svc_->set_on_earliest_changed(
HITCBC 189   3475 timer_service::callback(this, [](void* p) { 195   3646 timer_service::callback(this, [](void* p) {
HITCBC 190   2850 static_cast<select_scheduler*>(p)->interrupt_reactor(); 196   2980 static_cast<select_scheduler*>(p)->interrupt_reactor();
HITCBC 191   2850 })); 197   2980 }));
192   198  
HITCBC 193   625 get_resolver_service(ctx, *this); 199   666 get_resolver_service(ctx, *this);
HITCBC 194   625 get_signal_service(ctx, *this); 200   666 get_signal_service(ctx, *this);
HITCBC 195   625 get_stream_file_service(ctx, *this); 201   666 get_stream_file_service(ctx, *this);
HITCBC 196   625 get_random_access_file_service(ctx, *this); 202   666 get_random_access_file_service(ctx, *this);
197   203  
HITCBC 198   625 completed_ops_.push(&task_op_); 204   666 completed_ops_.push(&task_op_);
HITCBC 199   625 } 205   666 }
200   206  
HITCBC 201   1250 inline select_scheduler::~select_scheduler() 207   1332 inline select_scheduler::~select_scheduler()
202   { 208   {
HITCBC 203   625 if (pipe_fds_[0] >= 0) 209   666 if (pipe_fds_[0] >= 0)
HITCBC 204   625 ::close(pipe_fds_[0]); 210   666 ::close(pipe_fds_[0]);
HITCBC 205   625 if (pipe_fds_[1] >= 0) 211   666 if (pipe_fds_[1] >= 0)
HITCBC 206   625 ::close(pipe_fds_[1]); 212   666 ::close(pipe_fds_[1]);
HITCBC 207   1250 } 213   1332 }
208   214  
209   inline void 215   inline void
HITCBC 210   625 select_scheduler::shutdown() 216   666 select_scheduler::shutdown()
211   { 217   {
HITCBC 212   625 shutdown_drain(); 218   666 shutdown_drain();
213   219  
HITCBC 214   625 if (pipe_fds_[1] >= 0) 220   666 if (pipe_fds_[1] >= 0)
HITCBC 215   625 interrupt_reactor(); 221   666 interrupt_reactor();
HITCBC 216   625 } 222   666 }
217   223  
218 - inline void 224 + inline std::error_code
HITCBC 219   4368 select_scheduler::register_descriptor( 225   4809 select_scheduler::register_descriptor(
220   int fd, reactor_descriptor_state* desc) const 226   int fd, reactor_descriptor_state* desc) const
221   { 227   {
HITCBC 222   4368 if (fd < 0 || fd >= FD_SETSIZE) 228   4809 if (fd < 0 || fd >= FD_SETSIZE)
MISUBC 223 - ✗ detail::throw_system_error(make_err(EINVAL), "select: fd out of range"); 229 + ✗ return make_err(EMFILE);
224   230  
HITCBC 225   4368 desc->registered_events = reactor_event_read | reactor_event_write; 231   4809 desc->registered_events = reactor_event_read | reactor_event_write;
HITCBC 226   4368 desc->fd = fd; 232   4809 desc->fd = fd;
HITCBC 227   4368 desc->scheduler_ = this; 233   4809 desc->scheduler_ = this;
HITCBC 228   4368 desc->mutex.set_enabled(reactor_io_locking_); 234   4809 desc->mutex.set_enabled(reactor_io_locking_);
HITCBC 229   4368 desc->ready_events_.store(0, std::memory_order_relaxed); 235   4809 desc->ready_events_.store(0, std::memory_order_relaxed);
230   236  
231   { 237   {
HITCBC 232   4368 conditionally_enabled_mutex::scoped_lock lock(desc->mutex); 238   4809 conditionally_enabled_mutex::scoped_lock lock(desc->mutex);
HITCBC 233   4368 desc->impl_ref_.reset(); 239   4809 desc->impl_ref_.reset();
HITCBC 234   4368 desc->read_ready = false; 240   4809 desc->read_ready = false;
HITCBC 235   4368 desc->write_ready = false; 241   4809 desc->write_ready = false;
HITCBC 236   4368 } 242   4809 }
237   243  
238   { 244   {
HITCBC 239   4368 mutex_type::scoped_lock lock(mutex_); 245   4809 mutex_type::scoped_lock lock(mutex_);
ECB 240 - 4368 registered_descs_[fd] = desc; 246 + try
  247 + {
HITGNC   248 + 4809 registered_descs_[fd] = desc;
  249 + }
MISUNC   250 + ✗ catch (std::bad_alloc const&)
  251 + {
MISUNC   252 + ✗ return make_err(ENOMEM);
MISUNC   253 + ✗ }
HITCBC 241   4368 if (fd > max_fd_) 254   4809 if (fd > max_fd_)
HITCBC 242   4362 max_fd_ = fd; 255   4799 max_fd_ = fd;
HITCBC 243   4368 } 256   4809 }
244   257  
HITCBC 245   4368 interrupt_reactor(); 258   4809 interrupt_reactor();
HITGNC   259 + 4809 return {};
ECB 246   4368 } 260   }
247   261  
248   inline void 262   inline void
HITCBC 249   4327 select_scheduler::deregister_descriptor(int fd) const 263   4768 select_scheduler::deregister_descriptor(int fd) const
250   { 264   {
HITCBC 251   4327 mutex_type::scoped_lock lock(mutex_); 265   4768 mutex_type::scoped_lock lock(mutex_);
252   266  
HITCBC 253   4327 auto it = registered_descs_.find(fd); 267   4768 auto it = registered_descs_.find(fd);
HITCBC 254   4327 if (it == registered_descs_.end()) 268   4768 if (it == registered_descs_.end())
MISUBC 255   ✗ return; 269   ✗ return;
256   270  
HITCBC 257   4327 registered_descs_.erase(it); 271   4768 registered_descs_.erase(it);
258   272  
HITCBC 259   4327 if (fd == max_fd_) 273   4768 if (fd == max_fd_)
260   { 274   {
HITCBC 261   4193 max_fd_ = pipe_fds_[0]; 275   4612 max_fd_ = pipe_fds_[0];
HITCBC 262   8118 for (auto& [registered_fd, state] : registered_descs_) 276   8925 for (auto& [registered_fd, state] : registered_descs_)
263   { 277   {
HITCBC 264   3925 if (registered_fd > max_fd_) 278   4313 if (registered_fd > max_fd_)
HITCBC 265   3876 max_fd_ = registered_fd; 279   4258 max_fd_ = registered_fd;
266   } 280   }
267   } 281   }
HITCBC 268   4327 } 282   4768 }
269   283  
270   inline void 284   inline void
HITCBC 271   2053 select_scheduler::notify_reactor() const 285   2242 select_scheduler::notify_reactor() const
272   { 286   {
HITCBC 273   2053 interrupt_reactor(); 287   2242 interrupt_reactor();
HITCBC 274   2053 } 288   2242 }
275   289  
276   inline void 290   inline void
HITCBC 277   10390 select_scheduler::interrupt_reactor() const 291   11236 select_scheduler::interrupt_reactor() const
278   { 292   {
HITCBC 279   10390 char byte = 1; 293   11236 char byte = 1;
HITCBC 280   10390 [[maybe_unused]] auto r = ::write(pipe_fds_[1], &byte, 1); 294   11236 [[maybe_unused]] auto r = ::write(pipe_fds_[1], &byte, 1);
HITCBC 281   10390 } 295   11236 }
282   296  
283   inline long 297   inline long
HITCBC 284   240823 select_scheduler::calculate_timeout(long requested_timeout_us) const 298   302220 select_scheduler::calculate_timeout(long requested_timeout_us) const
285   { 299   {
HITCBC 286   240823 if (requested_timeout_us == 0) 300   302220 if (requested_timeout_us == 0)
MISUBC 287   ✗ return 0; 301   ✗ return 0;
288   302  
HITCBC 289   240823 auto nearest = timer_svc_->nearest_expiry(); 303   302220 auto nearest = timer_svc_->nearest_expiry();
HITCBC 290   240823 if (nearest == timer_service::time_point::max()) 304   302220 if (nearest == timer_service::time_point::max())
HITCBC 291   573 return requested_timeout_us; 305   587 return requested_timeout_us;
292   306  
HITCBC 293   240250 auto now = std::chrono::steady_clock::now(); 307   301633 auto now = std::chrono::steady_clock::now();
HITCBC 294   240250 if (nearest <= now) 308   301633 if (nearest <= now)
HITCBC 295   809 return 0; 309   1314 return 0;
296   310  
297   auto timer_timeout_us = 311   auto timer_timeout_us =
HITCBC 298   239441 std::chrono::duration_cast<std::chrono::microseconds>(nearest - now) 312   300319 std::chrono::duration_cast<std::chrono::microseconds>(nearest - now)
HITCBC 299   239441 .count(); 313   300319 .count();
300   314  
HITCBC 301   239441 constexpr auto long_max = 315   300319 constexpr auto long_max =
302   static_cast<long long>((std::numeric_limits<long>::max)()); 316   static_cast<long long>((std::numeric_limits<long>::max)());
303   auto capped_timer_us = 317   auto capped_timer_us =
HITCBC 304   239441 (std::min)((std::max)(static_cast<long long>(timer_timeout_us), 318   300319 (std::min)((std::max)(static_cast<long long>(timer_timeout_us),
HITCBC 305   239441 static_cast<long long>(0)), 319   300319 static_cast<long long>(0)),
HITCBC 306   239441 long_max); 320   300319 long_max);
307   321  
HITCBC 308   239441 if (requested_timeout_us < 0) 322   300319 if (requested_timeout_us < 0)
HITCBC 309   239441 return static_cast<long>(capped_timer_us); 323   300319 return static_cast<long>(capped_timer_us);
310   324  
311   return static_cast<long>( 325   return static_cast<long>(
MISUBC 312   ✗ (std::min)(static_cast<long long>(requested_timeout_us), 326   ✗ (std::min)(static_cast<long long>(requested_timeout_us),
MISUBC 313   ✗ capped_timer_us)); 327   ✗ capped_timer_us));
314   } 328   }
315   329  
316   inline void 330   inline void
HITCBC 317   263488 select_scheduler::run_task( 331   322396 select_scheduler::run_task(
318   lock_type& lock, context_type* ctx, long timeout_us) 332   lock_type& lock, context_type* ctx, long timeout_us)
319   { 333   {
320   long effective_timeout_us = 334   long effective_timeout_us =
HITCBC 321   263488 task_interrupted_ ? 0 : calculate_timeout(timeout_us); 335   322396 task_interrupted_ ? 0 : calculate_timeout(timeout_us);
322   336  
323   // Snapshot registered descriptors while holding lock. 337   // Snapshot registered descriptors while holding lock.
324   // Record which fds need write monitoring to avoid a hot loop: 338   // Record which fds need write monitoring to avoid a hot loop:
325   // select is level-triggered so writable sockets (nearly always 339   // select is level-triggered so writable sockets (nearly always
326   // writable) would cause select() to return immediately every 340   // writable) would cause select() to return immediately every
327 - // iteration if unconditionally added to write_fds. 341 + // iteration if unconditionally added to write_fds. Membership
  342 + // stays opt-in: a parked write wait opts in the same way a
  343 + // parked write or connect op does.
328   struct fd_entry 344   struct fd_entry
329   { 345   {
330   int fd; 346   int fd;
331   reactor_descriptor_state* desc; 347   reactor_descriptor_state* desc;
332   bool needs_write; 348   bool needs_write;
333   }; 349   };
334   fd_entry snapshot[FD_SETSIZE]; 350   fd_entry snapshot[FD_SETSIZE];
HITCBC 335   263488 int snapshot_count = 0; 351   322396 int snapshot_count = 0;
336   352  
HITCBC 337   667963 for (auto& [fd, desc] : registered_descs_) 353   770543 for (auto& [fd, desc] : registered_descs_)
338   { 354   {
HITCBC 339   404475 if (snapshot_count < FD_SETSIZE) 355   448147 if (snapshot_count < FD_SETSIZE)
340   { 356   {
HITCBC 341   404475 conditionally_enabled_mutex::scoped_lock desc_lock(desc->mutex); 357   448147 conditionally_enabled_mutex::scoped_lock desc_lock(desc->mutex);
HITCBC 342   404475 snapshot[snapshot_count].fd = fd; 358   448147 snapshot[snapshot_count].fd = fd;
HITCBC 343   404475 snapshot[snapshot_count].desc = desc; 359   448147 snapshot[snapshot_count].desc = desc;
HITCBC 344   404475 snapshot[snapshot_count].needs_write = 360   448147 snapshot[snapshot_count].needs_write =
HITCBC 345 - 404475 (desc->write_op || desc->connect_op); 361 + 883434 (desc->write_op || desc->connect_op ||
HITGNC   362 + 435287 desc->wait_write_op);
HITCBC 346   404475 ++snapshot_count; 363   448147 ++snapshot_count;
HITCBC 347   404475 } 364   448147 }
348   } 365   }
349   366  
HITCBC 350   263488 if (lock.owns_lock()) 367   322396 if (lock.owns_lock())
HITCBC 351   240824 lock.unlock(); 368   302221 lock.unlock();
352   369  
HITCBC 353   263488 task_cleanup on_exit{this, &lock, ctx}; 370   322396 task_cleanup on_exit{this, &lock, ctx};
354   371  
355   fd_set read_fds, write_fds, except_fds; 372   fd_set read_fds, write_fds, except_fds;
HITCBC 356   4479296 FD_ZERO(&read_fds); 373   5480732 FD_ZERO(&read_fds);
HITCBC 357   4479296 FD_ZERO(&write_fds); 374   5480732 FD_ZERO(&write_fds);
HITCBC 358   4479296 FD_ZERO(&except_fds); 375   5480732 FD_ZERO(&except_fds);
359   376  
HITCBC 360   263488 FD_SET(pipe_fds_[0], &read_fds); 377   322396 FD_SET(pipe_fds_[0], &read_fds);
HITCBC 361   263488 int nfds = pipe_fds_[0]; 378   322396 int nfds = pipe_fds_[0];
362   379  
HITCBC 363   667963 for (int i = 0; i < snapshot_count; ++i) 380   770543 for (int i = 0; i < snapshot_count; ++i)
364   { 381   {
HITCBC 365   404475 int fd = snapshot[i].fd; 382   448147 int fd = snapshot[i].fd;
HITCBC 366   404475 FD_SET(fd, &read_fds); 383   448147 FD_SET(fd, &read_fds);
HITCBC 367   404475 if (snapshot[i].needs_write) 384   448147 if (snapshot[i].needs_write)
HITCBC 368   12504 FD_SET(fd, &write_fds); 385   12864 FD_SET(fd, &write_fds);
HITCBC 369   404475 FD_SET(fd, &except_fds); 386   448147 FD_SET(fd, &except_fds);
HITCBC 370   404475 if (fd > nfds) 387   448147 if (fd > nfds)
HITCBC 371   263128 nfds = fd; 388   322029 nfds = fd;
372   } 389   }
373   390  
374   struct timeval tv; 391   struct timeval tv;
HITCBC 375   263488 struct timeval* tv_ptr = nullptr; 392   322396 struct timeval* tv_ptr = nullptr;
HITCBC 376   263488 if (effective_timeout_us >= 0) 393   322396 if (effective_timeout_us >= 0)
377   { 394   {
HITCBC 378   262918 tv.tv_sec = effective_timeout_us / 1000000; 395   321813 tv.tv_sec = effective_timeout_us / 1000000;
HITCBC 379   262918 tv.tv_usec = effective_timeout_us % 1000000; 396   321813 tv.tv_usec = effective_timeout_us % 1000000;
HITCBC 380   262918 tv_ptr = &tv; 397   321813 tv_ptr = &tv;
381   } 398   }
382   399  
HITCBC 383   263488 int ready = ::select(nfds + 1, &read_fds, &write_fds, &except_fds, tv_ptr); 400   322396 int ready = ::select(nfds + 1, &read_fds, &write_fds, &except_fds, tv_ptr);
384   401  
385   // EINTR: signal interrupted select(), just retry. 402   // EINTR: signal interrupted select(), just retry.
386   // EBADF: an fd was closed between snapshot and select(); retry 403   // EBADF: an fd was closed between snapshot and select(); retry
387   // with a fresh snapshot from registered_descs_. 404   // with a fresh snapshot from registered_descs_.
HITCBC 388   263488 if (ready < 0) 405   322396 if (ready < 0)
389   { 406   {
MISUBC 390   ✗ if (errno == EINTR || errno == EBADF) 407   ✗ if (errno == EINTR || errno == EBADF)
MISUBC 391   ✗ return; 408   ✗ return;
MISUBC 392   ✗ detail::throw_system_error(make_err(errno), "select"); 409   ✗ detail::throw_system_error(make_err(errno), "select");
393   } 410   }
394   411  
395   // Process timers outside the lock 412   // Process timers outside the lock
HITCBC 396   263488 timer_svc_->process_expired(); 413   322396 timer_svc_->process_expired();
397   414  
HITCBC 398   263488 ready_queue local_ops; 415   322396 ready_queue local_ops;
399   416  
HITCBC 400   263488 if (ready > 0) 417   322396 if (ready > 0)
401   { 418   {
HITCBC 402   247883 if (FD_ISSET(pipe_fds_[0], &read_fds)) 419   309077 if (FD_ISSET(pipe_fds_[0], &read_fds))
403   { 420   {
404   char buf[256]; 421   char buf[256];
HITCBC 405   9212 while (::read(pipe_fds_[0], buf, sizeof(buf)) > 0) 422   10058 while (::read(pipe_fds_[0], buf, sizeof(buf)) > 0)
406   { 423   {
407   } 424   }
408   } 425   }
409   426  
HITCBC 410   605354 for (int i = 0; i < snapshot_count; ++i) 427   715836 for (int i = 0; i < snapshot_count; ++i)
411   { 428   {
HITCBC 412   357471 int fd = snapshot[i].fd; 429   406759 int fd = snapshot[i].fd;
HITCBC 413   357471 reactor_descriptor_state* desc = snapshot[i].desc; 430   406759 reactor_descriptor_state* desc = snapshot[i].desc;
414   431  
HITCBC 415   357471 std::uint32_t flags = 0; 432   406759 std::uint32_t flags = 0;
HITCBC 416   357471 if (FD_ISSET(fd, &read_fds)) 433   406759 if (FD_ISSET(fd, &read_fds))
HITCBC 417   245489 flags |= reactor_event_read; 434   306478 flags |= reactor_event_read;
HITCBC 418   357471 if (FD_ISSET(fd, &write_fds)) 435   406759 if (FD_ISSET(fd, &write_fds))
HITCBC 419   2049 flags |= reactor_event_write; 436   2237 flags |= reactor_event_write;
HITCBC 420   357471 if (FD_ISSET(fd, &except_fds)) 437   406759 if (FD_ISSET(fd, &except_fds))
MISUBC 421   ✗ flags |= reactor_event_error; 438   ✗ flags |= reactor_event_error;
422   439  
HITCBC 423   357471 if (flags == 0) 440   406759 if (flags == 0)
HITCBC 424   109943 continue; 441   98054 continue;
425   442  
HITCBC 426   247528 desc->add_ready_events(flags); 443   308705 desc->add_ready_events(flags);
427   444  
HITCBC 428   247528 bool expected = false; 445   308705 bool expected = false;
HITCBC 429   247528 if (desc->is_enqueued_.compare_exchange_strong( 446   308705 if (desc->is_enqueued_.compare_exchange_strong(
430   expected, true, std::memory_order_release, 447   expected, true, std::memory_order_release,
431   std::memory_order_relaxed)) 448   std::memory_order_relaxed))
432   { 449   {
HITCBC 433   247528 local_ops.push(desc); 450   308705 local_ops.push(desc);
434   } 451   }
435   } 452   }
436   } 453   }
437   454  
HITCBC 438   263488 lock.lock(); 455   322396 lock.lock();
439   456  
HITCBC 440   263488 completed_ops_.splice(local_ops); 457   322396 completed_ops_.splice(local_ops);
HITCBC 441   263488 } 458   322396 }
442   459  
443   } // namespace boost::corosio::detail 460   } // namespace boost::corosio::detail
444   461  
445   #endif // BOOST_COROSIO_HAS_SELECT 462   #endif // BOOST_COROSIO_HAS_SELECT
446   463  
447   #endif // BOOST_COROSIO_NATIVE_DETAIL_SELECT_SELECT_SCHEDULER_HPP 464   #endif // BOOST_COROSIO_NATIVE_DETAIL_SELECT_SELECT_SCHEDULER_HPP