. /include/boost/corosio /native /detail /select /select_scheduler.hpp
99.41% Lines (168/169) 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).
135   [[nodiscard]] std::error_code 135   [[nodiscard]] std::error_code
HITCBC 136   55 register_signal_reader(int read_fd) override 136   55 register_signal_reader(int read_fd) override
137   { 137   {
HITCBC 138   55 return register_descriptor(read_fd, signal_pipe_reader_.arm()); 138   55 return register_descriptor(read_fd, signal_pipe_reader_.arm());
139   } 139   }
140   140  
141   private: 141   private:
142   void 142   void
143   run_task(lock_type& lock, context_type& ctx, 143   run_task(lock_type& lock, context_type& ctx,
144   long timeout_us) override; 144   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*> registered_descs_; 156   mutable std::unordered_map<int, reactor_descriptor_state*> registered_descs_;
157   mutable int max_fd_ = -1; 157   mutable int max_fd_ = -1;
158   }; 158   };
159   159  
HITCBC 160   881 inline select_scheduler::select_scheduler(capy::execution_context& ctx, int) 160   886 inline select_scheduler::select_scheduler(capy::execution_context& ctx, int)
HITCBC 161   881 : pipe_fds_{-1, -1} 161   886 : pipe_fds_{-1, -1}
HITCBC 162   881 , max_fd_(-1) 162   886 , max_fd_(-1)
163   { 163   {
HITCBC 164   881 if (::pipe(pipe_fds_) < 0) 164   886 if (::pipe(pipe_fds_) < 0)
HITCBC 165   1 detail::throw_system_error(make_err(errno), "pipe"); 165   1 detail::throw_system_error(make_err(errno), "pipe");
166   166  
HITCBC 167   2631 for (int i = 0; i < 2; ++i) 167   2646 for (int i = 0; i < 2; ++i)
168   { 168   {
HITCBC 169   1757 int flags = ::fcntl(pipe_fds_[i], F_GETFL, 0); 169   1767 int flags = ::fcntl(pipe_fds_[i], F_GETFL, 0);
HITCBC 170   1757 if (flags == -1) 170   1767 if (flags == -1)
171   { 171   {
HITCBC 172   2 int errn = errno; 172   2 int errn = errno;
HITCBC 173   2 ::close(pipe_fds_[0]); 173   2 ::close(pipe_fds_[0]);
HITCBC 174   2 ::close(pipe_fds_[1]); 174   2 ::close(pipe_fds_[1]);
HITCBC 175   2 detail::throw_system_error(make_err(errn), "fcntl F_GETFL"); 175   2 detail::throw_system_error(make_err(errn), "fcntl F_GETFL");
176   } 176   }
HITCBC 177   1755 if (::fcntl(pipe_fds_[i], F_SETFL, flags | O_NONBLOCK) == -1) 177   1765 if (::fcntl(pipe_fds_[i], F_SETFL, flags | O_NONBLOCK) == -1)
178   { 178   {
HITCBC 179   2 int errn = errno; 179   2 int errn = errno;
HITCBC 180   2 ::close(pipe_fds_[0]); 180   2 ::close(pipe_fds_[0]);
HITCBC 181   2 ::close(pipe_fds_[1]); 181   2 ::close(pipe_fds_[1]);
HITCBC 182   2 detail::throw_system_error(make_err(errn), "fcntl F_SETFL"); 182   2 detail::throw_system_error(make_err(errn), "fcntl F_SETFL");
183   } 183   }
HITCBC 184   1753 if (::fcntl(pipe_fds_[i], F_SETFD, FD_CLOEXEC) == -1) 184   1763 if (::fcntl(pipe_fds_[i], F_SETFD, FD_CLOEXEC) == -1)
185   { 185   {
HITCBC 186   2 int errn = errno; 186   2 int errn = errno;
HITCBC 187   2 ::close(pipe_fds_[0]); 187   2 ::close(pipe_fds_[0]);
HITCBC 188   2 ::close(pipe_fds_[1]); 188   2 ::close(pipe_fds_[1]);
HITCBC 189   2 detail::throw_system_error(make_err(errn), "fcntl F_SETFD"); 189   2 detail::throw_system_error(make_err(errn), "fcntl F_SETFD");
190   } 190   }
191   } 191   }
192   192  
HITCBC 193   874 timer_svc_ = &get_timer_service(ctx, *this); 193   879 timer_svc_ = &get_timer_service(ctx, *this);
HITCBC 194   874 timer_svc_->set_on_earliest_changed( 194   879 timer_svc_->set_on_earliest_changed(
HITCBC 195   5282 timer_service::callback(this, [](void* p) { 195   3714 timer_service::callback(this, [](void* p) {
HITCBC 196   4408 static_cast<select_scheduler*>(p)->interrupt_reactor(); 196   2835 static_cast<select_scheduler*>(p)->interrupt_reactor();
HITCBC 197   4408 })); 197   2835 }));
198   198  
HITCBC 199   874 get_resolver_service(ctx, *this); 199   879 get_resolver_service(ctx, *this);
HITCBC 200   874 get_signal_service(ctx, *this); 200   879 get_signal_service(ctx, *this);
HITCBC 201   874 get_stream_file_service(ctx, *this); 201   879 get_stream_file_service(ctx, *this);
HITCBC 202   874 get_random_access_file_service(ctx, *this); 202   879 get_random_access_file_service(ctx, *this);
203   203  
HITCBC 204   874 completed_ops_.push(&task_op_); 204   879 completed_ops_.push(&task_op_);
HITCBC 205   895 } 205   900 }
206   206  
HITCBC 207   1748 inline select_scheduler::~select_scheduler() 207   1758 inline select_scheduler::~select_scheduler()
208   { 208   {
HITCBC 209   874 if (pipe_fds_[0] >= 0) 209   879 if (pipe_fds_[0] >= 0)
HITCBC 210   874 ::close(pipe_fds_[0]); 210   879 ::close(pipe_fds_[0]);
HITCBC 211   874 if (pipe_fds_[1] >= 0) 211   879 if (pipe_fds_[1] >= 0)
HITCBC 212   874 ::close(pipe_fds_[1]); 212   879 ::close(pipe_fds_[1]);
HITCBC 213   1748 } 213   1758 }
214   214  
215   inline void 215   inline void
HITCBC 216   874 select_scheduler::shutdown() 216   879 select_scheduler::shutdown()
217   { 217   {
HITCBC 218   874 shutdown_drain(); 218   879 shutdown_drain();
219   219  
HITCBC 220   874 if (pipe_fds_[1] >= 0) 220   879 if (pipe_fds_[1] >= 0)
HITCBC 221   874 interrupt_reactor(); 221   879 interrupt_reactor();
HITCBC 222   874 } 222   879 }
223   223  
224   inline std::error_code 224   inline std::error_code
HITCBC 225   8022 select_scheduler::register_descriptor( 225   5061 select_scheduler::register_descriptor(
226   int fd, reactor_descriptor_state* desc) const 226   int fd, reactor_descriptor_state* desc) const
227   { 227   {
HITCBC 228   8022 if (fd < 0 || fd >= FD_SETSIZE) 228   5061 if (fd < 0 || fd >= FD_SETSIZE)
HITCBC 229   1 return make_err(EMFILE); 229   1 return make_err(EMFILE);
230   230  
HITCBC 231   8021 desc->registered_events = reactor_event_read | reactor_event_write; 231   5060 desc->registered_events = reactor_event_read | reactor_event_write;
HITCBC 232   8021 desc->fd = fd; 232   5060 desc->fd = fd;
HITCBC 233   8021 desc->scheduler_ = this; 233   5060 desc->scheduler_ = this;
HITCBC 234   8021 desc->mutex.set_enabled(reactor_io_locking_); 234   5060 desc->mutex.set_enabled(reactor_io_locking_);
HITCBC 235   8021 desc->ready_events_.store(0, std::memory_order_relaxed); 235   5060 desc->ready_events_.store(0, std::memory_order_relaxed);
236   236  
237   { 237   {
HITCBC 238   8021 conditionally_enabled_mutex::scoped_lock lock(desc->mutex); 238   5060 conditionally_enabled_mutex::scoped_lock lock(desc->mutex);
HITCBC 239   8021 desc->impl_ref_.reset(); 239   5060 desc->impl_ref_.reset();
HITCBC 240   8021 desc->read_ready = false; 240   5060 desc->read_ready = false;
HITCBC 241   8021 desc->write_ready = false; 241   5060 desc->write_ready = false;
HITCBC 242   8021 } 242   5060 }
243   243  
244   { 244   {
HITCBC 245   8021 mutex_type::scoped_lock lock(mutex_); 245   5060 mutex_type::scoped_lock lock(mutex_);
246   try 246   try
247   { 247   {
HITCBC 248   8021 registered_descs_[fd] = desc; 248   5060 registered_descs_[fd] = desc;
249   } 249   }
HITCBC 250   1 catch (std::bad_alloc const&) 250   1 catch (std::bad_alloc const&)
251   { 251   {
HITCBC 252   1 return make_err(ENOMEM); 252   1 return make_err(ENOMEM);
HITCBC 253   1 } 253   1 }
HITCBC 254   8020 if (fd > max_fd_) 254   5059 if (fd > max_fd_)
HITCBC 255   7966 max_fd_ = fd; 255   5005 max_fd_ = fd;
HITCBC 256   8021 } 256   5060 }
257   257  
HITCBC 258   8020 interrupt_reactor(); 258   5059 interrupt_reactor();
HITCBC 259   8020 return {}; 259   5059 return {};
260   } 260   }
261   261  
262   inline void 262   inline void
HITCBC 263   7966 select_scheduler::deregister_descriptor(int fd) const 263   5005 select_scheduler::deregister_descriptor(int fd) const
264   { 264   {
HITCBC 265   7966 mutex_type::scoped_lock lock(mutex_); 265   5005 mutex_type::scoped_lock lock(mutex_);
266   266  
HITCBC 267   7966 auto it = registered_descs_.find(fd); 267   5005 auto it = registered_descs_.find(fd);
HITCBC 268   7966 if (it == registered_descs_.end()) 268   5005 if (it == registered_descs_.end())
MISUBC 269   ✗ return; 269   ✗ return;
270   270  
HITCBC 271   7966 registered_descs_.erase(it); 271   5005 registered_descs_.erase(it);
272   272  
HITCBC 273   7966 if (fd == max_fd_) 273   5005 if (fd == max_fd_)
274   { 274   {
HITCBC 275   7629 max_fd_ = pipe_fds_[0]; 275   4669 max_fd_ = pipe_fds_[0];
HITCBC 276   14859 for (auto& [registered_fd, state] : registered_descs_) 276   8936 for (auto& [registered_fd, state] : registered_descs_)
277   { 277   {
HITCBC 278   7230 if (registered_fd > max_fd_) 278   4267 if (registered_fd > max_fd_)
HITCBC 279   7137 max_fd_ = registered_fd; 279   4174 max_fd_ = registered_fd;
280   } 280   }
281   } 281   }
HITCBC 282   7966 } 282   5005 }
283   283  
284   inline void 284   inline void
HITCBC 285   3745 select_scheduler::notify_reactor() const 285   2262 select_scheduler::notify_reactor() const
286   { 286   {
HITCBC 287   3745 interrupt_reactor(); 287   2262 interrupt_reactor();
HITCBC 288   3745 } 288   2262 }
289   289  
290   inline void 290   inline void
HITCBC 291   17846 select_scheduler::interrupt_reactor() const 291   11839 select_scheduler::interrupt_reactor() const
292   { 292   {
HITCBC 293   17846 char byte = 1; 293   11839 char byte = 1;
HITCBC 294   17846 [[maybe_unused]] auto r = ::write(pipe_fds_[1], &byte, 1); 294   11839 [[maybe_unused]] auto r = ::write(pipe_fds_[1], &byte, 1);
HITCBC 295   17846 } 295   11839 }
296   296  
297   inline long 297   inline long
HITCBC 298   334796 select_scheduler::calculate_timeout(long requested_timeout_us) const 298   334914 select_scheduler::calculate_timeout(long requested_timeout_us) const
299   { 299   {
HITCBC 300   334796 if (requested_timeout_us == 0) 300   334914 if (requested_timeout_us == 0)
EUB 301 - ✗ return 0; 301 + return 0; // LCOV_EXCL_LINE run_task passes 0 via task_interrupted_, never through this argument
302   302  
HITCBC 303   334796 auto nearest = timer_svc_->nearest_expiry(); 303   334914 auto nearest = timer_svc_->nearest_expiry();
HITCBC 304   334796 if (nearest == timer_service::time_point::max()) 304   334914 if (nearest == timer_service::time_point::max())
HITCBC 305   744 return requested_timeout_us; 305   734 return requested_timeout_us;
306   306  
HITCBC 307   334052 auto now = std::chrono::steady_clock::now(); 307   334180 auto now = std::chrono::steady_clock::now();
HITCBC 308   334052 if (nearest <= now) 308   334180 if (nearest <= now)
HITCBC 309   1842 return 0; 309   1083 return 0;
310   310  
311   auto timer_timeout_us = 311   auto timer_timeout_us =
HITCBC 312   332210 std::chrono::duration_cast<std::chrono::microseconds>(nearest - now) 312   333097 std::chrono::duration_cast<std::chrono::microseconds>(nearest - now)
HITCBC 313   332210 .count(); 313   333097 .count();
314   314  
HITCBC 315   332210 constexpr auto long_max = 315   333097 constexpr auto long_max =
316   static_cast<long long>((std::numeric_limits<long>::max)()); 316   static_cast<long long>((std::numeric_limits<long>::max)());
317   auto capped_timer_us = 317   auto capped_timer_us =
HITCBC 318   332210 (std::min)((std::max)(static_cast<long long>(timer_timeout_us), 318   333097 (std::min)((std::max)(static_cast<long long>(timer_timeout_us),
HITCBC 319   332210 static_cast<long long>(0)), 319   333097 static_cast<long long>(0)),
HITCBC 320   332210 long_max); 320   333097 long_max);
321   321  
HITCBC 322   332210 if (requested_timeout_us < 0) 322   333097 if (requested_timeout_us < 0)
HITCBC 323   332208 return static_cast<long>(capped_timer_us); 323   333095 return static_cast<long>(capped_timer_us);
324   324  
325   return static_cast<long>( 325   return static_cast<long>(
HITCBC 326   2 (std::min)(static_cast<long long>(requested_timeout_us), 326   2 (std::min)(static_cast<long long>(requested_timeout_us),
HITCBC 327   2 capped_timer_us)); 327   2 capped_timer_us));
328   } 328   }
329   329  
330   inline void 330   inline void
HITCBC 331   361461 select_scheduler::run_task( 331   360544 select_scheduler::run_task(
332   lock_type& lock, context_type& ctx, long timeout_us) 332   lock_type& lock, context_type& ctx, long timeout_us)
333   { 333   {
334   long effective_timeout_us = 334   long effective_timeout_us =
HITCBC 335   361461 task_interrupted_ ? 0 : calculate_timeout(timeout_us); 335   360544 task_interrupted_ ? 0 : calculate_timeout(timeout_us);
336   336  
337   // Snapshot registered descriptors while holding lock. 337   // Snapshot registered descriptors while holding lock.
338   // Record which fds need write monitoring to avoid a hot loop: 338   // Record which fds need write monitoring to avoid a hot loop:
339   // select is level-triggered so writable sockets (nearly always 339   // select is level-triggered so writable sockets (nearly always
340   // writable) would cause select() to return immediately every 340   // writable) would cause select() to return immediately every
341   // iteration if unconditionally added to write_fds. Membership 341   // iteration if unconditionally added to write_fds. Membership
342   // stays opt-in: a parked write wait opts in the same way a 342   // stays opt-in: a parked write wait opts in the same way a
343   // parked write or connect op does. 343   // parked write or connect op does.
344   struct fd_entry 344   struct fd_entry
345   { 345   {
346   int fd; 346   int fd;
347   reactor_descriptor_state* desc; 347   reactor_descriptor_state* desc;
348   bool needs_write; 348   bool needs_write;
349   }; 349   };
350   fd_entry snapshot[FD_SETSIZE]; 350   fd_entry snapshot[FD_SETSIZE];
HITCBC 351   361461 int snapshot_count = 0; 351   360544 int snapshot_count = 0;
352   352  
HITCBC 353   918884 for (auto& [fd, desc] : registered_descs_) 353   909725 for (auto& [fd, desc] : registered_descs_)
354   { 354   {
HITCBC 355   557423 if (snapshot_count < FD_SETSIZE) 355   549181 if (snapshot_count < FD_SETSIZE)
356   { 356   {
HITCBC 357   557423 conditionally_enabled_mutex::scoped_lock desc_lock(desc->mutex); 357   549181 conditionally_enabled_mutex::scoped_lock desc_lock(desc->mutex);
HITCBC 358   557423 snapshot[snapshot_count].fd = fd; 358   549181 snapshot[snapshot_count].fd = fd;
HITCBC 359   557423 snapshot[snapshot_count].desc = desc; 359   549181 snapshot[snapshot_count].desc = desc;
HITCBC 360   557423 snapshot[snapshot_count].needs_write = 360   549181 snapshot[snapshot_count].needs_write =
HITCBC 361   1100331 (desc->write_op || desc->connect_op || 361   1087087 (desc->write_op || desc->connect_op ||
HITCBC 362   542908 desc->wait_write_op); 362   537906 desc->wait_write_op);
HITCBC 363   557423 ++snapshot_count; 363   549181 ++snapshot_count;
HITCBC 364   557423 } 364   549181 }
365   } 365   }
366   366  
HITCBC 367   361461 if (lock.owns_lock()) 367   360544 if (lock.owns_lock())
HITCBC 368   334797 lock.unlock(); 368   334915 lock.unlock();
369   369  
HITCBC 370   361461 task_cleanup on_exit{this, &lock, ctx}; 370   360544 task_cleanup on_exit{this, &lock, ctx};
371   371  
372   fd_set read_fds, write_fds, except_fds; 372   fd_set read_fds, write_fds, except_fds;
HITCBC 373   6144837 FD_ZERO(&read_fds); 373   6129248 FD_ZERO(&read_fds);
HITCBC 374   6144837 FD_ZERO(&write_fds); 374   6129248 FD_ZERO(&write_fds);
HITCBC 375   6144837 FD_ZERO(&except_fds); 375   6129248 FD_ZERO(&except_fds);
376   376  
HITCBC 377   361461 FD_SET(pipe_fds_[0], &read_fds); 377   360544 FD_SET(pipe_fds_[0], &read_fds);
HITCBC 378   361461 int nfds = pipe_fds_[0]; 378   360544 int nfds = pipe_fds_[0];
379   379  
HITCBC 380   918884 for (int i = 0; i < snapshot_count; ++i) 380   909725 for (int i = 0; i < snapshot_count; ++i)
381   { 381   {
HITCBC 382   557423 int fd = snapshot[i].fd; 382   549181 int fd = snapshot[i].fd;
HITCBC 383   557423 FD_SET(fd, &read_fds); 383   549181 FD_SET(fd, &read_fds);
HITCBC 384   557423 if (snapshot[i].needs_write) 384   549181 if (snapshot[i].needs_write)
HITCBC 385   14521 FD_SET(fd, &write_fds); 385   11281 FD_SET(fd, &write_fds);
HITCBC 386   557423 FD_SET(fd, &except_fds); 386   549181 FD_SET(fd, &except_fds);
HITCBC 387   557423 if (fd > nfds) 387   549181 if (fd > nfds)
HITCBC 388   361064 nfds = fd; 388   360156 nfds = fd;
389   } 389   }
390   390  
391   struct timeval tv; 391   struct timeval tv;
HITCBC 392   361461 struct timeval* tv_ptr = nullptr; 392   360544 struct timeval* tv_ptr = nullptr;
HITCBC 393   361461 if (effective_timeout_us >= 0) 393   360544 if (effective_timeout_us >= 0)
394   { 394   {
HITCBC 395   360735 tv.tv_sec = effective_timeout_us / 1000000; 395   359826 tv.tv_sec = effective_timeout_us / 1000000;
HITCBC 396   360735 tv.tv_usec = effective_timeout_us % 1000000; 396   359826 tv.tv_usec = effective_timeout_us % 1000000;
HITCBC 397   360735 tv_ptr = &tv; 397   359826 tv_ptr = &tv;
398   } 398   }
399   399  
HITCBC 400   361461 int ready = ::select(nfds + 1, &read_fds, &write_fds, &except_fds, tv_ptr); 400   360544 int ready = ::select(nfds + 1, &read_fds, &write_fds, &except_fds, tv_ptr);
401   401  
402   // EINTR: signal interrupted select(), just retry. 402   // EINTR: signal interrupted select(), just retry.
403   // EBADF: an fd was closed between snapshot and select(); retry 403   // EBADF: an fd was closed between snapshot and select(); retry
404   // with a fresh snapshot from registered_descs_. 404   // with a fresh snapshot from registered_descs_.
405   // Both fall through with no ready descriptors rather than 405   // Both fall through with no ready descriptors rather than
406   // returning: the caller handed this function an owned lock that 406   // returning: the caller handed this function an owned lock that
407   // only the epilogue below re-acquires. 407   // only the epilogue below re-acquires.
HITCBC 408   361461 if (ready < 0) 408   360544 if (ready < 0)
409   { 409   {
HITCBC 410   3 if (errno != EINTR && errno != EBADF) 410   3 if (errno != EINTR && errno != EBADF)
HITCBC 411   1 detail::throw_system_error(make_err(errno), "select"); 411   1 detail::throw_system_error(make_err(errno), "select");
HITCBC 412   2 ready = 0; 412   2 ready = 0;
413   } 413   }
414   414  
415   // Process timers outside the lock 415   // Process timers outside the lock
HITCBC 416   361460 timer_svc_->process_expired(); 416   360543 timer_svc_->process_expired();
417   417  
HITCBC 418   361460 ready_queue local_ops; 418   360543 ready_queue local_ops;
419   419  
HITCBC 420   361460 if (ready > 0) 420   360543 if (ready > 0)
421   { 421   {
HITCBC 422   342346 if (FD_ISSET(pipe_fds_[0], &read_fds)) 422   343635 if (FD_ISSET(pipe_fds_[0], &read_fds))
423   { 423   {
424   char buf[256]; 424   char buf[256];
HITCBC 425   16410 while (::read(pipe_fds_[0], buf, sizeof(buf)) > 0) 425   10466 while (::read(pipe_fds_[0], buf, sizeof(buf)) > 0)
426   { 426   {
427   } 427   }
428   } 428   }
429   429  
HITCBC 430   846363 for (int i = 0; i < snapshot_count; ++i) 430   840180 for (int i = 0; i < snapshot_count; ++i)
431   { 431   {
HITCBC 432   504017 int fd = snapshot[i].fd; 432   496545 int fd = snapshot[i].fd;
HITCBC 433   504017 reactor_descriptor_state* desc = snapshot[i].desc; 433   496545 reactor_descriptor_state* desc = snapshot[i].desc;
434   434  
HITCBC 435   504017 std::uint32_t flags = 0; 435   496545 std::uint32_t flags = 0;
HITCBC 436   504017 if (FD_ISSET(fd, &read_fds)) 436   496545 if (FD_ISSET(fd, &read_fds))
HITCBC 437   339969 flags |= reactor_event_read; 437   343030 flags |= reactor_event_read;
HITCBC 438   504017 if (FD_ISSET(fd, &write_fds)) 438   496545 if (FD_ISSET(fd, &write_fds))
HITCBC 439   3737 flags |= reactor_event_write; 439   2254 flags |= reactor_event_write;
HITCBC 440   504017 if (FD_ISSET(fd, &except_fds)) 440   496545 if (FD_ISSET(fd, &except_fds))
HITCBC 441   16 flags |= reactor_event_error; 441   16 flags |= reactor_event_error;
442   442  
HITCBC 443   504017 if (flags == 0) 443   496545 if (flags == 0)
HITCBC 444   160320 continue; 444   151269 continue;
445   445  
HITCBC 446   343697 desc->add_ready_events(flags); 446   345276 desc->add_ready_events(flags);
447   447  
HITCBC 448   343697 bool expected = false; 448   345276 bool expected = false;
HITCBC 449   343697 if (desc->is_enqueued_.compare_exchange_strong( 449   345276 if (desc->is_enqueued_.compare_exchange_strong(
450   expected, true, std::memory_order_release, 450   expected, true, std::memory_order_release,
451   std::memory_order_relaxed)) 451   std::memory_order_relaxed))
452   { 452   {
HITCBC 453   343697 local_ops.push(desc); 453   345276 local_ops.push(desc);
454   } 454   }
455   } 455   }
456   } 456   }
457   457  
HITCBC 458   361460 lock.lock(); 458   360543 lock.lock();
459   459  
HITCBC 460   361460 completed_ops_.splice(local_ops); 460   360543 completed_ops_.splice(local_ops);
HITCBC 461   361461 } 461   360544 }
462   462  
463   } // namespace boost::corosio::detail 463   } // namespace boost::corosio::detail
464   464  
465   #endif // BOOST_COROSIO_HAS_SELECT 465   #endif // BOOST_COROSIO_HAS_SELECT
466   466  
467   #endif // BOOST_COROSIO_NATIVE_DETAIL_SELECT_SELECT_SCHEDULER_HPP 467   #endif // BOOST_COROSIO_NATIVE_DETAIL_SELECT_SELECT_SCHEDULER_HPP