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