. /include/boost/corosio /native /detail /select /select_scheduler.hpp
99.44% Lines (178/179) 100.00% Functions (13/13)
[ ] prev/next file n p prev/next uncovered line
TLA Baseline Branch
Line Hits Code Line Hits Code
1   // 1   //
2   // Copyright (c) 2026 Steve Gerbino 2   // Copyright (c) 2026 Steve Gerbino
3   // Copyright (c) 2026 Michael Vandeberg 3   // Copyright (c) 2026 Michael Vandeberg
4   // 4   //
5   // Distributed under the Boost Software License, Version 1.0. (See accompanying 5   // Distributed under the Boost Software License, Version 1.0. (See accompanying
6   // file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt) 6   // file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
7   // 7   //
8   // Official repository: https://github.com/cppalliance/corosio 8   // Official repository: https://github.com/cppalliance/corosio
9   // 9   //
10   10  
11   #ifndef BOOST_COROSIO_NATIVE_DETAIL_SELECT_SELECT_SCHEDULER_HPP 11   #ifndef BOOST_COROSIO_NATIVE_DETAIL_SELECT_SELECT_SCHEDULER_HPP
12   #define BOOST_COROSIO_NATIVE_DETAIL_SELECT_SELECT_SCHEDULER_HPP 12   #define BOOST_COROSIO_NATIVE_DETAIL_SELECT_SELECT_SCHEDULER_HPP
13   13  
14   #include <boost/corosio/detail/platform.hpp> 14   #include <boost/corosio/detail/platform.hpp>
15   15  
16   #if BOOST_COROSIO_HAS_SELECT 16   #if BOOST_COROSIO_HAS_SELECT
17   17  
18   #include <boost/corosio/detail/config.hpp> 18   #include <boost/corosio/detail/config.hpp>
19   #include <boost/capy/ex/execution_context.hpp> 19   #include <boost/capy/ex/execution_context.hpp>
20   20  
21   #include <boost/corosio/native/detail/reactor/reactor_scheduler.hpp> 21   #include <boost/corosio/native/detail/reactor/reactor_scheduler.hpp>
22   #include <boost/corosio/native/detail/reactor/reactor_signal_pipe.hpp> 22   #include <boost/corosio/native/detail/reactor/reactor_signal_pipe.hpp>
23   23  
24   #include <boost/corosio/native/detail/select/select_traits.hpp> 24   #include <boost/corosio/native/detail/select/select_traits.hpp>
25   #include <boost/corosio/detail/timer_service.hpp> 25   #include <boost/corosio/detail/timer_service.hpp>
26   #include <boost/corosio/native/detail/make_err.hpp> 26   #include <boost/corosio/native/detail/make_err.hpp>
27   27  
28   #include <boost/corosio/detail/except.hpp> 28   #include <boost/corosio/detail/except.hpp>
29   29  
30   #include <sys/select.h> 30   #include <sys/select.h>
31   #include <unistd.h> 31   #include <unistd.h>
32   #include <errno.h> 32   #include <errno.h>
33   #include <fcntl.h> 33   #include <fcntl.h>
34   34  
35   #include <atomic> 35   #include <atomic>
36   #include <chrono> 36   #include <chrono>
37   #include <cstdint> 37   #include <cstdint>
38   #include <limits> 38   #include <limits>
39   #include <mutex> 39   #include <mutex>
40   #include <new> 40   #include <new>
41   #include <unordered_map> 41   #include <unordered_map>
42   42  
43   namespace boost::corosio::detail { 43   namespace boost::corosio::detail {
44   44  
45   struct select_op; 45   struct select_op;
46   46  
47   /** POSIX scheduler using select() for I/O multiplexing. 47   /** POSIX scheduler using select() for I/O multiplexing.
48   48  
49   This scheduler implements the scheduler interface using the POSIX select() 49   This scheduler implements the scheduler interface using the POSIX select()
50   call for I/O event notification. It inherits the shared reactor threading 50   call for I/O event notification. It inherits the shared reactor threading
51   model from reactor_scheduler: signal state machine, inline completion 51   model from reactor_scheduler: signal state machine, inline completion
52   budget, work counting, and the do_one event loop. 52   budget, work counting, and the do_one event loop.
53   53  
54   The design mirrors epoll_scheduler for behavioral consistency: 54   The design mirrors epoll_scheduler for behavioral consistency:
55   - Same single-reactor thread coordination model 55   - Same single-reactor thread coordination model
56   - Same deferred I/O pattern (reactor marks ready; workers do I/O) 56   - Same deferred I/O pattern (reactor marks ready; workers do I/O)
57   - Same timer integration pattern 57   - Same timer integration pattern
58   58  
59   Known Limitations: 59   Known Limitations:
60   - FD_SETSIZE (~1024) limits maximum concurrent connections 60   - FD_SETSIZE (~1024) limits maximum concurrent connections
61   - O(n) scanning: rebuilds fd_sets each iteration 61   - O(n) scanning: rebuilds fd_sets each iteration
62   - Level-triggered only (no edge-triggered mode) 62   - Level-triggered only (no edge-triggered mode)
63   63  
64   @par Thread Safety 64   @par Thread Safety
65   All public member functions are thread-safe. 65   All public member functions are thread-safe.
66   */ 66   */
67   class BOOST_COROSIO_DECL select_scheduler final : public reactor_scheduler 67   class BOOST_COROSIO_DECL select_scheduler final : public reactor_scheduler
68   { 68   {
69   public: 69   public:
70   /** Construct the scheduler. 70   /** Construct the scheduler.
71   71  
72   Creates a self-pipe for reactor interruption. 72   Creates a self-pipe for reactor interruption.
73   73  
74   @param ctx Reference to the owning execution_context. 74   @param ctx Reference to the owning execution_context.
75   @param concurrency_hint Hint for expected thread count (unused). 75   @param concurrency_hint Hint for expected thread count (unused).
76   */ 76   */
77   select_scheduler(capy::execution_context& ctx, int concurrency_hint = -1); 77   select_scheduler(capy::execution_context& ctx, int concurrency_hint = -1);
78   78  
79   /// Destroy the scheduler. 79   /// Destroy the scheduler.
80   ~select_scheduler() override; 80   ~select_scheduler() override;
81   81  
82   select_scheduler(select_scheduler const&) = delete; 82   select_scheduler(select_scheduler const&) = delete;
83   select_scheduler& operator=(select_scheduler const&) = delete; 83   select_scheduler& operator=(select_scheduler const&) = delete;
84   84  
85   /// Shut down the scheduler, draining pending operations. 85   /// Shut down the scheduler, draining pending operations.
86   void shutdown() override; 86   void shutdown() override;
87   87  
88   /** Return the maximum file descriptor value supported. 88   /** Return the maximum file descriptor value supported.
89   89  
90   Returns FD_SETSIZE - 1, the maximum fd value that can be 90   Returns FD_SETSIZE - 1, the maximum fd value that can be
91   monitored by select(). Operations with fd >= FD_SETSIZE 91   monitored by select(). Operations with fd >= FD_SETSIZE
92   will fail with EINVAL. 92   will fail with EINVAL.
93   93  
94   @return The maximum supported file descriptor value. 94   @return The maximum supported file descriptor value.
95   */ 95   */
96   static constexpr int max_fd() noexcept 96   static constexpr int max_fd() noexcept
97   { 97   {
98   return FD_SETSIZE - 1; 98   return FD_SETSIZE - 1;
99   } 99   }
100   100  
101   /** Register a descriptor for persistent monitoring. 101   /** Register a descriptor for persistent monitoring.
102   102  
103   The fd is added to the registered_descs_ map and will be 103   The fd is added to the registered_descs_ map and will be
104   included in subsequent select() calls. The reactor is 104   included in subsequent select() calls. The reactor is
105   interrupted so a blocked select() rebuilds its fd_sets. 105   interrupted so a blocked select() rebuilds its fd_sets.
106   106  
107   @param fd The file descriptor to register. 107   @param fd The file descriptor to register.
108   @param desc Pointer to descriptor state for this fd. 108   @param desc Pointer to descriptor state for this fd.
109   109  
110   @return The error if the fd cannot be tracked, otherwise a 110   @return The error if the fd cannot be tracked, otherwise a
111   default constructed error code. 111   default constructed error code.
112   */ 112   */
113   std::error_code 113   std::error_code
114   register_descriptor(int fd, reactor_descriptor_state* desc) const; 114   register_descriptor(int fd, reactor_descriptor_state* desc) const;
115   115  
116   /// No-op: write readiness is watched from registration on. 116   /// No-op: write readiness is watched from registration on.
117   std::error_code 117   std::error_code
HITCBC 118   2716 ensure_write_registered(int, reactor_descriptor_state*) const noexcept 118   2703 ensure_write_registered(int, reactor_descriptor_state*) const noexcept
119   { 119   {
HITCBC 120   2716 return {}; 120   2703 return {};
121   } 121   }
122   122  
123   /** Deregister a persistently registered descriptor. 123   /** Deregister a persistently registered descriptor.
124   124  
125   @param fd The file descriptor to deregister. 125   @param fd The file descriptor to deregister.
126   */ 126   */
127   void deregister_descriptor(int fd) const; 127   void deregister_descriptor(int fd) const;
128   128  
129   /** Interrupt the reactor so it rebuilds its fd_sets. 129   /** Interrupt the reactor so it rebuilds its fd_sets.
130   130  
131   Called when a write, connect, or write-wait op is registered 131   Called when a write, connect, or write-wait op is registered
132   after the reactor's snapshot was taken. Without this, 132   after the reactor's snapshot was taken. Without this,
133   select() may block not watching for writability on the fd. 133   select() may block not watching for writability on the fd.
134   */ 134   */
135   void notify_reactor() const; 135   void notify_reactor() const;
136   136  
137   /// Watch the read end of the POSIX signal self-pipe (see scheduler.hpp). 137   /// Watch the read end of the POSIX signal self-pipe (see scheduler.hpp).
HITCBC 138   272 [[nodiscard]] std::error_code register_signal_reader(int read_fd) override 138   272 [[nodiscard]] std::error_code register_signal_reader(int read_fd) override
139   { 139   {
HITCBC 140   272 return register_descriptor(read_fd, signal_pipe_reader_.arm()); 140   272 return register_descriptor(read_fd, signal_pipe_reader_.arm());
141   } 141   }
142   142  
143   private: 143   private:
144   void run_task(lock_type& lock, context_type& ctx, long timeout_us) override; 144   void run_task(lock_type& lock, context_type& ctx, long timeout_us) override;
145   void interrupt_reactor() const override; 145   void interrupt_reactor() const override;
146   146  
147   /// Rebuild the poll set a blocked select() was given. 147   /// Rebuild the poll set a blocked select() was given.
HITCBC 148   475 void notify_reparked() const override 148   416 void notify_reparked() const override
149   { 149   {
HITCBC 150   475 interrupt_reactor(); 150   416 interrupt_reactor();
HITCBC 151   475 } 151   416 }
152   long calculate_timeout(long requested_timeout_us) const; 152   long calculate_timeout(long requested_timeout_us) const;
153   153  
154   // Watches the global signal self-pipe's read end (armed lazily by 154   // Watches the global signal self-pipe's read end (armed lazily by
155   // register_signal_reader on the first signal registration). 155   // register_signal_reader on the first signal registration).
156   reactor_signal_pipe_reader signal_pipe_reader_; 156   reactor_signal_pipe_reader signal_pipe_reader_;
157   157  
158   // Self-pipe for interrupting select() 158   // Self-pipe for interrupting select()
159   int pipe_fds_[2]; // [0]=read, [1]=write 159   int pipe_fds_[2]; // [0]=read, [1]=write
160   160  
161   // Per-fd tracking for fd_set building 161   // Per-fd tracking for fd_set building
162   mutable std::unordered_map<int, reactor_descriptor_state*> 162   mutable std::unordered_map<int, reactor_descriptor_state*>
163   registered_descs_; 163   registered_descs_;
164   mutable int max_fd_ = -1; 164   mutable int max_fd_ = -1;
165   }; 165   };
166   166  
HITCBC 167   1953 inline select_scheduler::select_scheduler(capy::execution_context& ctx, int) 167   1953 inline select_scheduler::select_scheduler(capy::execution_context& ctx, int)
HITCBC 168   1953 : pipe_fds_{-1, -1} 168   1953 : pipe_fds_{-1, -1}
HITCBC 169   1953 , max_fd_(-1) 169   1953 , max_fd_(-1)
170   { 170   {
HITCBC 171   1953 if (::pipe(pipe_fds_) < 0) 171   1953 if (::pipe(pipe_fds_) < 0)
HITCBC 172   1 detail::throw_system_error(make_err(errno), "pipe"); 172   1 detail::throw_system_error(make_err(errno), "pipe");
173   173  
HITCBC 174   5847 for (int i = 0; i < 2; ++i) 174   5847 for (int i = 0; i < 2; ++i)
175   { 175   {
HITCBC 176   3901 int flags = ::fcntl(pipe_fds_[i], F_GETFL, 0); 176   3901 int flags = ::fcntl(pipe_fds_[i], F_GETFL, 0);
HITCBC 177   3901 if (flags == -1) 177   3901 if (flags == -1)
178   { 178   {
HITCBC 179   2 int errn = errno; 179   2 int errn = errno;
HITCBC 180   2 ::close(pipe_fds_[0]); 180   2 ::close(pipe_fds_[0]);
HITCBC 181   2 ::close(pipe_fds_[1]); 181   2 ::close(pipe_fds_[1]);
HITCBC 182   2 detail::throw_system_error(make_err(errn), "fcntl F_GETFL"); 182   2 detail::throw_system_error(make_err(errn), "fcntl F_GETFL");
183   } 183   }
HITCBC 184   3899 if (::fcntl(pipe_fds_[i], F_SETFL, flags | O_NONBLOCK) == -1) 184   3899 if (::fcntl(pipe_fds_[i], F_SETFL, flags | O_NONBLOCK) == -1)
185   { 185   {
HITCBC 186   2 int errn = errno; 186   2 int errn = errno;
HITCBC 187   2 ::close(pipe_fds_[0]); 187   2 ::close(pipe_fds_[0]);
HITCBC 188   2 ::close(pipe_fds_[1]); 188   2 ::close(pipe_fds_[1]);
HITCBC 189   2 detail::throw_system_error(make_err(errn), "fcntl F_SETFL"); 189   2 detail::throw_system_error(make_err(errn), "fcntl F_SETFL");
190   } 190   }
HITCBC 191   3897 if (::fcntl(pipe_fds_[i], F_SETFD, FD_CLOEXEC) == -1) 191   3897 if (::fcntl(pipe_fds_[i], F_SETFD, FD_CLOEXEC) == -1)
192   { 192   {
HITCBC 193   2 int errn = errno; 193   2 int errn = errno;
HITCBC 194   2 ::close(pipe_fds_[0]); 194   2 ::close(pipe_fds_[0]);
HITCBC 195   2 ::close(pipe_fds_[1]); 195   2 ::close(pipe_fds_[1]);
HITCBC 196   2 detail::throw_system_error(make_err(errn), "fcntl F_SETFD"); 196   2 detail::throw_system_error(make_err(errn), "fcntl F_SETFD");
197   } 197   }
198   } 198   }
199   199  
HITCBC 200   1946 timer_svc_ = &get_timer_service(ctx, *this); 200   1946 timer_svc_ = &get_timer_service(ctx, *this);
HITCBC 201   1946 timer_svc_->set_on_earliest_changed( 201   1946 timer_svc_->set_on_earliest_changed(
HITCBC 202   5034 timer_service::callback(this, [](void* p) { 202   5045 timer_service::callback(this, [](void* p) {
HITCBC 203   3088 static_cast<select_scheduler*>(p)->interrupt_reactor(); 203   3099 static_cast<select_scheduler*>(p)->interrupt_reactor();
HITCBC 204   3088 })); 204   3099 }));
205   205  
HITCBC 206   1946 completed_ops_.push(&task_op_); 206   1946 completed_ops_.push(&task_op_);
HITCBC 207   1967 } 207   1967 }
208   208  
HITCBC 209   3892 inline select_scheduler::~select_scheduler() 209   3892 inline select_scheduler::~select_scheduler()
210   { 210   {
HITCBC 211   1946 if (pipe_fds_[0] >= 0) 211   1946 if (pipe_fds_[0] >= 0)
HITCBC 212   1946 ::close(pipe_fds_[0]); 212   1946 ::close(pipe_fds_[0]);
HITCBC 213   1946 if (pipe_fds_[1] >= 0) 213   1946 if (pipe_fds_[1] >= 0)
HITCBC 214   1946 ::close(pipe_fds_[1]); 214   1946 ::close(pipe_fds_[1]);
HITCBC 215   3892 } 215   3892 }
216   216  
217   inline void 217   inline void
HITCBC 218   1946 select_scheduler::shutdown() 218   1946 select_scheduler::shutdown()
219   { 219   {
HITCBC 220   1946 shutdown_drain(); 220   1946 shutdown_drain();
221   221  
HITCBC 222   1946 if (pipe_fds_[1] >= 0) 222   1946 if (pipe_fds_[1] >= 0)
HITCBC 223   1946 interrupt_reactor(); 223   1946 interrupt_reactor();
HITCBC 224   1946 } 224   1946 }
225   225  
226   inline std::error_code 226   inline std::error_code
HITCBC 227   9973 select_scheduler::register_descriptor( 227   9892 select_scheduler::register_descriptor(
228   int fd, reactor_descriptor_state* desc) const 228   int fd, reactor_descriptor_state* desc) const
229   { 229   {
HITCBC 230   9973 if (fd < 0 || fd >= FD_SETSIZE) 230   9892 if (fd < 0 || fd >= FD_SETSIZE)
HITCBC 231   1 return make_err(EMFILE); 231   1 return make_err(EMFILE);
232   232  
233   // A completion queued by an earlier registration of the same state 233   // A completion queued by an earlier registration of the same state
234   // may still be running, so the fields it reads change under its 234   // may still be running, so the fields it reads change under its
235   // mutex, and the mutex mode is only written when it differs. 235   // mutex, and the mutex mode is only written when it differs.
HITCBC 236   9972 if (desc->mutex.enabled() != reactor_io_locking_) 236   9891 if (desc->mutex.enabled() != reactor_io_locking_)
HITCBC 237   3 desc->mutex.set_enabled(reactor_io_locking_); 237   3 desc->mutex.set_enabled(reactor_io_locking_);
238   { 238   {
HITCBC 239   9972 conditionally_enabled_mutex::scoped_lock lock(desc->mutex); 239   9891 conditionally_enabled_mutex::scoped_lock lock(desc->mutex);
HITCBC 240   9972 desc->registered_events = reactor_event_read | reactor_event_write; 240   9891 desc->registered_events = reactor_event_read | reactor_event_write;
HITCBC 241   9972 desc->unpollable = false; // the state is reused across adoptions 241   9891 desc->unpollable = false; // the state is reused across adoptions
HITCBC 242   9972 desc->fd = fd; 242   9891 desc->fd = fd;
HITCBC 243   9972 desc->scheduler_ = this; 243   9891 desc->scheduler_ = this;
HITCBC 244   9972 desc->ready_events_.store(0, std::memory_order_release); 244   9891 desc->ready_events_.store(0, std::memory_order_release);
245   // object_ref_ is not touched: a queued invocation from an 245   // object_ref_ is not touched: a queued invocation from an
246   // earlier registration still owns the reference in it. 246   // earlier registration still owns the reference in it.
HITCBC 247   9972 desc->read_ready = false; 247   9891 desc->read_ready = false;
HITCBC 248   9972 desc->write_ready = false; 248   9891 desc->write_ready = false;
HITCBC 249   9972 } 249   9891 }
250   250  
251   { 251   {
HITCBC 252   9972 mutex_type::scoped_lock lock(mutex_); 252   9891 mutex_type::scoped_lock lock(mutex_);
253   try 253   try
254   { 254   {
HITCBC 255   9972 registered_descs_[fd] = desc; 255   9891 registered_descs_[fd] = desc;
256   } 256   }
HITCBC 257   1 catch (std::bad_alloc const&) 257   1 catch (std::bad_alloc const&)
258   { 258   {
HITCBC 259   1 return make_err(ENOMEM); 259   1 return make_err(ENOMEM);
HITCBC 260   1 } 260   1 }
HITCBC 261   9971 if (fd > max_fd_) 261   9890 if (fd > max_fd_)
HITCBC 262   9382 max_fd_ = fd; 262   9357 max_fd_ = fd;
HITCBC 263   9972 } 263   9891 }
264   264  
HITCBC 265   9971 interrupt_reactor(); 265   9890 interrupt_reactor();
HITCBC 266   9971 return {}; 266   9890 return {};
267   } 267   }
268   268  
269   inline void 269   inline void
HITCBC 270   9700 select_scheduler::deregister_descriptor(int fd) const 270   9619 select_scheduler::deregister_descriptor(int fd) const
271   { 271   {
HITCBC 272   9700 mutex_type::scoped_lock lock(mutex_); 272   9619 mutex_type::scoped_lock lock(mutex_);
273   273  
HITCBC 274   9700 auto it = registered_descs_.find(fd); 274   9619 auto it = registered_descs_.find(fd);
HITCBC 275   9700 if (it == registered_descs_.end()) 275   9619 if (it == registered_descs_.end())
MISUBC 276   ✗ return; 276   ✗ return;
277   277  
HITCBC 278   9700 registered_descs_.erase(it); 278   9619 registered_descs_.erase(it);
279   279  
HITCBC 280   9700 if (fd == max_fd_) 280   9619 if (fd == max_fd_)
281   { 281   {
HITCBC 282   7658 max_fd_ = pipe_fds_[0]; 282   7647 max_fd_ = pipe_fds_[0];
HITCBC 283   12771 for (auto& [registered_fd, state] : registered_descs_) 283   12737 for (auto& [registered_fd, state] : registered_descs_)
284   { 284   {
HITCBC 285   5113 if (registered_fd > max_fd_) 285   5090 if (registered_fd > max_fd_)
HITCBC 286   4866 max_fd_ = registered_fd; 286   4826 max_fd_ = registered_fd;
287   } 287   }
288   } 288   }
HITCBC 289   9700 } 289   9619 }
290   290  
291   inline void 291   inline void
HITCBC 292   6716 select_scheduler::notify_reactor() const 292   6715 select_scheduler::notify_reactor() const
293   { 293   {
HITCBC 294   6716 interrupt_reactor(); 294   6715 interrupt_reactor();
HITCBC 295   6716 } 295   6715 }
296   296  
297   inline void 297   inline void
HITCBC 298   25428 select_scheduler::interrupt_reactor() const 298   25192 select_scheduler::interrupt_reactor() const
299   { 299   {
HITCBC 300   25428 char byte = 1; 300   25192 char byte = 1;
HITCBC 301   25428 [[maybe_unused]] auto r = ::write(pipe_fds_[1], &byte, 1); 301   25192 [[maybe_unused]] auto r = ::write(pipe_fds_[1], &byte, 1);
HITCBC 302   25428 } 302   25192 }
303   303  
304   inline long 304   inline long
HITCBC 305   19611 select_scheduler::calculate_timeout(long requested_timeout_us) const 305   14571 select_scheduler::calculate_timeout(long requested_timeout_us) const
306   { 306   {
HITCBC 307   19611 if (requested_timeout_us == 0) 307   14571 if (requested_timeout_us == 0)
308   return 0; // LCOV_EXCL_LINE run_task passes 0 via task_interrupted_, never through this argument 308   return 0; // LCOV_EXCL_LINE run_task passes 0 via task_interrupted_, never through this argument
309   309  
HITCBC 310   19611 auto nearest = timer_svc_->nearest_expiry(); 310   14571 auto nearest = timer_svc_->nearest_expiry();
HITCBC 311   19611 if (nearest == timer_service::time_point::max()) 311   14571 if (nearest == timer_service::time_point::max())
HITCBC 312   13161 return requested_timeout_us; 312   8179 return requested_timeout_us;
313   313  
HITCBC 314   6450 auto now = std::chrono::steady_clock::now(); 314   6392 auto now = std::chrono::steady_clock::now();
HITCBC 315   6450 if (nearest <= now) 315   6392 if (nearest <= now)
HITCBC 316   63 return 0; 316   74 return 0;
317   317  
318   auto timer_timeout_us = 318   auto timer_timeout_us =
HITCBC 319   6387 std::chrono::duration_cast<std::chrono::microseconds>(nearest - now) 319   6318 std::chrono::duration_cast<std::chrono::microseconds>(nearest - now)
HITCBC 320   6387 .count(); 320   6318 .count();
321   321  
HITCBC 322   6387 constexpr auto long_max = 322   6318 constexpr auto long_max =
323   static_cast<long long>((std::numeric_limits<long>::max)()); 323   static_cast<long long>((std::numeric_limits<long>::max)());
324   auto capped_timer_us = 324   auto capped_timer_us =
HITCBC 325   6387 (std::min)((std::max)(static_cast<long long>(timer_timeout_us), 325   6318 (std::min)((std::max)(static_cast<long long>(timer_timeout_us),
HITCBC 326   6387 static_cast<long long>(0)), 326   6318 static_cast<long long>(0)),
HITCBC 327   6387 long_max); 327   6318 long_max);
328   328  
HITCBC 329   6387 if (requested_timeout_us < 0) 329   6318 if (requested_timeout_us < 0)
HITCBC 330   6381 return static_cast<long>(capped_timer_us); 330   6312 return static_cast<long>(capped_timer_us);
331   331  
332   return static_cast<long>( 332   return static_cast<long>(
HITCBC 333   6 (std::min)(static_cast<long long>(requested_timeout_us), 333   6 (std::min)(static_cast<long long>(requested_timeout_us),
HITCBC 334   6 capped_timer_us)); 334   6 capped_timer_us));
335   } 335   }
336   336  
337   inline void 337   inline void
HITCBC 338   148949 select_scheduler::run_task(lock_type& lock, context_type& ctx, long timeout_us) 338   145804 select_scheduler::run_task(lock_type& lock, context_type& ctx, long timeout_us)
339   { 339   {
340   long effective_timeout_us = 340   long effective_timeout_us =
HITCBC 341   148949 task_interrupted_ ? 0 : calculate_timeout(timeout_us); 341   145804 task_interrupted_ ? 0 : calculate_timeout(timeout_us);
342   342  
343   // Snapshot registered descriptors while holding lock. 343   // Snapshot registered descriptors while holding lock.
344   // Record which directions each fd needs monitored to avoid a hot 344   // Record which directions each fd needs monitored to avoid a hot
345   // loop: select is level-triggered, so a writable socket (nearly 345   // loop: select is level-triggered, so a writable socket (nearly
346   // always writable) or an always-readable fd (/dev/zero, a pipe at 346   // always writable) or an always-readable fd (/dev/zero, a pipe at
347   // EOF) would return select() immediately every iteration if 347   // EOF) would return select() immediately every iteration if
348   // unconditionally added. Membership in both sets is opt-in: a 348   // unconditionally added. Membership in both sets is opt-in: a
349   // parked op or wait in a direction opts that direction in. The 349   // parked op or wait in a direction opts that direction in. The
350   // exceptional set is opt-in too, for any parked op or wait: 350   // exceptional set is opt-in too, for any parked op or wait:
351   // Darwin reports a character device (/dev/zero, /dev/null) as 351   // Darwin reports a character device (/dev/zero, /dev/null) as
352   // exceptional on every call. 352   // exceptional on every call.
353   struct fd_entry 353   struct fd_entry
354   { 354   {
355   int fd; 355   int fd;
356   reactor_descriptor_state* desc; 356   reactor_descriptor_state* desc;
357   std::uint32_t want; 357   std::uint32_t want;
358   }; 358   };
359   fd_entry snapshot[FD_SETSIZE]; 359   fd_entry snapshot[FD_SETSIZE];
HITCBC 360   148949 int snapshot_count = 0; 360   145804 int snapshot_count = 0;
361   361  
362   // The caller may have released the lock to hand queued work to a 362   // The caller may have released the lock to hand queued work to a
363   // peer, and deregister_descriptor() erases from this map under it. 363   // peer, and deregister_descriptor() erases from this map under it.
HITCBC 364   148949 if (!lock.owns_lock()) 364   145804 if (!lock.owns_lock())
HITCBC 365   129335 lock.lock(); 365   131230 lock.lock();
366   366  
HITCBC 367   408543 for (auto& [fd, desc] : registered_descs_) 367   404250 for (auto& [fd, desc] : registered_descs_)
368   { 368   {
HITCBC 369   259594 if (snapshot_count < FD_SETSIZE) 369   258446 if (snapshot_count < FD_SETSIZE)
370   { 370   {
HITCBC 371   259594 conditionally_enabled_mutex::scoped_lock desc_lock(desc->mutex); 371   258446 conditionally_enabled_mutex::scoped_lock desc_lock(desc->mutex);
HITCBC 372   259594 snapshot[snapshot_count].fd = fd; 372   258446 snapshot[snapshot_count].fd = fd;
HITCBC 373   259594 snapshot[snapshot_count].desc = desc; 373   258446 snapshot[snapshot_count].desc = desc;
HITCBC 374   259594 snapshot[snapshot_count].want = 374   258446 snapshot[snapshot_count].want =
HITCBC 375   259594 ((desc->read_op || desc->wait_read_op) ? reactor_event_read 375   258446 ((desc->read_op || desc->wait_read_op) ? reactor_event_read
HITCBC 376   259594 : 0) | 376   258446 : 0) |
HITCBC 377   259214 ((desc->write_op || desc->connect_op || desc->wait_write_op) 377   258066 ((desc->write_op || desc->connect_op || desc->wait_write_op)
HITCBC 378   518808 ? reactor_event_write 378   516512 ? reactor_event_write
HITCBC 379   259594 : 0) | 379   258446 : 0) |
HITCBC 380   259594 (desc->wait_error_op ? reactor_event_error : 0); 380   258446 (desc->wait_error_op ? reactor_event_error : 0);
HITCBC 381   259594 ++snapshot_count; 381   258446 ++snapshot_count;
HITCBC 382   259594 } 382   258446 }
383   } 383   }
384   384  
HITCBC 385   148949 if (lock.owns_lock()) 385   145804 if (lock.owns_lock())
HITCBC 386   148948 lock.unlock(); 386   145803 lock.unlock();
387   387  
HITCBC 388   148949 task_cleanup on_exit{this, &lock, ctx}; 388   145804 task_cleanup on_exit{this, &lock, ctx};
389   389  
390   fd_set read_fds, write_fds, except_fds; 390   fd_set read_fds, write_fds, except_fds;
HITCBC 391   2532133 FD_ZERO(&read_fds); 391   2478668 FD_ZERO(&read_fds);
HITCBC 392   2532133 FD_ZERO(&write_fds); 392   2478668 FD_ZERO(&write_fds);
HITCBC 393   2532133 FD_ZERO(&except_fds); 393   2478668 FD_ZERO(&except_fds);
394   394  
HITCBC 395   148949 FD_SET(pipe_fds_[0], &read_fds); 395   145804 FD_SET(pipe_fds_[0], &read_fds);
HITCBC 396   148949 int nfds = pipe_fds_[0]; 396   145804 int nfds = pipe_fds_[0];
397   397  
HITCBC 398   408543 for (int i = 0; i < snapshot_count; ++i) 398   404250 for (int i = 0; i < snapshot_count; ++i)
399   { 399   {
HITCBC 400   259594 int fd = snapshot[i].fd; 400   258446 int fd = snapshot[i].fd;
HITCBC 401   259594 if (snapshot[i].want & reactor_event_read) 401   258446 if (snapshot[i].want & reactor_event_read)
HITCBC 402   120543 FD_SET(fd, &read_fds); 402   116251 FD_SET(fd, &read_fds);
HITCBC 403   259594 if (snapshot[i].want & reactor_event_write) 403   258446 if (snapshot[i].want & reactor_event_write)
HITCBC 404   2990 FD_SET(fd, &write_fds); 404   2977 FD_SET(fd, &write_fds);
HITCBC 405   259594 if (snapshot[i].want != 0) 405   258446 if (snapshot[i].want != 0)
HITCBC 406   123823 FD_SET(fd, &except_fds); 406   119518 FD_SET(fd, &except_fds);
HITCBC 407   259594 if (fd > nfds) 407   258446 if (fd > nfds)
HITCBC 408   147118 nfds = fd; 408   143903 nfds = fd;
409   } 409   }
410   410  
411   struct timeval tv; 411   struct timeval tv;
HITCBC 412   148949 struct timeval* tv_ptr = nullptr; 412   145804 struct timeval* tv_ptr = nullptr;
HITCBC 413   148949 if (effective_timeout_us >= 0) 413   145804 if (effective_timeout_us >= 0)
414   { 414   {
HITCBC 415   136430 tv.tv_sec = effective_timeout_us / 1000000; 415   138275 tv.tv_sec = effective_timeout_us / 1000000;
HITCBC 416   136430 tv.tv_usec = effective_timeout_us % 1000000; 416   138275 tv.tv_usec = effective_timeout_us % 1000000;
HITCBC 417   136430 tv_ptr = &tv; 417   138275 tv_ptr = &tv;
418   } 418   }
419   419  
HITCBC 420   148949 int ready = ::select(nfds + 1, &read_fds, &write_fds, &except_fds, tv_ptr); 420   145804 int ready = ::select(nfds + 1, &read_fds, &write_fds, &except_fds, tv_ptr);
421   421  
422   // EINTR: signal interrupted select(), just retry. 422   // EINTR: signal interrupted select(), just retry.
423   // EBADF: an fd was closed between snapshot and select(); retry 423   // EBADF: an fd was closed between snapshot and select(); retry
424   // with a fresh snapshot from registered_descs_. 424   // with a fresh snapshot from registered_descs_.
425   // Both fall through with no ready descriptors rather than 425   // Both fall through with no ready descriptors rather than
426   // returning: the caller handed this function an owned lock that 426   // returning: the caller handed this function an owned lock that
427   // only the epilogue below re-acquires. 427   // only the epilogue below re-acquires.
HITCBC 428   148949 if (ready < 0) 428   145804 if (ready < 0)
429   { 429   {
HITCBC 430   3 if (errno != EINTR && errno != EBADF) 430   3 if (errno != EINTR && errno != EBADF)
HITCBC 431   1 detail::throw_system_error(make_err(errno), "select"); 431   1 detail::throw_system_error(make_err(errno), "select");
HITCBC 432   2 ready = 0; 432   2 ready = 0;
433   } 433   }
434   434  
435   // Process timers outside the lock 435   // Process timers outside the lock
HITCBC 436   148948 timer_svc_->process_expired(); 436   145803 timer_svc_->process_expired();
437   437  
HITCBC 438   148948 ready_queue local_ops; 438   145803 ready_queue local_ops;
439   439  
HITCBC 440   148948 if (ready > 0) 440   145803 if (ready > 0)
441   { 441   {
HITCBC 442   19933 if (FD_ISSET(pipe_fds_[0], &read_fds)) 442   14964 if (FD_ISSET(pipe_fds_[0], &read_fds))
443   { 443   {
444   char buf[256]; 444   char buf[256];
HITCBC 445   22633 while (::read(pipe_fds_[0], buf, sizeof(buf)) > 0) 445   22240 while (::read(pipe_fds_[0], buf, sizeof(buf)) > 0)
446   { 446   {
447   } 447   }
448   } 448   }
449   449  
HITCBC 450   108550 for (int i = 0; i < snapshot_count; ++i) 450   99058 for (int i = 0; i < snapshot_count; ++i)
451   { 451   {
HITCBC 452   88617 int fd = snapshot[i].fd; 452   84094 int fd = snapshot[i].fd;
HITCBC 453   88617 reactor_descriptor_state* desc = snapshot[i].desc; 453   84094 reactor_descriptor_state* desc = snapshot[i].desc;
454   454  
HITCBC 455   88617 std::uint32_t flags = 0; 455   84094 std::uint32_t flags = 0;
HITCBC 456   88617 if (FD_ISSET(fd, &read_fds)) 456   84094 if (FD_ISSET(fd, &read_fds))
HITCBC 457   11377 flags |= reactor_event_read; 457   6618 flags |= reactor_event_read;
HITCBC 458   88617 if (FD_ISSET(fd, &write_fds)) 458   84094 if (FD_ISSET(fd, &write_fds))
HITCBC 459   2696 flags |= reactor_event_write; 459   2683 flags |= reactor_event_write;
HITCBC 460   88617 if (FD_ISSET(fd, &except_fds)) 460   84094 if (FD_ISSET(fd, &except_fds))
HITCBC 461   11 flags |= reactor_event_error; 461   8 flags |= reactor_event_error;
462   462  
HITCBC 463   88617 if (flags == 0) 463   84094 if (flags == 0)
HITCBC 464   74542 continue; 464   74791 continue;
465   465  
HITCBC 466   14075 desc->add_ready_events(flags); 466   9303 desc->add_ready_events(flags);
467   467  
HITCBC 468   14075 bool expected = false; 468   9303 bool expected = false;
HITCBC 469   14075 if (desc->is_enqueued_.compare_exchange_strong( 469   9303 if (desc->is_enqueued_.compare_exchange_strong(
470   expected, true, std::memory_order_release, 470   expected, true, std::memory_order_release,
471   std::memory_order_relaxed)) 471   std::memory_order_relaxed))
472   { 472   {
HITCBC 473   6233 local_ops.push(desc); 473   6175 local_ops.push(desc);
474   } 474   }
475   } 475   }
476   } 476   }
477   477  
HITCBC 478   148948 lock.lock(); 478   145803 lock.lock();
479   479  
HITCBC 480   148948 completed_ops_.splice(local_ops); 480   145803 completed_ops_.splice(local_ops);
HITCBC 481   148949 } 481   145804 }
482   482  
483   } // namespace boost::corosio::detail 483   } // namespace boost::corosio::detail
484   484  
485   #endif // BOOST_COROSIO_HAS_SELECT 485   #endif // BOOST_COROSIO_HAS_SELECT
486   486  
487   #endif // BOOST_COROSIO_NATIVE_DETAIL_SELECT_SELECT_SCHEDULER_HPP 487   #endif // BOOST_COROSIO_NATIVE_DETAIL_SELECT_SELECT_SCHEDULER_HPP