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 : /** Deregister a persistently registered descriptor.
117 :
118 : @param fd The file descriptor to deregister.
119 : */
120 : void deregister_descriptor(int fd) const;
121 :
122 : /** Interrupt the reactor so it rebuilds its fd_sets.
123 :
124 : Called when a write, connect, or write-wait op is registered
125 : after the reactor's snapshot was taken. Without this,
126 : select() may block not watching for writability on the fd.
127 : */
128 : void notify_reactor() const;
129 :
130 : /// Watch the read end of the POSIX signal self-pipe (see scheduler.hpp).
131 HIT 61 : [[nodiscard]] std::error_code register_signal_reader(int read_fd) override
132 : {
133 61 : return register_descriptor(read_fd, signal_pipe_reader_.arm());
134 : }
135 :
136 : private:
137 : void run_task(lock_type& lock, context_type& ctx, long timeout_us) override;
138 : void interrupt_reactor() const override;
139 : long calculate_timeout(long requested_timeout_us) const;
140 :
141 : // Watches the global signal self-pipe's read end (armed lazily by
142 : // register_signal_reader on the first signal registration).
143 : reactor_signal_pipe_reader signal_pipe_reader_;
144 :
145 : // Self-pipe for interrupting select()
146 : int pipe_fds_[2]; // [0]=read, [1]=write
147 :
148 : // Per-fd tracking for fd_set building
149 : mutable std::unordered_map<int, reactor_descriptor_state*>
150 : registered_descs_;
151 : mutable int max_fd_ = -1;
152 : };
153 :
154 983 : inline select_scheduler::select_scheduler(capy::execution_context& ctx, int)
155 983 : : pipe_fds_{-1, -1}
156 983 : , max_fd_(-1)
157 : {
158 983 : if (::pipe(pipe_fds_) < 0)
159 1 : detail::throw_system_error(make_err(errno), "pipe");
160 :
161 2937 : for (int i = 0; i < 2; ++i)
162 : {
163 1961 : int flags = ::fcntl(pipe_fds_[i], F_GETFL, 0);
164 1961 : if (flags == -1)
165 : {
166 2 : int errn = errno;
167 2 : ::close(pipe_fds_[0]);
168 2 : ::close(pipe_fds_[1]);
169 2 : detail::throw_system_error(make_err(errn), "fcntl F_GETFL");
170 : }
171 1959 : if (::fcntl(pipe_fds_[i], F_SETFL, flags | O_NONBLOCK) == -1)
172 : {
173 2 : int errn = errno;
174 2 : ::close(pipe_fds_[0]);
175 2 : ::close(pipe_fds_[1]);
176 2 : detail::throw_system_error(make_err(errn), "fcntl F_SETFL");
177 : }
178 1957 : if (::fcntl(pipe_fds_[i], F_SETFD, FD_CLOEXEC) == -1)
179 : {
180 2 : int errn = errno;
181 2 : ::close(pipe_fds_[0]);
182 2 : ::close(pipe_fds_[1]);
183 2 : detail::throw_system_error(make_err(errn), "fcntl F_SETFD");
184 : }
185 : }
186 :
187 976 : timer_svc_ = &get_timer_service(ctx, *this);
188 976 : timer_svc_->set_on_earliest_changed(
189 3640 : timer_service::callback(this, [](void* p) {
190 2664 : static_cast<select_scheduler*>(p)->interrupt_reactor();
191 2664 : }));
192 :
193 976 : completed_ops_.push(&task_op_);
194 997 : }
195 :
196 1952 : inline select_scheduler::~select_scheduler()
197 : {
198 976 : if (pipe_fds_[0] >= 0)
199 976 : ::close(pipe_fds_[0]);
200 976 : if (pipe_fds_[1] >= 0)
201 976 : ::close(pipe_fds_[1]);
202 1952 : }
203 :
204 : inline void
205 976 : select_scheduler::shutdown()
206 : {
207 976 : shutdown_drain();
208 :
209 976 : if (pipe_fds_[1] >= 0)
210 976 : interrupt_reactor();
211 976 : }
212 :
213 : inline std::error_code
214 4888 : select_scheduler::register_descriptor(
215 : int fd, reactor_descriptor_state* desc) const
216 : {
217 4888 : if (fd < 0 || fd >= FD_SETSIZE)
218 1 : return make_err(EMFILE);
219 :
220 4887 : desc->registered_events = reactor_event_read | reactor_event_write;
221 4887 : desc->fd = fd;
222 4887 : desc->scheduler_ = this;
223 4887 : desc->mutex.set_enabled(reactor_io_locking_);
224 4887 : desc->ready_events_.store(0, std::memory_order_relaxed);
225 :
226 : {
227 4887 : conditionally_enabled_mutex::scoped_lock lock(desc->mutex);
228 4887 : desc->impl_ref_.reset();
229 4887 : desc->read_ready = false;
230 4887 : desc->write_ready = false;
231 4887 : }
232 :
233 : {
234 4887 : mutex_type::scoped_lock lock(mutex_);
235 : try
236 : {
237 4887 : registered_descs_[fd] = desc;
238 : }
239 1 : catch (std::bad_alloc const&)
240 : {
241 1 : return make_err(ENOMEM);
242 1 : }
243 4886 : if (fd > max_fd_)
244 4835 : max_fd_ = fd;
245 4887 : }
246 :
247 4886 : interrupt_reactor();
248 4886 : return {};
249 : }
250 :
251 : inline void
252 4826 : select_scheduler::deregister_descriptor(int fd) const
253 : {
254 4826 : mutex_type::scoped_lock lock(mutex_);
255 :
256 4826 : auto it = registered_descs_.find(fd);
257 4826 : if (it == registered_descs_.end())
258 MIS 0 : return;
259 :
260 HIT 4826 : registered_descs_.erase(it);
261 :
262 4826 : if (fd == max_fd_)
263 : {
264 4491 : max_fd_ = pipe_fds_[0];
265 8554 : for (auto& [registered_fd, state] : registered_descs_)
266 : {
267 4063 : if (registered_fd > max_fd_)
268 3970 : max_fd_ = registered_fd;
269 : }
270 : }
271 4826 : }
272 :
273 : inline void
274 2162 : select_scheduler::notify_reactor() const
275 : {
276 2162 : interrupt_reactor();
277 2162 : }
278 :
279 : inline void
280 12311 : select_scheduler::interrupt_reactor() const
281 : {
282 12311 : char byte = 1;
283 12311 : [[maybe_unused]] auto r = ::write(pipe_fds_[1], &byte, 1);
284 12311 : }
285 :
286 : inline long
287 308536 : select_scheduler::calculate_timeout(long requested_timeout_us) const
288 : {
289 308536 : if (requested_timeout_us == 0)
290 : return 0; // LCOV_EXCL_LINE run_task passes 0 via task_interrupted_, never through this argument
291 :
292 308536 : auto nearest = timer_svc_->nearest_expiry();
293 308536 : if (nearest == timer_service::time_point::max())
294 1023 : return requested_timeout_us;
295 :
296 307513 : auto now = std::chrono::steady_clock::now();
297 307513 : if (nearest <= now)
298 439 : return 0;
299 :
300 : auto timer_timeout_us =
301 307074 : std::chrono::duration_cast<std::chrono::microseconds>(nearest - now)
302 307074 : .count();
303 :
304 307074 : constexpr auto long_max =
305 : static_cast<long long>((std::numeric_limits<long>::max)());
306 : auto capped_timer_us =
307 307074 : (std::min)((std::max)(static_cast<long long>(timer_timeout_us),
308 307074 : static_cast<long long>(0)),
309 307074 : long_max);
310 :
311 307074 : if (requested_timeout_us < 0)
312 307072 : return static_cast<long>(capped_timer_us);
313 :
314 : return static_cast<long>(
315 2 : (std::min)(static_cast<long long>(requested_timeout_us),
316 2 : capped_timer_us));
317 : }
318 :
319 : inline void
320 332654 : select_scheduler::run_task(lock_type& lock, context_type& ctx, long timeout_us)
321 : {
322 : long effective_timeout_us =
323 332654 : task_interrupted_ ? 0 : calculate_timeout(timeout_us);
324 :
325 : // Snapshot registered descriptors while holding lock.
326 : // Record which fds need write monitoring to avoid a hot loop:
327 : // select is level-triggered so writable sockets (nearly always
328 : // writable) would cause select() to return immediately every
329 : // iteration if unconditionally added to write_fds. Membership
330 : // stays opt-in: a parked write wait opts in the same way a
331 : // parked write or connect op does.
332 : struct fd_entry
333 : {
334 : int fd;
335 : reactor_descriptor_state* desc;
336 : bool needs_write;
337 : };
338 : fd_entry snapshot[FD_SETSIZE];
339 332654 : int snapshot_count = 0;
340 :
341 866974 : for (auto& [fd, desc] : registered_descs_)
342 : {
343 534320 : if (snapshot_count < FD_SETSIZE)
344 : {
345 534320 : conditionally_enabled_mutex::scoped_lock desc_lock(desc->mutex);
346 534320 : snapshot[snapshot_count].fd = fd;
347 534320 : snapshot[snapshot_count].desc = desc;
348 534320 : snapshot[snapshot_count].needs_write =
349 534320 : (desc->write_op || desc->connect_op || desc->wait_write_op);
350 534320 : ++snapshot_count;
351 534320 : }
352 : }
353 :
354 332654 : if (lock.owns_lock())
355 308537 : lock.unlock();
356 :
357 332654 : task_cleanup on_exit{this, &lock, ctx};
358 :
359 : fd_set read_fds, write_fds, except_fds;
360 5655118 : FD_ZERO(&read_fds);
361 5655118 : FD_ZERO(&write_fds);
362 5655118 : FD_ZERO(&except_fds);
363 :
364 332654 : FD_SET(pipe_fds_[0], &read_fds);
365 332654 : int nfds = pipe_fds_[0];
366 :
367 866974 : for (int i = 0; i < snapshot_count; ++i)
368 : {
369 534320 : int fd = snapshot[i].fd;
370 534320 : FD_SET(fd, &read_fds);
371 534320 : if (snapshot[i].needs_write)
372 7948 : FD_SET(fd, &write_fds);
373 534320 : FD_SET(fd, &except_fds);
374 534320 : if (fd > nfds)
375 332043 : nfds = fd;
376 : }
377 :
378 : struct timeval tv;
379 332654 : struct timeval* tv_ptr = nullptr;
380 332654 : if (effective_timeout_us >= 0)
381 : {
382 331877 : tv.tv_sec = effective_timeout_us / 1000000;
383 331877 : tv.tv_usec = effective_timeout_us % 1000000;
384 331877 : tv_ptr = &tv;
385 : }
386 :
387 332654 : int ready = ::select(nfds + 1, &read_fds, &write_fds, &except_fds, tv_ptr);
388 :
389 : // EINTR: signal interrupted select(), just retry.
390 : // EBADF: an fd was closed between snapshot and select(); retry
391 : // with a fresh snapshot from registered_descs_.
392 : // Both fall through with no ready descriptors rather than
393 : // returning: the caller handed this function an owned lock that
394 : // only the epilogue below re-acquires.
395 332654 : if (ready < 0)
396 : {
397 3 : if (errno != EINTR && errno != EBADF)
398 1 : detail::throw_system_error(make_err(errno), "select");
399 2 : ready = 0;
400 : }
401 :
402 : // Process timers outside the lock
403 332653 : timer_svc_->process_expired();
404 :
405 332653 : ready_queue local_ops;
406 :
407 332653 : if (ready > 0)
408 : {
409 316405 : if (FD_ISSET(pipe_fds_[0], &read_fds))
410 : {
411 : char buf[256];
412 11654 : while (::read(pipe_fds_[0], buf, sizeof(buf)) > 0)
413 : {
414 : }
415 : }
416 :
417 804357 : for (int i = 0; i < snapshot_count; ++i)
418 : {
419 487952 : int fd = snapshot[i].fd;
420 487952 : reactor_descriptor_state* desc = snapshot[i].desc;
421 :
422 487952 : std::uint32_t flags = 0;
423 487952 : if (FD_ISSET(fd, &read_fds))
424 315227 : flags |= reactor_event_read;
425 487952 : if (FD_ISSET(fd, &write_fds))
426 2154 : flags |= reactor_event_write;
427 487952 : if (FD_ISSET(fd, &except_fds))
428 16 : flags |= reactor_event_error;
429 :
430 487952 : if (flags == 0)
431 170580 : continue;
432 :
433 317372 : desc->add_ready_events(flags);
434 :
435 317372 : bool expected = false;
436 317372 : if (desc->is_enqueued_.compare_exchange_strong(
437 : expected, true, std::memory_order_release,
438 : std::memory_order_relaxed))
439 : {
440 317372 : local_ops.push(desc);
441 : }
442 : }
443 : }
444 :
445 332653 : lock.lock();
446 :
447 332653 : completed_ops_.splice(local_ops);
448 332654 : }
449 :
450 : } // namespace boost::corosio::detail
451 :
452 : #endif // BOOST_COROSIO_HAS_SELECT
453 :
454 : #endif // BOOST_COROSIO_NATIVE_DETAIL_SELECT_SELECT_SCHEDULER_HPP
|
