include/boost/corosio/native/detail/select/select_scheduler.hpp

99.4% Lines (178 / 179, 1 excl) 100.0% Functions (13 / 13)

[ ] prev/next file n p prev/next uncovered line

select_scheduler.hpp

f(x) Functions (13)

Function Calls Lines Blocks

Line	Hits	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	2768x	ensure_write_registered(int, reactor_descriptor_state*) const noexcept
119		{
120	2768x	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	271x	[[nodiscard]] std::error_code register_signal_reader(int read_fd) override
139		{
140	271x	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	482x	void notify_reparked() const override
149		{
150	482x	interrupt_reactor();
151	482x	}
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	1952x	inline select_scheduler::select_scheduler(capy::execution_context& ctx, int)
168	1952x	: pipe_fds_{-1, -1}
169	1952x	, max_fd_(-1)
170		{
171	1952x	if (::pipe(pipe_fds_) < 0)
172	1x	detail::throw_system_error(make_err(errno), "pipe");
173
174	5844x	for (int i = 0; i < 2; ++i)
175		{
176	3899x	int flags = ::fcntl(pipe_fds_[i], F_GETFL, 0);
177	3899x	if (flags == -1)
178		{
179	2x	int errn = errno;
180	2x	::close(pipe_fds_[0]);
181	2x	::close(pipe_fds_[1]);
182	2x	detail::throw_system_error(make_err(errn), "fcntl F_GETFL");
183		}
184	3897x	if (::fcntl(pipe_fds_[i], F_SETFL, flags \| O_NONBLOCK) == -1)
185		{
186	2x	int errn = errno;
187	2x	::close(pipe_fds_[0]);
188	2x	::close(pipe_fds_[1]);
189	2x	detail::throw_system_error(make_err(errn), "fcntl F_SETFL");
190		}
191	3895x	if (::fcntl(pipe_fds_[i], F_SETFD, FD_CLOEXEC) == -1)
192		{
193	2x	int errn = errno;
194	2x	::close(pipe_fds_[0]);
195	2x	::close(pipe_fds_[1]);
196	2x	detail::throw_system_error(make_err(errn), "fcntl F_SETFD");
197		}
198		}
199
200	1945x	timer_svc_ = &get_timer_service(ctx, *this);
201	1945x	timer_svc_->set_on_earliest_changed(
202	5160x	timer_service::callback(this, [](void* p) {
203	3215x	static_cast<select_scheduler*>(p)->interrupt_reactor();
204	3215x	}));
205
206	1945x	completed_ops_.push(&task_op_);
207	1966x	}
208
209	3890x	inline select_scheduler::~select_scheduler()
210		{
211	1945x	if (pipe_fds_[0] >= 0)
212	1945x	::close(pipe_fds_[0]);
213	1945x	if (pipe_fds_[1] >= 0)
214	1945x	::close(pipe_fds_[1]);
215	3890x	}
216
217		inline void
218	1945x	select_scheduler::shutdown()
219		{
220	1945x	shutdown_drain();
221
222	1945x	if (pipe_fds_[1] >= 0)
223	1945x	interrupt_reactor();
224	1945x	}
225
226		inline std::error_code
227	10011x	select_scheduler::register_descriptor(
228		int fd, reactor_descriptor_state* desc) const
229		{
230	10011x	if (fd < 0 \|\| fd >= FD_SETSIZE)
231	1x	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	10010x	if (desc->mutex.enabled() != reactor_io_locking_)
237	3x	desc->mutex.set_enabled(reactor_io_locking_);
238		{
239	10010x	conditionally_enabled_mutex::scoped_lock lock(desc->mutex);
240	10010x	desc->registered_events = reactor_event_read \| reactor_event_write;
241	10010x	desc->unpollable = false; // the state is reused across adoptions
242	10010x	desc->fd = fd;
243	10010x	desc->scheduler_ = this;
244	10010x	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	10010x	desc->read_ready = false;
248	10010x	desc->write_ready = false;
249	10010x	}
250
251		{
252	10010x	mutex_type::scoped_lock lock(mutex_);
253		try
254		{
255	10010x	registered_descs_[fd] = desc;
256		}
257	1x	catch (std::bad_alloc const&)
258		{
259	1x	return make_err(ENOMEM);
260	1x	}
261	10009x	if (fd > max_fd_)
262	9499x	max_fd_ = fd;
263	10010x	}
264
265	10009x	interrupt_reactor();
266	10009x	return {};
267		}
268
269		inline void
270	9739x	select_scheduler::deregister_descriptor(int fd) const
271		{
272	9739x	mutex_type::scoped_lock lock(mutex_);
273
274	9739x	auto it = registered_descs_.find(fd);
275	9739x	if (it == registered_descs_.end())
276	✗	return;
277
278	9739x	registered_descs_.erase(it);
279
280	9739x	if (fd == max_fd_)
281		{
282	7796x	max_fd_ = pipe_fds_[0];
283	13038x	for (auto& [registered_fd, state] : registered_descs_)
284		{
285	5242x	if (registered_fd > max_fd_)
286	4979x	max_fd_ = registered_fd;
287		}
288		}
289	9739x	}
290
291		inline void
292	6829x	select_scheduler::notify_reactor() const
293		{
294	6829x	interrupt_reactor();
295	6829x	}
296
297		inline void
298	25599x	select_scheduler::interrupt_reactor() const
299		{
300	25599x	char byte = 1;
301	25599x	[[maybe_unused]] auto r = ::write(pipe_fds_[1], &byte, 1);
302	25599x	}
303
304		inline long
305	81669x	select_scheduler::calculate_timeout(long requested_timeout_us) const
306		{
307	81669x	if (requested_timeout_us == 0)
308	−	return 0; // LCOV_EXCL_LINE run_task passes 0 via task_interrupted_, never through this argument
309
310	81669x	auto nearest = timer_svc_->nearest_expiry();
311	81669x	if (nearest == timer_service::time_point::max())
312	75053x	return requested_timeout_us;
313
314	6616x	auto now = std::chrono::steady_clock::now();
315	6616x	if (nearest <= now)
316	51x	return 0;
317
318		auto timer_timeout_us =
319	6565x	std::chrono::duration_cast<std::chrono::microseconds>(nearest - now)
320	6565x	.count();
321
322	6565x	constexpr auto long_max =
323		static_cast<long long>((std::numeric_limits<long>::max)());
324		auto capped_timer_us =
325	6565x	(std::min)((std::max)(static_cast<long long>(timer_timeout_us),
326	6565x	static_cast<long long>(0)),
327	6565x	long_max);
328
329	6565x	if (requested_timeout_us < 0)
330	6559x	return static_cast<long>(capped_timer_us);
331
332		return static_cast<long>(
333	6x	(std::min)(static_cast<long long>(requested_timeout_us),
334	6x	capped_timer_us));
335		}
336
337		inline void
338	210590x	select_scheduler::run_task(lock_type& lock, context_type& ctx, long timeout_us)
339		{
340		long effective_timeout_us =
341	210590x	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	210590x	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	210590x	if (!lock.owns_lock())
365	128918x	lock.lock();
366
367	537480x	for (auto& [fd, desc] : registered_descs_)
368		{
369	326890x	if (snapshot_count < FD_SETSIZE)
370		{
371	326890x	conditionally_enabled_mutex::scoped_lock desc_lock(desc->mutex);
372	326890x	snapshot[snapshot_count].fd = fd;
373	326890x	snapshot[snapshot_count].desc = desc;
374	326890x	snapshot[snapshot_count].want =
375	326890x	((desc->read_op \|\| desc->wait_read_op) ? reactor_event_read
376	326890x	: 0) \|
377	326510x	((desc->write_op \|\| desc->connect_op \|\| desc->wait_write_op)
378	653400x	? reactor_event_write
379	326890x	: 0) \|
380	326890x	(desc->wait_error_op ? reactor_event_error : 0);
381	326890x	++snapshot_count;
382	326890x	}
383		}
384
385	210590x	if (lock.owns_lock())
386	210589x	lock.unlock();
387
388	210590x	task_cleanup on_exit{this, &lock, ctx};
389
390		fd_set read_fds, write_fds, except_fds;
391	3580030x	FD_ZERO(&read_fds);
392	3580030x	FD_ZERO(&write_fds);
393	3580030x	FD_ZERO(&except_fds);
394
395	210590x	FD_SET(pipe_fds_[0], &read_fds);
396	210590x	int nfds = pipe_fds_[0];
397
398	537480x	for (int i = 0; i < snapshot_count; ++i)
399		{
400	326890x	int fd = snapshot[i].fd;
401	326890x	if (snapshot[i].want & reactor_event_read)
402	183192x	FD_SET(fd, &read_fds);
403	326890x	if (snapshot[i].want & reactor_event_write)
404	3042x	FD_SET(fd, &write_fds);
405	326890x	if (snapshot[i].want != 0)
406	186524x	FD_SET(fd, &except_fds);
407	326890x	if (fd > nfds)
408	208504x	nfds = fd;
409		}
410
411		struct timeval tv;
412	210590x	struct timeval* tv_ptr = nullptr;
413	210590x	if (effective_timeout_us >= 0)
414		{
415	136127x	tv.tv_sec = effective_timeout_us / 1000000;
416	136127x	tv.tv_usec = effective_timeout_us % 1000000;
417	136127x	tv_ptr = &tv;
418		}
419
420	210590x	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	210590x	if (ready < 0)
429		{
430	3x	if (errno != EINTR && errno != EBADF)
431	1x	detail::throw_system_error(make_err(errno), "select");
432	2x	ready = 0;
433		}
434
435		// Process timers outside the lock
436	210589x	timer_svc_->process_expired();
437
438	210589x	ready_queue local_ops;
439
440	210589x	if (ready > 0)
441		{
442	81941x	if (FD_ISSET(pipe_fds_[0], &read_fds))
443		{
444		char buf[256];
445	22377x	while (::read(pipe_fds_[0], buf, sizeof(buf)) > 0)
446		{
447		}
448		}
449
450	236601x	for (int i = 0; i < snapshot_count; ++i)
451		{
452	154660x	int fd = snapshot[i].fd;
453	154660x	reactor_descriptor_state* desc = snapshot[i].desc;
454
455	154660x	std::uint32_t flags = 0;
456	154660x	if (FD_ISSET(fd, &read_fds))
457	73655x	flags \|= reactor_event_read;
458	154660x	if (FD_ISSET(fd, &write_fds))
459	2748x	flags \|= reactor_event_write;
460	154660x	if (FD_ISSET(fd, &except_fds))
461	13x	flags \|= reactor_event_error;
462
463	154660x	if (flags == 0)
464	78255x	continue;
465
466	76405x	desc->add_ready_events(flags);
467
468	76405x	bool expected = false;
469	76405x	if (desc->is_enqueued_.compare_exchange_strong(
470		expected, true, std::memory_order_release,
471		std::memory_order_relaxed))
472		{
473	6356x	local_ops.push(desc);
474		}
475		}
476		}
477
478	210589x	lock.lock();
479
480	210589x	completed_ops_.splice(local_ops);
481	210590x	}
482
483		} // namespace boost::corosio::detail
484
485		#endif // BOOST_COROSIO_HAS_SELECT
486
487		#endif // BOOST_COROSIO_NATIVE_DETAIL_SELECT_SELECT_SCHEDULER_HPP
488