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