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