. /include/boost/corosio /native /detail /select /select_scheduler.hpp
99.43% Lines (173/174) 100.00% Functions (12/12)
[ ] 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   2200 ensure_write_registered(int, reactor_descriptor_state*) const noexcept 118   2176 ensure_write_registered(int, reactor_descriptor_state*) const noexcept
119   { 119   {
HITCBC 120   2200 return {}; 120   2176 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   61 [[nodiscard]] std::error_code register_signal_reader(int read_fd) override 138   61 [[nodiscard]] std::error_code register_signal_reader(int read_fd) override
139   { 139   {
HITCBC 140   61 return register_descriptor(read_fd, signal_pipe_reader_.arm()); 140   61 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   long calculate_timeout(long requested_timeout_us) const; 146   long calculate_timeout(long requested_timeout_us) const;
147   147  
148   // Watches the global signal self-pipe's read end (armed lazily by 148   // Watches the global signal self-pipe's read end (armed lazily by
149   // register_signal_reader on the first signal registration). 149   // register_signal_reader on the first signal registration).
150   reactor_signal_pipe_reader signal_pipe_reader_; 150   reactor_signal_pipe_reader signal_pipe_reader_;
151   151  
152   // Self-pipe for interrupting select() 152   // Self-pipe for interrupting select()
153   int pipe_fds_[2]; // [0]=read, [1]=write 153   int pipe_fds_[2]; // [0]=read, [1]=write
154   154  
155   // Per-fd tracking for fd_set building 155   // Per-fd tracking for fd_set building
156   mutable std::unordered_map<int, reactor_descriptor_state*> 156   mutable std::unordered_map<int, reactor_descriptor_state*>
157   registered_descs_; 157   registered_descs_;
158   mutable int max_fd_ = -1; 158   mutable int max_fd_ = -1;
159   }; 159   };
160   160  
HITCBC 161   1220 inline select_scheduler::select_scheduler(capy::execution_context& ctx, int) 161   1220 inline select_scheduler::select_scheduler(capy::execution_context& ctx, int)
HITCBC 162   1220 : pipe_fds_{-1, -1} 162   1220 : pipe_fds_{-1, -1}
HITCBC 163   1220 , max_fd_(-1) 163   1220 , max_fd_(-1)
164   { 164   {
HITCBC 165   1220 if (::pipe(pipe_fds_) < 0) 165   1220 if (::pipe(pipe_fds_) < 0)
HITCBC 166   1 detail::throw_system_error(make_err(errno), "pipe"); 166   1 detail::throw_system_error(make_err(errno), "pipe");
167   167  
HITCBC 168   3648 for (int i = 0; i < 2; ++i) 168   3648 for (int i = 0; i < 2; ++i)
169   { 169   {
HITCBC 170   2435 int flags = ::fcntl(pipe_fds_[i], F_GETFL, 0); 170   2435 int flags = ::fcntl(pipe_fds_[i], F_GETFL, 0);
HITCBC 171   2435 if (flags == -1) 171   2435 if (flags == -1)
172   { 172   {
HITCBC 173   2 int errn = errno; 173   2 int errn = errno;
HITCBC 174   2 ::close(pipe_fds_[0]); 174   2 ::close(pipe_fds_[0]);
HITCBC 175   2 ::close(pipe_fds_[1]); 175   2 ::close(pipe_fds_[1]);
HITCBC 176   2 detail::throw_system_error(make_err(errn), "fcntl F_GETFL"); 176   2 detail::throw_system_error(make_err(errn), "fcntl F_GETFL");
177   } 177   }
HITCBC 178   2433 if (::fcntl(pipe_fds_[i], F_SETFL, flags | O_NONBLOCK) == -1) 178   2433 if (::fcntl(pipe_fds_[i], F_SETFL, flags | O_NONBLOCK) == -1)
179   { 179   {
HITCBC 180   2 int errn = errno; 180   2 int errn = errno;
HITCBC 181   2 ::close(pipe_fds_[0]); 181   2 ::close(pipe_fds_[0]);
HITCBC 182   2 ::close(pipe_fds_[1]); 182   2 ::close(pipe_fds_[1]);
HITCBC 183   2 detail::throw_system_error(make_err(errn), "fcntl F_SETFL"); 183   2 detail::throw_system_error(make_err(errn), "fcntl F_SETFL");
184   } 184   }
HITCBC 185   2431 if (::fcntl(pipe_fds_[i], F_SETFD, FD_CLOEXEC) == -1) 185   2431 if (::fcntl(pipe_fds_[i], F_SETFD, FD_CLOEXEC) == -1)
186   { 186   {
HITCBC 187   2 int errn = errno; 187   2 int errn = errno;
HITCBC 188   2 ::close(pipe_fds_[0]); 188   2 ::close(pipe_fds_[0]);
HITCBC 189   2 ::close(pipe_fds_[1]); 189   2 ::close(pipe_fds_[1]);
HITCBC 190   2 detail::throw_system_error(make_err(errn), "fcntl F_SETFD"); 190   2 detail::throw_system_error(make_err(errn), "fcntl F_SETFD");
191   } 191   }
192   } 192   }
193   193  
HITCBC 194   1213 timer_svc_ = &get_timer_service(ctx, *this); 194   1213 timer_svc_ = &get_timer_service(ctx, *this);
HITCBC 195   1213 timer_svc_->set_on_earliest_changed( 195   1213 timer_svc_->set_on_earliest_changed(
HITCBC 196   4130 timer_service::callback(this, [](void* p) { 196   4008 timer_service::callback(this, [](void* p) {
HITCBC 197   2917 static_cast<select_scheduler*>(p)->interrupt_reactor(); 197   2795 static_cast<select_scheduler*>(p)->interrupt_reactor();
HITCBC 198   2917 })); 198   2795 }));
199   199  
HITCBC 200   1213 completed_ops_.push(&task_op_); 200   1213 completed_ops_.push(&task_op_);
HITCBC 201   1234 } 201   1234 }
202   202  
HITCBC 203   2426 inline select_scheduler::~select_scheduler() 203   2426 inline select_scheduler::~select_scheduler()
204   { 204   {
HITCBC 205   1213 if (pipe_fds_[0] >= 0) 205   1213 if (pipe_fds_[0] >= 0)
HITCBC 206   1213 ::close(pipe_fds_[0]); 206   1213 ::close(pipe_fds_[0]);
HITCBC 207   1213 if (pipe_fds_[1] >= 0) 207   1213 if (pipe_fds_[1] >= 0)
HITCBC 208   1213 ::close(pipe_fds_[1]); 208   1213 ::close(pipe_fds_[1]);
HITCBC 209   2426 } 209   2426 }
210   210  
211   inline void 211   inline void
HITCBC 212   1213 select_scheduler::shutdown() 212   1213 select_scheduler::shutdown()
213   { 213   {
HITCBC 214   1213 shutdown_drain(); 214   1213 shutdown_drain();
215   215  
HITCBC 216   1213 if (pipe_fds_[1] >= 0) 216   1213 if (pipe_fds_[1] >= 0)
HITCBC 217   1213 interrupt_reactor(); 217   1213 interrupt_reactor();
HITCBC 218   1213 } 218   1213 }
219   219  
220   inline std::error_code 220   inline std::error_code
HITCBC 221   4983 select_scheduler::register_descriptor( 221   4935 select_scheduler::register_descriptor(
222   int fd, reactor_descriptor_state* desc) const 222   int fd, reactor_descriptor_state* desc) const
223   { 223   {
HITCBC 224   4983 if (fd < 0 || fd >= FD_SETSIZE) 224   4935 if (fd < 0 || fd >= FD_SETSIZE)
HITCBC 225   1 return make_err(EMFILE); 225   1 return make_err(EMFILE);
226   226  
HITCBC 227   4982 desc->registered_events = reactor_event_read | reactor_event_write; 227   4934 desc->registered_events = reactor_event_read | reactor_event_write;
HITCBC 228   4982 desc->unpollable = false; // the state is reused across adoptions 228   4934 desc->unpollable = false; // the state is reused across adoptions
HITCBC 229   4982 desc->fd = fd; 229   4934 desc->fd = fd;
HITCBC 230   4982 desc->scheduler_ = this; 230   4934 desc->scheduler_ = this;
HITCBC 231   4982 desc->mutex.set_enabled(reactor_io_locking_); 231   4934 desc->mutex.set_enabled(reactor_io_locking_);
HITCBC 232   4982 desc->ready_events_.store(0, std::memory_order_relaxed); 232   4934 desc->ready_events_.store(0, std::memory_order_relaxed);
233   233  
234   { 234   {
HITCBC 235   4982 conditionally_enabled_mutex::scoped_lock lock(desc->mutex); 235   4934 conditionally_enabled_mutex::scoped_lock lock(desc->mutex);
HITCBC 236   4982 desc->impl_ref_.reset(); 236   4934 desc->impl_ref_.reset();
HITCBC 237   4982 desc->read_ready = false; 237   4934 desc->read_ready = false;
HITCBC 238   4982 desc->write_ready = false; 238   4934 desc->write_ready = false;
HITCBC 239   4982 } 239   4934 }
240   240  
241   { 241   {
HITCBC 242   4982 mutex_type::scoped_lock lock(mutex_); 242   4934 mutex_type::scoped_lock lock(mutex_);
243   try 243   try
244   { 244   {
HITCBC 245   4982 registered_descs_[fd] = desc; 245   4934 registered_descs_[fd] = desc;
246   } 246   }
HITCBC 247   1 catch (std::bad_alloc const&) 247   1 catch (std::bad_alloc const&)
248   { 248   {
HITCBC 249   1 return make_err(ENOMEM); 249   1 return make_err(ENOMEM);
HITCBC 250   1 } 250   1 }
HITCBC 251   4981 if (fd > max_fd_) 251   4933 if (fd > max_fd_)
HITCBC 252   4932 max_fd_ = fd; 252   4884 max_fd_ = fd;
HITCBC 253   4982 } 253   4934 }
254   254  
HITCBC 255   4981 interrupt_reactor(); 255   4933 interrupt_reactor();
HITCBC 256   4981 return {}; 256   4933 return {};
257   } 257   }
258   258  
259   inline void 259   inline void
HITCBC 260   4921 select_scheduler::deregister_descriptor(int fd) const 260   4873 select_scheduler::deregister_descriptor(int fd) const
261   { 261   {
HITCBC 262   4921 mutex_type::scoped_lock lock(mutex_); 262   4873 mutex_type::scoped_lock lock(mutex_);
263   263  
HITCBC 264   4921 auto it = registered_descs_.find(fd); 264   4873 auto it = registered_descs_.find(fd);
HITCBC 265   4921 if (it == registered_descs_.end()) 265   4873 if (it == registered_descs_.end())
MISUBC 266   ✗ return; 266   ✗ return;
267   267  
HITCBC 268   4921 registered_descs_.erase(it); 268   4873 registered_descs_.erase(it);
269   269  
HITCBC 270   4921 if (fd == max_fd_) 270   4873 if (fd == max_fd_)
271   { 271   {
HITCBC 272   4572 max_fd_ = pipe_fds_[0]; 272   4524 max_fd_ = pipe_fds_[0];
HITCBC 273   8692 for (auto& [registered_fd, state] : registered_descs_) 273   8596 for (auto& [registered_fd, state] : registered_descs_)
274   { 274   {
HITCBC 275   4120 if (registered_fd > max_fd_) 275   4072 if (registered_fd > max_fd_)
HITCBC 276   4022 max_fd_ = registered_fd; 276   3974 max_fd_ = registered_fd;
277   } 277   }
278   } 278   }
HITCBC 279   4921 } 279   4873 }
280   280  
281   inline void 281   inline void
HITCBC 282   4840 select_scheduler::notify_reactor() const 282   4792 select_scheduler::notify_reactor() const
283   { 283   {
HITCBC 284   4840 interrupt_reactor(); 284   4792 interrupt_reactor();
HITCBC 285   4840 } 285   4792 }
286   286  
287   inline void 287   inline void
HITCBC 288   16076 select_scheduler::interrupt_reactor() const 288   15956 select_scheduler::interrupt_reactor() const
289   { 289   {
HITCBC 290   16076 char byte = 1; 290   15956 char byte = 1;
HITCBC 291   16076 [[maybe_unused]] auto r = ::write(pipe_fds_[1], &byte, 1); 291   15956 [[maybe_unused]] auto r = ::write(pipe_fds_[1], &byte, 1);
HITCBC 292   16076 } 292   15956 }
293   293  
294   inline long 294   inline long
HITCBC 295   7399 select_scheduler::calculate_timeout(long requested_timeout_us) const 295   7382 select_scheduler::calculate_timeout(long requested_timeout_us) const
296   { 296   {
HITCBC 297   7399 if (requested_timeout_us == 0) 297   7382 if (requested_timeout_us == 0)
298   return 0; // LCOV_EXCL_LINE run_task passes 0 via task_interrupted_, never through this argument 298   return 0; // LCOV_EXCL_LINE run_task passes 0 via task_interrupted_, never through this argument
299   299  
HITCBC 300   7399 auto nearest = timer_svc_->nearest_expiry(); 300   7382 auto nearest = timer_svc_->nearest_expiry();
HITCBC 301   7399 if (nearest == timer_service::time_point::max()) 301   7382 if (nearest == timer_service::time_point::max())
HITCBC 302   1421 return requested_timeout_us; 302   1488 return requested_timeout_us;
303   303  
HITCBC 304   5978 auto now = std::chrono::steady_clock::now(); 304   5894 auto now = std::chrono::steady_clock::now();
HITCBC 305   5978 if (nearest <= now) 305   5894 if (nearest <= now)
HITCBC 306   513 return 0; 306   493 return 0;
307   307  
308   auto timer_timeout_us = 308   auto timer_timeout_us =
HITCBC 309   5465 std::chrono::duration_cast<std::chrono::microseconds>(nearest - now) 309   5401 std::chrono::duration_cast<std::chrono::microseconds>(nearest - now)
HITCBC 310   5465 .count(); 310   5401 .count();
311   311  
HITCBC 312   5465 constexpr auto long_max = 312   5401 constexpr auto long_max =
313   static_cast<long long>((std::numeric_limits<long>::max)()); 313   static_cast<long long>((std::numeric_limits<long>::max)());
314   auto capped_timer_us = 314   auto capped_timer_us =
HITCBC 315   5465 (std::min)((std::max)(static_cast<long long>(timer_timeout_us), 315   5401 (std::min)((std::max)(static_cast<long long>(timer_timeout_us),
HITCBC 316   5465 static_cast<long long>(0)), 316   5401 static_cast<long long>(0)),
HITCBC 317   5465 long_max); 317   5401 long_max);
318   318  
HITCBC 319   5465 if (requested_timeout_us < 0) 319   5401 if (requested_timeout_us < 0)
HITCBC 320   5463 return static_cast<long>(capped_timer_us); 320   5399 return static_cast<long>(capped_timer_us);
321   321  
322   return static_cast<long>( 322   return static_cast<long>(
HITCBC 323   2 (std::min)(static_cast<long long>(requested_timeout_us), 323   2 (std::min)(static_cast<long long>(requested_timeout_us),
HITCBC 324   2 capped_timer_us)); 324   2 capped_timer_us));
325   } 325   }
326   326  
327   inline void 327   inline void
HITCBC 328   34267 select_scheduler::run_task(lock_type& lock, context_type& ctx, long timeout_us) 328   32183 select_scheduler::run_task(lock_type& lock, context_type& ctx, long timeout_us)
329   { 329   {
330   long effective_timeout_us = 330   long effective_timeout_us =
HITCBC 331   34267 task_interrupted_ ? 0 : calculate_timeout(timeout_us); 331   32183 task_interrupted_ ? 0 : calculate_timeout(timeout_us);
332   332  
333   // Snapshot registered descriptors while holding lock. 333   // Snapshot registered descriptors while holding lock.
334   // Record which directions each fd needs monitored to avoid a hot 334   // Record which directions each fd needs monitored to avoid a hot
335   // loop: select is level-triggered, so a writable socket (nearly 335   // loop: select is level-triggered, so a writable socket (nearly
336   // always writable) or an always-readable fd (/dev/zero, a pipe at 336   // always writable) or an always-readable fd (/dev/zero, a pipe at
337   // EOF) would return select() immediately every iteration if 337   // EOF) would return select() immediately every iteration if
338   // unconditionally added. Membership in both sets is opt-in: a 338   // unconditionally added. Membership in both sets is opt-in: a
339   // parked op or wait in a direction opts that direction in. The 339   // parked op or wait in a direction opts that direction in. The
340   // exceptional set is opt-in too, for any parked op or wait: 340   // exceptional set is opt-in too, for any parked op or wait:
341   // Darwin reports a character device (/dev/zero, /dev/null) as 341   // Darwin reports a character device (/dev/zero, /dev/null) as
342   // exceptional on every call. 342   // exceptional on every call.
343   struct fd_entry 343   struct fd_entry
344   { 344   {
345   int fd; 345   int fd;
346   reactor_descriptor_state* desc; 346   reactor_descriptor_state* desc;
347   std::uint32_t want; 347   std::uint32_t want;
348   }; 348   };
349   fd_entry snapshot[FD_SETSIZE]; 349   fd_entry snapshot[FD_SETSIZE];
HITCBC 350   34267 int snapshot_count = 0; 350   32183 int snapshot_count = 0;
351   351  
HITCBC 352   113765 for (auto& [fd, desc] : registered_descs_) 352   109007 for (auto& [fd, desc] : registered_descs_)
353   { 353   {
HITCBC 354   79498 if (snapshot_count < FD_SETSIZE) 354   76824 if (snapshot_count < FD_SETSIZE)
355   { 355   {
HITCBC 356   79498 conditionally_enabled_mutex::scoped_lock desc_lock(desc->mutex); 356   76824 conditionally_enabled_mutex::scoped_lock desc_lock(desc->mutex);
HITCBC 357   79498 snapshot[snapshot_count].fd = fd; 357   76824 snapshot[snapshot_count].fd = fd;
HITCBC 358   79498 snapshot[snapshot_count].desc = desc; 358   76824 snapshot[snapshot_count].desc = desc;
HITCBC 359   79498 snapshot[snapshot_count].want = 359   76824 snapshot[snapshot_count].want =
HITCBC 360   79498 ((desc->read_op || desc->wait_read_op) ? reactor_event_read 360   76824 ((desc->read_op || desc->wait_read_op) ? reactor_event_read
HITCBC 361   79498 : 0) | 361   76824 : 0) |
HITCBC 362   79118 ((desc->write_op || desc->connect_op || desc->wait_write_op) 362   76444 ((desc->write_op || desc->connect_op || desc->wait_write_op)
HITCBC 363   158616 ? reactor_event_write 363   153268 ? reactor_event_write
HITCBC 364   79498 : 0) | 364   76824 : 0) |
HITCBC 365   79498 (desc->wait_error_op ? reactor_event_error : 0); 365   76824 (desc->wait_error_op ? reactor_event_error : 0);
HITCBC 366   79498 ++snapshot_count; 366   76824 ++snapshot_count;
HITCBC 367   79498 } 367   76824 }
368   } 368   }
369   369  
HITCBC 370   34267 if (lock.owns_lock()) 370   32183 if (lock.owns_lock())
HITCBC 371   7400 lock.unlock(); 371   7383 lock.unlock();
372   372  
HITCBC 373   34267 task_cleanup on_exit{this, &lock, ctx}; 373   32183 task_cleanup on_exit{this, &lock, ctx};
374   374  
375   fd_set read_fds, write_fds, except_fds; 375   fd_set read_fds, write_fds, except_fds;
HITCBC 376   582539 FD_ZERO(&read_fds); 376   547111 FD_ZERO(&read_fds);
HITCBC 377   582539 FD_ZERO(&write_fds); 377   547111 FD_ZERO(&write_fds);
HITCBC 378   582539 FD_ZERO(&except_fds); 378   547111 FD_ZERO(&except_fds);
379   379  
HITCBC 380   34267 FD_SET(pipe_fds_[0], &read_fds); 380   32183 FD_SET(pipe_fds_[0], &read_fds);
HITCBC 381   34267 int nfds = pipe_fds_[0]; 381   32183 int nfds = pipe_fds_[0];
382   382  
HITCBC 383   113765 for (int i = 0; i < snapshot_count; ++i) 383   109007 for (int i = 0; i < snapshot_count; ++i)
384   { 384   {
HITCBC 385   79498 int fd = snapshot[i].fd; 385   76824 int fd = snapshot[i].fd;
HITCBC 386   79498 if (snapshot[i].want & reactor_event_read) 386   76824 if (snapshot[i].want & reactor_event_read)
HITCBC 387   9473 FD_SET(fd, &read_fds); 387   9361 FD_SET(fd, &read_fds);
HITCBC 388   79498 if (snapshot[i].want & reactor_event_write) 388   76824 if (snapshot[i].want & reactor_event_write)
HITCBC 389   2482 FD_SET(fd, &write_fds); 389   2458 FD_SET(fd, &write_fds);
HITCBC 390   79498 if (snapshot[i].want != 0) 390   76824 if (snapshot[i].want != 0)
HITCBC 391   12245 FD_SET(fd, &except_fds); 391   12109 FD_SET(fd, &except_fds);
HITCBC 392   79498 if (fd > nfds) 392   76824 if (fd > nfds)
HITCBC 393   33235 nfds = fd; 393   31104 nfds = fd;
394   } 394   }
395   395  
396   struct timeval tv; 396   struct timeval tv;
HITCBC 397   34267 struct timeval* tv_ptr = nullptr; 397   32183 struct timeval* tv_ptr = nullptr;
HITCBC 398   34267 if (effective_timeout_us >= 0) 398   32183 if (effective_timeout_us >= 0)
399   { 399   {
HITCBC 400   33171 tv.tv_sec = effective_timeout_us / 1000000; 400   31027 tv.tv_sec = effective_timeout_us / 1000000;
HITCBC 401   33171 tv.tv_usec = effective_timeout_us % 1000000; 401   31027 tv.tv_usec = effective_timeout_us % 1000000;
HITCBC 402   33171 tv_ptr = &tv; 402   31027 tv_ptr = &tv;
403   } 403   }
404   404  
HITCBC 405   34267 int ready = ::select(nfds + 1, &read_fds, &write_fds, &except_fds, tv_ptr); 405   32183 int ready = ::select(nfds + 1, &read_fds, &write_fds, &except_fds, tv_ptr);
406   406  
407   // EINTR: signal interrupted select(), just retry. 407   // EINTR: signal interrupted select(), just retry.
408   // EBADF: an fd was closed between snapshot and select(); retry 408   // EBADF: an fd was closed between snapshot and select(); retry
409   // with a fresh snapshot from registered_descs_. 409   // with a fresh snapshot from registered_descs_.
410   // Both fall through with no ready descriptors rather than 410   // Both fall through with no ready descriptors rather than
411   // returning: the caller handed this function an owned lock that 411   // returning: the caller handed this function an owned lock that
412   // only the epilogue below re-acquires. 412   // only the epilogue below re-acquires.
HITCBC 413   34267 if (ready < 0) 413   32183 if (ready < 0)
414   { 414   {
HITCBC 415   3 if (errno != EINTR && errno != EBADF) 415   3 if (errno != EINTR && errno != EBADF)
HITCBC 416   1 detail::throw_system_error(make_err(errno), "select"); 416   1 detail::throw_system_error(make_err(errno), "select");
HITCBC 417   2 ready = 0; 417   2 ready = 0;
418   } 418   }
419   419  
420   // Process timers outside the lock 420   // Process timers outside the lock
HITCBC 421   34266 timer_svc_->process_expired(); 421   32182 timer_svc_->process_expired();
422   422  
HITCBC 423   34266 ready_queue local_ops; 423   32182 ready_queue local_ops;
424   424  
HITCBC 425   34266 if (ready > 0) 425   32182 if (ready > 0)
426   { 426   {
HITCBC 427   6843 if (FD_ISSET(pipe_fds_[0], &read_fds)) 427   6885 if (FD_ISSET(pipe_fds_[0], &read_fds))
428   { 428   {
429   char buf[256]; 429   char buf[256];
HITCBC 430   13154 while (::read(pipe_fds_[0], buf, sizeof(buf)) > 0) 430   13238 while (::read(pipe_fds_[0], buf, sizeof(buf)) > 0)
431   { 431   {
432   } 432   }
433   } 433   }
434   434  
HITCBC 435   16798 for (int i = 0; i < snapshot_count; ++i) 435   16779 for (int i = 0; i < snapshot_count; ++i)
436   { 436   {
HITCBC 437   9955 int fd = snapshot[i].fd; 437   9894 int fd = snapshot[i].fd;
HITCBC 438   9955 reactor_descriptor_state* desc = snapshot[i].desc; 438   9894 reactor_descriptor_state* desc = snapshot[i].desc;
439   439  
HITCBC 440   9955 std::uint32_t flags = 0; 440   9894 std::uint32_t flags = 0;
HITCBC 441   9955 if (FD_ISSET(fd, &read_fds)) 441   9894 if (FD_ISSET(fd, &read_fds))
HITCBC 442   2505 flags |= reactor_event_read; 442   2481 flags |= reactor_event_read;
HITCBC 443   9955 if (FD_ISSET(fd, &write_fds)) 443   9894 if (FD_ISSET(fd, &write_fds))
HITCBC 444   2188 flags |= reactor_event_write; 444   2164 flags |= reactor_event_write;
HITCBC 445   9955 if (FD_ISSET(fd, &except_fds)) 445   9894 if (FD_ISSET(fd, &except_fds))
HITCBC 446   6 flags |= reactor_event_error; 446   6 flags |= reactor_event_error;
447   447  
HITCBC 448   9955 if (flags == 0) 448   9894 if (flags == 0)
HITCBC 449   5260 continue; 449   5247 continue;
450   450  
HITCBC 451   4695 desc->add_ready_events(flags); 451   4647 desc->add_ready_events(flags);
452   452  
HITCBC 453   4695 bool expected = false; 453   4647 bool expected = false;
HITCBC 454   4695 if (desc->is_enqueued_.compare_exchange_strong( 454   4647 if (desc->is_enqueued_.compare_exchange_strong(
455   expected, true, std::memory_order_release, 455   expected, true, std::memory_order_release,
456   std::memory_order_relaxed)) 456   std::memory_order_relaxed))
457   { 457   {
HITCBC 458   4695 local_ops.push(desc); 458   4647 local_ops.push(desc);
459   } 459   }
460   } 460   }
461   } 461   }
462   462  
HITCBC 463   34266 lock.lock(); 463   32182 lock.lock();
464   464  
HITCBC 465   34266 completed_ops_.splice(local_ops); 465   32182 completed_ops_.splice(local_ops);
HITCBC 466   34267 } 466   32183 }
467   467  
468   } // namespace boost::corosio::detail 468   } // namespace boost::corosio::detail
469   469  
470   #endif // BOOST_COROSIO_HAS_SELECT 470   #endif // BOOST_COROSIO_HAS_SELECT
471   471  
472   #endif // BOOST_COROSIO_NATIVE_DETAIL_SELECT_SELECT_SCHEDULER_HPP 472   #endif // BOOST_COROSIO_NATIVE_DETAIL_SELECT_SELECT_SCHEDULER_HPP