. /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).
HITCBC 135   61 [[nodiscard]] std::error_code register_signal_reader(int read_fd) override 135   61 [[nodiscard]] std::error_code register_signal_reader(int read_fd) override
136   { 136   {
HITCBC 137   61 return register_descriptor(read_fd, signal_pipe_reader_.arm()); 137   61 return register_descriptor(read_fd, signal_pipe_reader_.arm());
138   } 138   }
139   139  
140   private: 140   private:
141   void run_task(lock_type& lock, context_type& ctx, long timeout_us) override; 141   void run_task(lock_type& lock, context_type& ctx, long timeout_us) override;
142   void interrupt_reactor() const override; 142   void interrupt_reactor() const override;
143   long calculate_timeout(long requested_timeout_us) const; 143   long calculate_timeout(long requested_timeout_us) const;
144   144  
145   // 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
146   // register_signal_reader on the first signal registration). 146   // register_signal_reader on the first signal registration).
147   reactor_signal_pipe_reader signal_pipe_reader_; 147   reactor_signal_pipe_reader signal_pipe_reader_;
148   148  
149   // Self-pipe for interrupting select() 149   // Self-pipe for interrupting select()
150   int pipe_fds_[2]; // [0]=read, [1]=write 150   int pipe_fds_[2]; // [0]=read, [1]=write
151   151  
152   // Per-fd tracking for fd_set building 152   // Per-fd tracking for fd_set building
153   mutable std::unordered_map<int, reactor_descriptor_state*> 153   mutable std::unordered_map<int, reactor_descriptor_state*>
154   registered_descs_; 154   registered_descs_;
155   mutable int max_fd_ = -1; 155   mutable int max_fd_ = -1;
156   }; 156   };
157   157  
HITCBC 158   950 inline select_scheduler::select_scheduler(capy::execution_context& ctx, int) 158   950 inline select_scheduler::select_scheduler(capy::execution_context& ctx, int)
HITCBC 159   950 : pipe_fds_{-1, -1} 159   950 : pipe_fds_{-1, -1}
HITCBC 160   950 , max_fd_(-1) 160   950 , max_fd_(-1)
161   { 161   {
HITCBC 162   950 if (::pipe(pipe_fds_) < 0) 162   950 if (::pipe(pipe_fds_) < 0)
HITCBC 163   1 detail::throw_system_error(make_err(errno), "pipe"); 163   1 detail::throw_system_error(make_err(errno), "pipe");
164   164  
HITCBC 165   2838 for (int i = 0; i < 2; ++i) 165   2838 for (int i = 0; i < 2; ++i)
166   { 166   {
HITCBC 167   1895 int flags = ::fcntl(pipe_fds_[i], F_GETFL, 0); 167   1895 int flags = ::fcntl(pipe_fds_[i], F_GETFL, 0);
HITCBC 168   1895 if (flags == -1) 168   1895 if (flags == -1)
169   { 169   {
HITCBC 170   2 int errn = errno; 170   2 int errn = errno;
HITCBC 171   2 ::close(pipe_fds_[0]); 171   2 ::close(pipe_fds_[0]);
HITCBC 172   2 ::close(pipe_fds_[1]); 172   2 ::close(pipe_fds_[1]);
HITCBC 173   2 detail::throw_system_error(make_err(errn), "fcntl F_GETFL"); 173   2 detail::throw_system_error(make_err(errn), "fcntl F_GETFL");
174   } 174   }
HITCBC 175   1893 if (::fcntl(pipe_fds_[i], F_SETFL, flags | O_NONBLOCK) == -1) 175   1893 if (::fcntl(pipe_fds_[i], F_SETFL, flags | O_NONBLOCK) == -1)
176   { 176   {
HITCBC 177   2 int errn = errno; 177   2 int errn = errno;
HITCBC 178   2 ::close(pipe_fds_[0]); 178   2 ::close(pipe_fds_[0]);
HITCBC 179   2 ::close(pipe_fds_[1]); 179   2 ::close(pipe_fds_[1]);
HITCBC 180   2 detail::throw_system_error(make_err(errn), "fcntl F_SETFL"); 180   2 detail::throw_system_error(make_err(errn), "fcntl F_SETFL");
181   } 181   }
HITCBC 182   1891 if (::fcntl(pipe_fds_[i], F_SETFD, FD_CLOEXEC) == -1) 182   1891 if (::fcntl(pipe_fds_[i], F_SETFD, FD_CLOEXEC) == -1)
183   { 183   {
HITCBC 184   2 int errn = errno; 184   2 int errn = errno;
HITCBC 185   2 ::close(pipe_fds_[0]); 185   2 ::close(pipe_fds_[0]);
HITCBC 186   2 ::close(pipe_fds_[1]); 186   2 ::close(pipe_fds_[1]);
HITCBC 187   2 detail::throw_system_error(make_err(errn), "fcntl F_SETFD"); 187   2 detail::throw_system_error(make_err(errn), "fcntl F_SETFD");
188   } 188   }
189   } 189   }
190   190  
HITCBC 191   943 timer_svc_ = &get_timer_service(ctx, *this); 191   943 timer_svc_ = &get_timer_service(ctx, *this);
HITCBC 192   943 timer_svc_->set_on_earliest_changed( 192   943 timer_svc_->set_on_earliest_changed(
HITCBC 193   3716 timer_service::callback(this, [](void* p) { 193   3688 timer_service::callback(this, [](void* p) {
HITCBC 194   2773 static_cast<select_scheduler*>(p)->interrupt_reactor(); 194   2745 static_cast<select_scheduler*>(p)->interrupt_reactor();
HITCBC 195   2773 })); 195   2745 }));
196   196  
HITCBC 197   943 get_resolver_service(ctx, *this); 197   943 get_resolver_service(ctx, *this);
HITCBC 198   943 get_signal_service(ctx, *this); 198   943 get_signal_service(ctx, *this);
HITCBC 199   943 get_stream_file_service(ctx, *this); 199   943 get_stream_file_service(ctx, *this);
HITCBC 200   943 get_random_access_file_service(ctx, *this); 200   943 get_random_access_file_service(ctx, *this);
201   201  
HITCBC 202   943 completed_ops_.push(&task_op_); 202   943 completed_ops_.push(&task_op_);
HITCBC 203   964 } 203   964 }
204   204  
HITCBC 205   1886 inline select_scheduler::~select_scheduler() 205   1886 inline select_scheduler::~select_scheduler()
206   { 206   {
HITCBC 207   943 if (pipe_fds_[0] >= 0) 207   943 if (pipe_fds_[0] >= 0)
HITCBC 208   943 ::close(pipe_fds_[0]); 208   943 ::close(pipe_fds_[0]);
HITCBC 209   943 if (pipe_fds_[1] >= 0) 209   943 if (pipe_fds_[1] >= 0)
HITCBC 210   943 ::close(pipe_fds_[1]); 210   943 ::close(pipe_fds_[1]);
HITCBC 211   1886 } 211   1886 }
212   212  
213   inline void 213   inline void
HITCBC 214   943 select_scheduler::shutdown() 214   943 select_scheduler::shutdown()
215   { 215   {
HITCBC 216   943 shutdown_drain(); 216   943 shutdown_drain();
217   217  
HITCBC 218   943 if (pipe_fds_[1] >= 0) 218   943 if (pipe_fds_[1] >= 0)
HITCBC 219   943 interrupt_reactor(); 219   943 interrupt_reactor();
HITCBC 220   943 } 220   943 }
221   221  
222   inline std::error_code 222   inline std::error_code
HITCBC 223   4874 select_scheduler::register_descriptor( 223   4916 select_scheduler::register_descriptor(
224   int fd, reactor_descriptor_state* desc) const 224   int fd, reactor_descriptor_state* desc) const
225   { 225   {
HITCBC 226   4874 if (fd < 0 || fd >= FD_SETSIZE) 226   4916 if (fd < 0 || fd >= FD_SETSIZE)
HITCBC 227   1 return make_err(EMFILE); 227   1 return make_err(EMFILE);
228   228  
HITCBC 229   4873 desc->registered_events = reactor_event_read | reactor_event_write; 229   4915 desc->registered_events = reactor_event_read | reactor_event_write;
HITCBC 230   4873 desc->fd = fd; 230   4915 desc->fd = fd;
HITCBC 231   4873 desc->scheduler_ = this; 231   4915 desc->scheduler_ = this;
HITCBC 232   4873 desc->mutex.set_enabled(reactor_io_locking_); 232   4915 desc->mutex.set_enabled(reactor_io_locking_);
HITCBC 233   4873 desc->ready_events_.store(0, std::memory_order_relaxed); 233   4915 desc->ready_events_.store(0, std::memory_order_relaxed);
234   234  
235   { 235   {
HITCBC 236   4873 conditionally_enabled_mutex::scoped_lock lock(desc->mutex); 236   4915 conditionally_enabled_mutex::scoped_lock lock(desc->mutex);
HITCBC 237   4873 desc->impl_ref_.reset(); 237   4915 desc->impl_ref_.reset();
HITCBC 238   4873 desc->read_ready = false; 238   4915 desc->read_ready = false;
HITCBC 239   4873 desc->write_ready = false; 239   4915 desc->write_ready = false;
HITCBC 240   4873 } 240   4915 }
241   241  
242   { 242   {
HITCBC 243   4873 mutex_type::scoped_lock lock(mutex_); 243   4915 mutex_type::scoped_lock lock(mutex_);
244   try 244   try
245   { 245   {
HITCBC 246   4873 registered_descs_[fd] = desc; 246   4915 registered_descs_[fd] = desc;
247   } 247   }
HITCBC 248   1 catch (std::bad_alloc const&) 248   1 catch (std::bad_alloc const&)
249   { 249   {
HITCBC 250   1 return make_err(ENOMEM); 250   1 return make_err(ENOMEM);
HITCBC 251   1 } 251   1 }
HITCBC 252   4872 if (fd > max_fd_) 252   4914 if (fd > max_fd_)
HITCBC 253   4818 max_fd_ = fd; 253   4860 max_fd_ = fd;
HITCBC 254   4873 } 254   4915 }
255   255  
HITCBC 256   4872 interrupt_reactor(); 256   4914 interrupt_reactor();
HITCBC 257   4872 return {}; 257   4914 return {};
258   } 258   }
259   259  
260   inline void 260   inline void
HITCBC 261   4812 select_scheduler::deregister_descriptor(int fd) const 261   4854 select_scheduler::deregister_descriptor(int fd) const
262   { 262   {
HITCBC 263   4812 mutex_type::scoped_lock lock(mutex_); 263   4854 mutex_type::scoped_lock lock(mutex_);
264   264  
HITCBC 265   4812 auto it = registered_descs_.find(fd); 265   4854 auto it = registered_descs_.find(fd);
HITCBC 266   4812 if (it == registered_descs_.end()) 266   4854 if (it == registered_descs_.end())
MISUBC 267   ✗ return; 267   ✗ return;
268   268  
HITCBC 269   4812 registered_descs_.erase(it); 269   4854 registered_descs_.erase(it);
270   270  
HITCBC 271   4812 if (fd == max_fd_) 271   4854 if (fd == max_fd_)
272   { 272   {
HITCBC 273   4466 max_fd_ = pipe_fds_[0]; 273   4507 max_fd_ = pipe_fds_[0];
HITCBC 274   8523 for (auto& [registered_fd, state] : registered_descs_) 274   8604 for (auto& [registered_fd, state] : registered_descs_)
275   { 275   {
HITCBC 276   4057 if (registered_fd > max_fd_) 276   4097 if (registered_fd > max_fd_)
HITCBC 277   3964 max_fd_ = registered_fd; 277   4004 max_fd_ = registered_fd;
278   } 278   }
279   } 279   }
HITCBC 280   4812 } 280   4854 }
281   281  
282   inline void 282   inline void
HITCBC 283   2161 select_scheduler::notify_reactor() const 283   2182 select_scheduler::notify_reactor() const
284   { 284   {
HITCBC 285   2161 interrupt_reactor(); 285   2182 interrupt_reactor();
HITCBC 286   2161 } 286   2182 }
287   287  
288   inline void 288   inline void
HITCBC 289   12391 select_scheduler::interrupt_reactor() const 289   12424 select_scheduler::interrupt_reactor() const
290   { 290   {
HITCBC 291   12391 char byte = 1; 291   12424 char byte = 1;
HITCBC 292   12391 [[maybe_unused]] auto r = ::write(pipe_fds_[1], &byte, 1); 292   12424 [[maybe_unused]] auto r = ::write(pipe_fds_[1], &byte, 1);
HITCBC 293   12391 } 293   12424 }
294   294  
295   inline long 295   inline long
HITCBC 296   293498 select_scheduler::calculate_timeout(long requested_timeout_us) const 296   231184 select_scheduler::calculate_timeout(long requested_timeout_us) const
297   { 297   {
HITCBC 298   293498 if (requested_timeout_us == 0) 298   231184 if (requested_timeout_us == 0)
299   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
300   300  
HITCBC 301   293498 auto nearest = timer_svc_->nearest_expiry(); 301   231184 auto nearest = timer_svc_->nearest_expiry();
HITCBC 302   293498 if (nearest == timer_service::time_point::max()) 302   231184 if (nearest == timer_service::time_point::max())
HITCBC 303   1089 return requested_timeout_us; 303   1071 return requested_timeout_us;
304   304  
HITCBC 305   292409 auto now = std::chrono::steady_clock::now(); 305   230113 auto now = std::chrono::steady_clock::now();
HITCBC 306   292409 if (nearest <= now) 306   230113 if (nearest <= now)
HITCBC 307   627 return 0; 307   935 return 0;
308   308  
309   auto timer_timeout_us = 309   auto timer_timeout_us =
HITCBC 310   291782 std::chrono::duration_cast<std::chrono::microseconds>(nearest - now) 310   229178 std::chrono::duration_cast<std::chrono::microseconds>(nearest - now)
HITCBC 311   291782 .count(); 311   229178 .count();
312   312  
HITCBC 313   291782 constexpr auto long_max = 313   229178 constexpr auto long_max =
314   static_cast<long long>((std::numeric_limits<long>::max)()); 314   static_cast<long long>((std::numeric_limits<long>::max)());
315   auto capped_timer_us = 315   auto capped_timer_us =
HITCBC 316   291782 (std::min)((std::max)(static_cast<long long>(timer_timeout_us), 316   229178 (std::min)((std::max)(static_cast<long long>(timer_timeout_us),
HITCBC 317   291782 static_cast<long long>(0)), 317   229178 static_cast<long long>(0)),
HITCBC 318   291782 long_max); 318   229178 long_max);
319   319  
HITCBC 320   291782 if (requested_timeout_us < 0) 320   229178 if (requested_timeout_us < 0)
HITCBC 321   291780 return static_cast<long>(capped_timer_us); 321   229176 return static_cast<long>(capped_timer_us);
322   322  
323   return static_cast<long>( 323   return static_cast<long>(
HITCBC 324   2 (std::min)(static_cast<long long>(requested_timeout_us), 324   2 (std::min)(static_cast<long long>(requested_timeout_us),
HITCBC 325   2 capped_timer_us)); 325   2 capped_timer_us));
326   } 326   }
327   327  
328   inline void 328   inline void
HITCBC 329   316174 select_scheduler::run_task(lock_type& lock, context_type& ctx, long timeout_us) 329   255764 select_scheduler::run_task(lock_type& lock, context_type& ctx, long timeout_us)
330   { 330   {
331   long effective_timeout_us = 331   long effective_timeout_us =
HITCBC 332   316174 task_interrupted_ ? 0 : calculate_timeout(timeout_us); 332   255764 task_interrupted_ ? 0 : calculate_timeout(timeout_us);
333   333  
334   // Snapshot registered descriptors while holding lock. 334   // Snapshot registered descriptors while holding lock.
335   // Record which fds need write monitoring to avoid a hot loop: 335   // Record which fds need write monitoring to avoid a hot loop:
336   // select is level-triggered so writable sockets (nearly always 336   // select is level-triggered so writable sockets (nearly always
337   // writable) would cause select() to return immediately every 337   // writable) would cause select() to return immediately every
338   // iteration if unconditionally added to write_fds. Membership 338   // iteration if unconditionally added to write_fds. Membership
339   // 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
340   // parked write or connect op does. 340   // parked write or connect op does.
341   struct fd_entry 341   struct fd_entry
342   { 342   {
343   int fd; 343   int fd;
344   reactor_descriptor_state* desc; 344   reactor_descriptor_state* desc;
345   bool needs_write; 345   bool needs_write;
346   }; 346   };
347   fd_entry snapshot[FD_SETSIZE]; 347   fd_entry snapshot[FD_SETSIZE];
HITCBC 348   316174 int snapshot_count = 0; 348   255764 int snapshot_count = 0;
349   349  
HITCBC 350   833330 for (auto& [fd, desc] : registered_descs_) 350   686418 for (auto& [fd, desc] : registered_descs_)
351   { 351   {
HITCBC 352   517156 if (snapshot_count < FD_SETSIZE) 352   430654 if (snapshot_count < FD_SETSIZE)
353   { 353   {
HITCBC 354   517156 conditionally_enabled_mutex::scoped_lock desc_lock(desc->mutex); 354   430654 conditionally_enabled_mutex::scoped_lock desc_lock(desc->mutex);
HITCBC 355   517156 snapshot[snapshot_count].fd = fd; 355   430654 snapshot[snapshot_count].fd = fd;
HITCBC 356   517156 snapshot[snapshot_count].desc = desc; 356   430654 snapshot[snapshot_count].desc = desc;
HITCBC 357   517156 snapshot[snapshot_count].needs_write = 357   430654 snapshot[snapshot_count].needs_write =
HITCBC 358   517156 (desc->write_op || desc->connect_op || desc->wait_write_op); 358   430654 (desc->write_op || desc->connect_op || desc->wait_write_op);
HITCBC 359   517156 ++snapshot_count; 359   430654 ++snapshot_count;
HITCBC 360   517156 } 360   430654 }
361   } 361   }
362   362  
HITCBC 363   316174 if (lock.owns_lock()) 363   255764 if (lock.owns_lock())
HITCBC 364   293499 lock.unlock(); 364   231185 lock.unlock();
365   365  
HITCBC 366   316174 task_cleanup on_exit{this, &lock, ctx}; 366   255764 task_cleanup on_exit{this, &lock, ctx};
367   367  
368   fd_set read_fds, write_fds, except_fds; 368   fd_set read_fds, write_fds, except_fds;
HITCBC 369   5374958 FD_ZERO(&read_fds); 369   4347988 FD_ZERO(&read_fds);
HITCBC 370   5374958 FD_ZERO(&write_fds); 370   4347988 FD_ZERO(&write_fds);
HITCBC 371   5374958 FD_ZERO(&except_fds); 371   4347988 FD_ZERO(&except_fds);
372   372  
HITCBC 373   316174 FD_SET(pipe_fds_[0], &read_fds); 373   255764 FD_SET(pipe_fds_[0], &read_fds);
HITCBC 374   316174 int nfds = pipe_fds_[0]; 374   255764 int nfds = pipe_fds_[0];
375   375  
HITCBC 376   833330 for (int i = 0; i < snapshot_count; ++i) 376   686418 for (int i = 0; i < snapshot_count; ++i)
377   { 377   {
HITCBC 378   517156 int fd = snapshot[i].fd; 378   430654 int fd = snapshot[i].fd;
HITCBC 379   517156 FD_SET(fd, &read_fds); 379   430654 FD_SET(fd, &read_fds);
HITCBC 380   517156 if (snapshot[i].needs_write) 380   430654 if (snapshot[i].needs_write)
HITCBC 381   11524 FD_SET(fd, &write_fds); 381   11763 FD_SET(fd, &write_fds);
HITCBC 382   517156 FD_SET(fd, &except_fds); 382   430654 FD_SET(fd, &except_fds);
HITCBC 383   517156 if (fd > nfds) 383   430654 if (fd > nfds)
HITCBC 384   315588 nfds = fd; 384   255161 nfds = fd;
385   } 385   }
386   386  
387   struct timeval tv; 387   struct timeval tv;
HITCBC 388   316174 struct timeval* tv_ptr = nullptr; 388   255764 struct timeval* tv_ptr = nullptr;
HITCBC 389   316174 if (effective_timeout_us >= 0) 389   255764 if (effective_timeout_us >= 0)
390   { 390   {
HITCBC 391   315395 tv.tv_sec = effective_timeout_us / 1000000; 391   254987 tv.tv_sec = effective_timeout_us / 1000000;
HITCBC 392   315395 tv.tv_usec = effective_timeout_us % 1000000; 392   254987 tv.tv_usec = effective_timeout_us % 1000000;
HITCBC 393   315395 tv_ptr = &tv; 393   254987 tv_ptr = &tv;
394   } 394   }
395   395  
HITCBC 396   316174 int ready = ::select(nfds + 1, &read_fds, &write_fds, &except_fds, tv_ptr); 396   255764 int ready = ::select(nfds + 1, &read_fds, &write_fds, &except_fds, tv_ptr);
397   397  
398   // EINTR: signal interrupted select(), just retry. 398   // EINTR: signal interrupted select(), just retry.
399   // EBADF: an fd was closed between snapshot and select(); retry 399   // EBADF: an fd was closed between snapshot and select(); retry
400   // with a fresh snapshot from registered_descs_. 400   // with a fresh snapshot from registered_descs_.
401   // Both fall through with no ready descriptors rather than 401   // Both fall through with no ready descriptors rather than
402   // returning: the caller handed this function an owned lock that 402   // returning: the caller handed this function an owned lock that
403   // only the epilogue below re-acquires. 403   // only the epilogue below re-acquires.
HITCBC 404   316174 if (ready < 0) 404   255764 if (ready < 0)
405   { 405   {
HITCBC 406   3 if (errno != EINTR && errno != EBADF) 406   3 if (errno != EINTR && errno != EBADF)
HITCBC 407   1 detail::throw_system_error(make_err(errno), "select"); 407   1 detail::throw_system_error(make_err(errno), "select");
HITCBC 408   2 ready = 0; 408   2 ready = 0;
409   } 409   }
410   410  
411   // Process timers outside the lock 411   // Process timers outside the lock
HITCBC 412   316173 timer_svc_->process_expired(); 412   255763 timer_svc_->process_expired();
413   413  
HITCBC 414   316173 ready_queue local_ops; 414   255763 ready_queue local_ops;
415   415  
HITCBC 416   316173 if (ready > 0) 416   255763 if (ready > 0)
417   { 417   {
HITCBC 418   300641 if (FD_ISSET(pipe_fds_[0], &read_fds)) 418   240053 if (FD_ISSET(pipe_fds_[0], &read_fds))
419   { 419   {
420   char buf[256]; 420   char buf[256];
HITCBC 421   11636 while (::read(pipe_fds_[0], buf, sizeof(buf)) > 0) 421   11714 while (::read(pipe_fds_[0], buf, sizeof(buf)) > 0)
422   { 422   {
423   } 423   }
424   } 424   }
425   425  
HITCBC 426   770135 for (int i = 0; i < snapshot_count; ++i) 426   623204 for (int i = 0; i < snapshot_count; ++i)
427   { 427   {
HITCBC 428   469494 int fd = snapshot[i].fd; 428   383151 int fd = snapshot[i].fd;
HITCBC 429   469494 reactor_descriptor_state* desc = snapshot[i].desc; 429   383151 reactor_descriptor_state* desc = snapshot[i].desc;
430   430  
HITCBC 431   469494 std::uint32_t flags = 0; 431   383151 std::uint32_t flags = 0;
HITCBC 432   469494 if (FD_ISSET(fd, &read_fds)) 432   383151 if (FD_ISSET(fd, &read_fds))
HITCBC 433   300996 flags |= reactor_event_read; 433   238832 flags |= reactor_event_read;
HITCBC 434   469494 if (FD_ISSET(fd, &write_fds)) 434   383151 if (FD_ISSET(fd, &write_fds))
HITCBC 435   2153 flags |= reactor_event_write; 435   2174 flags |= reactor_event_write;
HITCBC 436   469494 if (FD_ISSET(fd, &except_fds)) 436   383151 if (FD_ISSET(fd, &except_fds))
HITCBC 437   16 flags |= reactor_event_error; 437   16 flags |= reactor_event_error;
438   438  
HITCBC 439   469494 if (flags == 0) 439   383151 if (flags == 0)
HITCBC 440   166353 continue; 440   142154 continue;
441   441  
HITCBC 442   303141 desc->add_ready_events(flags); 442   240997 desc->add_ready_events(flags);
443   443  
HITCBC 444   303141 bool expected = false; 444   240997 bool expected = false;
HITCBC 445   303141 if (desc->is_enqueued_.compare_exchange_strong( 445   240997 if (desc->is_enqueued_.compare_exchange_strong(
446   expected, true, std::memory_order_release, 446   expected, true, std::memory_order_release,
447   std::memory_order_relaxed)) 447   std::memory_order_relaxed))
448   { 448   {
HITCBC 449   303141 local_ops.push(desc); 449   240997 local_ops.push(desc);
450   } 450   }
451   } 451   }
452   } 452   }
453   453  
HITCBC 454   316173 lock.lock(); 454   255763 lock.lock();
455   455  
HITCBC 456   316173 completed_ops_.splice(local_ops); 456   255763 completed_ops_.splice(local_ops);
HITCBC 457   316174 } 457   255764 }
458   458  
459   } // namespace boost::corosio::detail 459   } // namespace boost::corosio::detail
460   460  
461   #endif // BOOST_COROSIO_HAS_SELECT 461   #endif // BOOST_COROSIO_HAS_SELECT
462   462  
463   #endif // BOOST_COROSIO_NATIVE_DETAIL_SELECT_SELECT_SCHEDULER_HPP 463   #endif // BOOST_COROSIO_NATIVE_DETAIL_SELECT_SELECT_SCHEDULER_HPP