include/boost/corosio/native/detail/epoll/epoll_scheduler.hpp

86.2% Lines (131/0/152) 100.0% List of functions (11/0/11)

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

epoll_scheduler.hpp

f(x) Functions (11)

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_EPOLL_EPOLL_SCHEDULER_HPP
12		#define BOOST_COROSIO_NATIVE_DETAIL_EPOLL_EPOLL_SCHEDULER_HPP
13
14		#include <boost/corosio/detail/platform.hpp>
15
16		#if BOOST_COROSIO_HAS_EPOLL
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/epoll/epoll_traits.hpp>
25		#include <boost/corosio/detail/timer_service.hpp>
26		#include <boost/corosio/native/detail/make_err.hpp>
27		#include <boost/corosio/native/detail/posix/posix_resolver_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>
30		#include <boost/corosio/native/detail/posix/posix_random_access_file_service.hpp>
31
32		#include <boost/corosio/detail/except.hpp>
33
34		#include <atomic>
35		#include <chrono>
36		#include <cstdint>
37		#include <mutex>
38		#include <vector>
39
40		#include <errno.h>
41		#include <sys/epoll.h>
42		#include <sys/eventfd.h>
43		#include <sys/timerfd.h>
44		#include <unistd.h>
45
46		namespace boost::corosio::detail {
47
48		/** Linux scheduler using epoll for I/O multiplexing.
49
50		This scheduler implements the scheduler interface using Linux epoll
51		for efficient I/O event notification. It uses a single reactor model
52		where one thread runs epoll_wait while other threads
53		wait on a condition variable for handler work. This design provides:
54
55		- Handler parallelism: N posted handlers can execute on N threads
56		- No thundering herd: condition_variable wakes exactly one thread
57		- IOCP parity: Behavior matches Windows I/O completion port semantics
58
59		When threads call run(), they first try to execute queued handlers.
60		If the queue is empty and no reactor is running, one thread becomes
61		the reactor and runs epoll_wait. Other threads wait on a condition
62		variable until handlers are available.
63
64		@par Thread Safety
65		All public member functions are thread-safe.
66		*/
67		class BOOST_COROSIO_DECL epoll_scheduler final : public reactor_scheduler
68		{
69		public:
70		/** Construct the scheduler.
71
72		Creates an epoll instance, eventfd for reactor interruption,
73		and timerfd for kernel-managed timer expiry.
74
75		@param ctx Reference to the owning execution_context.
76		@param concurrency_hint Hint for expected thread count (unused).
77		*/
78		epoll_scheduler(capy::execution_context& ctx, int concurrency_hint = -1);
79
80		/// Destroy the scheduler.
81		~epoll_scheduler() override;
82
83		epoll_scheduler(epoll_scheduler const&) = delete;
84		epoll_scheduler& operator=(epoll_scheduler const&) = delete;
85
86		/// Shut down the scheduler, draining pending operations.
87		void shutdown() override;
88
89		/// Apply runtime configuration, resizing the event buffer.
90		void configure_reactor(
91		unsigned max_events,
92		unsigned budget_init,
93		unsigned budget_max,
94		unsigned unassisted) override;
95
96		/** Return the epoll file descriptor.
97
98		Used by socket services to register file descriptors
99		for I/O event notification.
100
101		@return The epoll file descriptor.
102		*/
103		int epoll_fd() const noexcept
104		{
105		return epoll_fd_;
106		}
107
108		/** Register a descriptor for persistent monitoring.
109
110		The fd is registered once and stays registered until explicitly
111		deregistered. Events are dispatched via reactor_descriptor_state which
112		tracks pending read/write/connect operations.
113
114		@param fd The file descriptor to register.
115		@param desc Pointer to descriptor data (stored in epoll_event.data.ptr).
116
117		@return The error if registration fails, otherwise a default
118		constructed error code.
119		*/
120		std::error_code
121		register_descriptor(int fd, reactor_descriptor_state* desc) const;
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		/// Watch the read end of the POSIX signal self-pipe (see scheduler.hpp).
130		[[nodiscard]] std::error_code
131	51x	register_signal_reader(int read_fd) override
132		{
133	51x	return register_descriptor(read_fd, signal_pipe_reader_.arm());
134		}
135
136		private:
137		void
138		run_task(lock_type& lock, context_type* ctx,
139		long timeout_us) override;
140		void interrupt_reactor() const override;
141		void update_timerfd() const;
142
143		int epoll_fd_;
144		int event_fd_;
145		int timer_fd_;
146
147		// Watches the global signal self-pipe's read end (armed lazily by
148		// register_signal_reader on the first signal registration).
149		reactor_signal_pipe_reader signal_pipe_reader_;
150
151		// Edge-triggered eventfd state
152		mutable std::atomic<bool> eventfd_armed_{false};
153
154		// Set when the earliest timer changes; flushed before epoll_wait
155		mutable std::atomic<bool> timerfd_stale_{false};
156
157		// Event buffer sized from max_events_per_poll_ (set at construction,
158		// resized by configure_reactor via io_context_options).
159		std::vector<epoll_event> event_buffer_;
160		};
161
162	1247x	inline epoll_scheduler::epoll_scheduler(capy::execution_context& ctx, int)
163	1247x	: epoll_fd_(-1)
164	1247x	, event_fd_(-1)
165	1247x	, timer_fd_(-1)
166	2494x	, event_buffer_(max_events_per_poll_)
167		{
168	1247x	epoll_fd_ = ::epoll_create1(EPOLL_CLOEXEC);
169	1247x	if (epoll_fd_ < 0)
170	✗	detail::throw_system_error(make_err(errno), "epoll_create1");
171
172	1247x	event_fd_ = ::eventfd(0, EFD_NONBLOCK \| EFD_CLOEXEC);
173	1247x	if (event_fd_ < 0)
174		{
175	✗	int errn = errno;
176	✗	::close(epoll_fd_);
177	✗	detail::throw_system_error(make_err(errn), "eventfd");
178		}
179
180	1247x	timer_fd_ = ::timerfd_create(CLOCK_MONOTONIC, TFD_NONBLOCK \| TFD_CLOEXEC);
181	1247x	if (timer_fd_ < 0)
182		{
183	✗	int errn = errno;
184	✗	::close(event_fd_);
185	✗	::close(epoll_fd_);
186	✗	detail::throw_system_error(make_err(errn), "timerfd_create");
187		}
188
189	1247x	epoll_event ev{};
190	1247x	ev.events = EPOLLIN \| EPOLLET;
191	1247x	ev.data.ptr = nullptr;
192	1247x	if (::epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, event_fd_, &ev) < 0)
193		{
194	✗	int errn = errno;
195	✗	::close(timer_fd_);
196	✗	::close(event_fd_);
197	✗	::close(epoll_fd_);
198	✗	detail::throw_system_error(make_err(errn), "epoll_ctl");
199		}
200
201	1247x	epoll_event timer_ev{};
202	1247x	timer_ev.events = EPOLLIN \| EPOLLERR;
203	1247x	timer_ev.data.ptr = &timer_fd_;
204	1247x	if (::epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, timer_fd_, &timer_ev) < 0)
205		{
206	✗	int errn = errno;
207	✗	::close(timer_fd_);
208	✗	::close(event_fd_);
209	✗	::close(epoll_fd_);
210	✗	detail::throw_system_error(make_err(errn), "epoll_ctl (timerfd)");
211		}
212
213	1247x	timer_svc_ = &get_timer_service(ctx, *this);
214	1247x	timer_svc_->set_on_earliest_changed(
215	6740x	timer_service::callback(this, [](void* p) {
216	5493x	auto* self = static_cast<epoll_scheduler*>(p);
217	5493x	self->timerfd_stale_.store(true, std::memory_order_release);
218	5493x	self->interrupt_reactor();
219	5493x	}));
220
221	1247x	get_resolver_service(ctx, *this);
222	1247x	get_signal_service(ctx, *this);
223	1247x	get_stream_file_service(ctx, *this);
224	1247x	get_random_access_file_service(ctx, *this);
225
226	1247x	completed_ops_.push(&task_op_);
227	1247x	}
228
229	2494x	inline epoll_scheduler::~epoll_scheduler()
230		{
231	1247x	if (timer_fd_ >= 0)
232	1247x	::close(timer_fd_);
233	1247x	if (event_fd_ >= 0)
234	1247x	::close(event_fd_);
235	1247x	if (epoll_fd_ >= 0)
236	1247x	::close(epoll_fd_);
237	2494x	}
238
239		inline void
240	1247x	epoll_scheduler::shutdown()
241		{
242	1247x	shutdown_drain();
243
244	1247x	if (event_fd_ >= 0)
245	1247x	interrupt_reactor();
246	1247x	}
247
248		inline void
249	21x	epoll_scheduler::configure_reactor(
250		unsigned max_events,
251		unsigned budget_init,
252		unsigned budget_max,
253		unsigned unassisted)
254		{
255	21x	reactor_scheduler::configure_reactor(
256		max_events, budget_init, budget_max, unassisted);
257	20x	event_buffer_.resize(max_events_per_poll_);
258	20x	}
259
260		inline std::error_code
261	12746x	epoll_scheduler::register_descriptor(int fd, reactor_descriptor_state* desc) const
262		{
263	12746x	epoll_event ev{};
264	12746x	ev.events = EPOLLIN \| EPOLLOUT \| EPOLLET \| EPOLLERR \| EPOLLHUP;
265	12746x	ev.data.ptr = desc;
266
267	12746x	if (::epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, fd, &ev) < 0)
268	1x	return make_err(errno);
269
270	12745x	desc->registered_events = ev.events;
271	12745x	desc->fd = fd;
272	12745x	desc->scheduler_ = this;
273	12745x	desc->mutex.set_enabled(reactor_io_locking_);
274	12745x	desc->ready_events_.store(0, std::memory_order_relaxed);
275
276	12745x	conditionally_enabled_mutex::scoped_lock lock(desc->mutex);
277	12745x	desc->impl_ref_.reset();
278	12745x	desc->read_ready = false;
279	12745x	desc->write_ready = false;
280	12745x	return {};
281	12745x	}
282
283		inline void
284	12694x	epoll_scheduler::deregister_descriptor(int fd) const
285		{
286	12694x	::epoll_ctl(epoll_fd_, EPOLL_CTL_DEL, fd, nullptr);
287	12694x	}
288
289		inline void
290	11287x	epoll_scheduler::interrupt_reactor() const
291		{
292	11287x	bool expected = false;
293	11287x	if (eventfd_armed_.compare_exchange_strong(
294		expected, true, std::memory_order_release,
295		std::memory_order_relaxed))
296		{
297	9718x	std::uint64_t val = 1;
298	9718x	[[maybe_unused]] auto r = ::write(event_fd_, &val, sizeof(val));
299		}
300	11287x	}
301
302		inline void
303	8604x	epoll_scheduler::update_timerfd() const
304		{
305	8604x	auto nearest = timer_svc_->nearest_expiry();
306
307	8604x	itimerspec ts{};
308	8604x	int flags = 0;
309
310	8604x	if (nearest == timer_service::time_point::max())
311		{
312		// No timers — disarm by setting to 0 (relative)
313		}
314		else
315		{
316	8426x	auto now = std::chrono::steady_clock::now();
317	8426x	if (nearest <= now)
318		{
319		// Use 1ns instead of 0 — zero disarms the timerfd
320	152x	ts.it_value.tv_nsec = 1;
321		}
322		else
323		{
324	8274x	auto nsec = std::chrono::duration_cast<std::chrono::nanoseconds>(
325	8274x	nearest - now)
326	8274x	.count();
327	8274x	ts.it_value.tv_sec = nsec / 1000000000;
328	8274x	ts.it_value.tv_nsec = nsec % 1000000000;
329	8274x	if (ts.it_value.tv_sec == 0 && ts.it_value.tv_nsec == 0)
330	✗	ts.it_value.tv_nsec = 1;
331		}
332		}
333
334	8604x	if (::timerfd_settime(timer_fd_, flags, &ts, nullptr) < 0)
335	✗	detail::throw_system_error(make_err(errno), "timerfd_settime");
336	8604x	}
337
338		inline void
339	58212x	epoll_scheduler::run_task(
340		lock_type& lock, context_type* ctx, long timeout_us)
341		{
342		int timeout_ms;
343	58212x	if (task_interrupted_)
344	44283x	timeout_ms = 0;
345	13929x	else if (timeout_us < 0)
346	13925x	timeout_ms = -1;
347		else
348	4x	timeout_ms = static_cast<int>((timeout_us + 999) / 1000);
349
350	58212x	if (lock.owns_lock())
351	13931x	lock.unlock();
352
353	58212x	task_cleanup on_exit{this, &lock, ctx};
354
355		// Flush deferred timerfd programming before blocking
356	58212x	if (timerfd_stale_.exchange(false, std::memory_order_acquire))
357	4839x	update_timerfd();
358
359	58212x	int nfds = ::epoll_wait(
360		epoll_fd_, event_buffer_.data(),
361	58212x	static_cast<int>(event_buffer_.size()), timeout_ms);
362
363	58212x	if (nfds < 0 && errno != EINTR)
364	✗	detail::throw_system_error(make_err(errno), "epoll_wait");
365
366	58212x	bool check_timers = false;
367	58212x	ready_queue local_ops;
368
369	120283x	for (int i = 0; i < nfds; ++i)
370		{
371	62071x	if (event_buffer_[i].data.ptr == nullptr)
372		{
373		std::uint64_t val;
374		// NOLINTNEXTLINE(clang-analyzer-unix.BlockInCriticalSection)
375	8471x	[[maybe_unused]] auto r = ::read(event_fd_, &val, sizeof(val));
376	8471x	eventfd_armed_.store(false, std::memory_order_relaxed);
377	8471x	continue;
378	8471x	}
379
380	53600x	if (event_buffer_[i].data.ptr == &timer_fd_)
381		{
382		std::uint64_t expirations;
383		// NOLINTNEXTLINE(clang-analyzer-unix.BlockInCriticalSection)
384		[[maybe_unused]] auto r =
385	3765x	::read(timer_fd_, &expirations, sizeof(expirations));
386	3765x	check_timers = true;
387	3765x	continue;
388	3765x	}
389
390		auto* desc =
391	49835x	static_cast<reactor_descriptor_state*>(event_buffer_[i].data.ptr);
392	49835x	desc->add_ready_events(event_buffer_[i].events);
393
394	49835x	bool expected = false;
395	49835x	if (desc->is_enqueued_.compare_exchange_strong(
396		expected, true, std::memory_order_release,
397		std::memory_order_relaxed))
398		{
399	49835x	local_ops.push(desc);
400		}
401		}
402
403	58212x	if (check_timers)
404		{
405	3765x	timer_svc_->process_expired();
406	3765x	update_timerfd();
407		}
408
409	58212x	lock.lock();
410
411	58212x	completed_ops_.splice(local_ops);
412	58212x	}
413
414		} // namespace boost::corosio::detail
415
416		#endif // BOOST_COROSIO_HAS_EPOLL
417
418		#endif // BOOST_COROSIO_NATIVE_DETAIL_EPOLL_EPOLL_SCHEDULER_HPP
419