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

99.4% Lines (162 / 163) 100.0% Functions (12 / 12)

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

epoll_scheduler.hpp

f(x) Functions (12)

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
28		#include <boost/corosio/detail/except.hpp>
29
30		#include <atomic>
31		#include <chrono>
32		#include <cstdint>
33		#include <mutex>
34		#include <vector>
35
36		#include <errno.h>
37		#include <sys/epoll.h>
38		#include <sys/eventfd.h>
39		#include <sys/timerfd.h>
40		#include <unistd.h>
41
42		namespace boost::corosio::detail {
43
44		/** Linux scheduler using epoll for I/O multiplexing.
45
46		This scheduler implements the scheduler interface using Linux epoll
47		for efficient I/O event notification. It uses a single reactor model
48		where one thread runs epoll_wait while other threads
49		wait on a condition variable for handler work. This design provides:
50
51		- Handler parallelism: N posted handlers can execute on N threads
52		- No thundering herd: condition_variable wakes exactly one thread
53		- IOCP parity: Behavior matches Windows I/O completion port semantics
54
55		When threads call run(), they first try to execute queued handlers.
56		If the queue is empty and no reactor is running, one thread becomes
57		the reactor and runs epoll_wait. Other threads wait on a condition
58		variable until handlers are available.
59
60		@par Thread Safety
61		All public member functions are thread-safe.
62		*/
63		class BOOST_COROSIO_DECL epoll_scheduler final : public reactor_scheduler
64		{
65		public:
66		/** Construct the scheduler.
67
68		Creates an epoll instance, eventfd for reactor interruption,
69		and timerfd for kernel-managed timer expiry.
70
71		@param ctx Reference to the owning execution_context.
72		@param concurrency_hint Hint for expected thread count (unused).
73		*/
74		epoll_scheduler(capy::execution_context& ctx, int concurrency_hint = -1);
75
76		/// Destroy the scheduler.
77		~epoll_scheduler() override;
78
79		epoll_scheduler(epoll_scheduler const&) = delete;
80		epoll_scheduler& operator=(epoll_scheduler const&) = delete;
81
82		/// Shut down the scheduler, draining pending operations.
83		void shutdown() override;
84
85		/// Apply runtime configuration, resizing the event buffer.
86		void configure_reactor(
87		unsigned max_events,
88		unsigned budget_init,
89		unsigned budget_max,
90		unsigned unassisted) override;
91
92		/** Return the epoll file descriptor.
93
94		Used by socket services to register file descriptors
95		for I/O event notification.
96
97		@return The epoll file descriptor.
98		*/
99		int epoll_fd() const noexcept
100		{
101		return epoll_fd_;
102		}
103
104		/** Register a descriptor for persistent monitoring.
105
106		The fd is registered once and stays registered until explicitly
107		deregistered. Events are dispatched via reactor_descriptor_state which
108		tracks pending read/write/connect operations.
109
110		@param fd The file descriptor to register.
111		@param desc Pointer to descriptor data (stored in epoll_event.data.ptr).
112
113		@return The error if registration fails, otherwise a default
114		constructed error code.
115		*/
116		std::error_code
117		register_descriptor(int fd, reactor_descriptor_state* desc) const;
118
119		/// No-op: write readiness is watched from registration on.
120		std::error_code
121	5462x	ensure_write_registered(int, reactor_descriptor_state*) const noexcept
122		{
123	5462x	return {};
124		}
125
126		/** Deregister a persistently registered descriptor.
127
128		@param fd The file descriptor to deregister.
129		*/
130		void deregister_descriptor(int fd) const;
131
132		/// Watch the read end of the POSIX signal self-pipe (see scheduler.hpp).
133	285x	[[nodiscard]] std::error_code register_signal_reader(int read_fd) override
134		{
135	285x	return register_descriptor(read_fd, signal_pipe_reader_.arm());
136		}
137
138		private:
139		void run_task(lock_type& lock, context_type& ctx, 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	2474x	inline epoll_scheduler::epoll_scheduler(capy::execution_context& ctx, int)
163	2474x	: epoll_fd_(-1)
164	2474x	, event_fd_(-1)
165	2474x	, timer_fd_(-1)
166	4948x	, event_buffer_(max_events_per_poll_)
167		{
168	2474x	epoll_fd_ = ::epoll_create1(EPOLL_CLOEXEC);
169	2474x	if (epoll_fd_ < 0)
170	1x	detail::throw_system_error(make_err(errno), "epoll_create1");
171
172	2473x	event_fd_ = ::eventfd(0, EFD_NONBLOCK \| EFD_CLOEXEC);
173	2473x	if (event_fd_ < 0)
174		{
175	1x	int errn = errno;
176	1x	::close(epoll_fd_);
177	1x	detail::throw_system_error(make_err(errn), "eventfd");
178		}
179
180	2472x	timer_fd_ = ::timerfd_create(CLOCK_MONOTONIC, TFD_NONBLOCK \| TFD_CLOEXEC);
181	2472x	if (timer_fd_ < 0)
182		{
183	1x	int errn = errno;
184	1x	::close(event_fd_);
185	1x	::close(epoll_fd_);
186	1x	detail::throw_system_error(make_err(errn), "timerfd_create");
187		}
188
189	2471x	epoll_event ev{};
190	2471x	ev.events = EPOLLIN \| EPOLLET;
191	2471x	ev.data.ptr = nullptr;
192	2471x	if (::epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, event_fd_, &ev) < 0)
193		{
194	1x	int errn = errno;
195	1x	::close(timer_fd_);
196	1x	::close(event_fd_);
197	1x	::close(epoll_fd_);
198	1x	detail::throw_system_error(make_err(errn), "epoll_ctl");
199		}
200
201	2470x	epoll_event timer_ev{};
202	2470x	timer_ev.events = EPOLLIN \| EPOLLERR;
203	2470x	timer_ev.data.ptr = &timer_fd_;
204	2470x	if (::epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, timer_fd_, &timer_ev) < 0)
205		{
206	1x	int errn = errno;
207	1x	::close(timer_fd_);
208	1x	::close(event_fd_);
209	1x	::close(epoll_fd_);
210	1x	detail::throw_system_error(make_err(errn), "epoll_ctl (timerfd)");
211		}
212
213	2469x	timer_svc_ = &get_timer_service(ctx, *this);
214	2469x	timer_svc_->set_on_earliest_changed(
215	7462x	timer_service::callback(this, [](void* p) {
216	4993x	auto* self = static_cast<epoll_scheduler*>(p);
217	4993x	self->timerfd_stale_.store(true, std::memory_order_release);
218	4993x	self->interrupt_reactor();
219	4993x	}));
220
221	2469x	completed_ops_.push(&task_op_);
222	2484x	}
223
224	4938x	inline epoll_scheduler::~epoll_scheduler()
225		{
226	2469x	if (timer_fd_ >= 0)
227	2469x	::close(timer_fd_);
228	2469x	if (event_fd_ >= 0)
229	2469x	::close(event_fd_);
230	2469x	if (epoll_fd_ >= 0)
231	2469x	::close(epoll_fd_);
232	4938x	}
233
234		inline void
235	2469x	epoll_scheduler::shutdown()
236		{
237	2469x	shutdown_drain();
238
239	2469x	if (event_fd_ >= 0)
240	2469x	interrupt_reactor();
241	2469x	}
242
243		inline void
244	32x	epoll_scheduler::configure_reactor(
245		unsigned max_events,
246		unsigned budget_init,
247		unsigned budget_max,
248		unsigned unassisted)
249		{
250	32x	reactor_scheduler::configure_reactor(
251		max_events, budget_init, budget_max, unassisted);
252	30x	event_buffer_.resize(max_events_per_poll_);
253	30x	}
254
255		inline std::error_code
256	13273x	epoll_scheduler::register_descriptor(
257		int fd, reactor_descriptor_state* desc) const
258		{
259	13273x	epoll_event ev{};
260	13273x	ev.events = EPOLLIN \| EPOLLOUT \| EPOLLET \| EPOLLERR \| EPOLLHUP;
261	13273x	ev.data.ptr = desc;
262
263		// Initialized before the add: another run thread's reactor can
264		// dispatch an event for the descriptor as soon as it is added.
265		// A completion queued by an earlier registration of the same state
266		// may still be running, so the fields it reads change under its
267		// mutex, and the mutex mode is only written when it differs.
268	13273x	if (desc->mutex.enabled() != reactor_io_locking_)
269	3x	desc->mutex.set_enabled(reactor_io_locking_);
270		{
271	13273x	conditionally_enabled_mutex::scoped_lock lock(desc->mutex);
272	13273x	desc->registered_events = ev.events;
273	13273x	desc->unpollable = false;
274	13273x	desc->fd = fd;
275	13273x	desc->scheduler_ = this;
276	13273x	desc->ready_events_.store(0, std::memory_order_release);
277		// object_ref_ is not touched: a queued invocation from an
278		// earlier registration still owns the reference in it.
279	13273x	desc->read_ready = false;
280	13273x	desc->write_ready = false;
281	13273x	}
282
283	13273x	if (::epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, fd, &ev) < 0)
284		{
285	12x	int const err = errno;
286	12x	conditionally_enabled_mutex::scoped_lock lock(desc->mutex);
287	12x	desc->registered_events = 0;
288		// EPERM: a file type epoll cannot watch. Its I/O does not
289		// block, so adopt it unwatched, as asio does.
290	12x	if (err != EPERM)
291	7x	return make_err(err);
292	5x	desc->unpollable = true;
293	12x	}
294	13266x	return {};
295		}
296
297		inline void
298	12977x	epoll_scheduler::deregister_descriptor(int fd) const
299		{
300	12977x	::epoll_ctl(epoll_fd_, EPOLL_CTL_DEL, fd, nullptr);
301	12977x	}
302
303		inline void
304	13104x	epoll_scheduler::interrupt_reactor() const
305		{
306	13104x	bool expected = false;
307	13104x	if (eventfd_armed_.compare_exchange_strong(
308		expected, true, std::memory_order_release,
309		std::memory_order_relaxed))
310		{
311	10904x	std::uint64_t val = 1;
312	10904x	if (::write(event_fd_, &val, sizeof(val)) < 0)
313		{
314		// The flag is what coalesces later interrupts into a byte
315		// already in the eventfd; a write that failed put no byte
316		// there, so leaving it armed would swallow every interrupt
317		// that follows. Disarming keeps the cost to the interrupts
318		// already in flight -- the next one arms and writes again,
319		// instead of every one after this coalescing into a byte
320		// that does not exist.
321	2x	eventfd_armed_.store(false, std::memory_order_release);
322		}
323		}
324	13104x	}
325
326		inline void
327	12436x	epoll_scheduler::update_timerfd() const
328		{
329	12436x	auto nearest = timer_svc_->nearest_expiry();
330
331	12436x	itimerspec ts{};
332	12436x	int flags = 0;
333
334	12436x	if (nearest == timer_service::time_point::max())
335		{
336		// No timers — disarm by setting to 0 (relative)
337		}
338		else
339		{
340	11004x	auto now = std::chrono::steady_clock::now();
341	11004x	if (nearest <= now)
342		{
343		// Use 1ns instead of 0 — zero disarms the timerfd
344	810x	ts.it_value.tv_nsec = 1;
345		}
346		else
347		{
348	10194x	auto nsec = std::chrono::duration_cast<std::chrono::nanoseconds>(
349	10194x	nearest - now)
350	10194x	.count();
351	10194x	ts.it_value.tv_sec = nsec / 1000000000;
352	10194x	ts.it_value.tv_nsec = nsec % 1000000000;
353	10194x	if (ts.it_value.tv_sec == 0 && ts.it_value.tv_nsec == 0)
354	✗	ts.it_value.tv_nsec = 1;
355		}
356		}
357
358	12436x	if (::timerfd_settime(timer_fd_, flags, &ts, nullptr) < 0)
359	1x	detail::throw_system_error(make_err(errno), "timerfd_settime");
360	12435x	}
361
362		inline void
363	155252x	epoll_scheduler::run_task(lock_type& lock, context_type& ctx, long timeout_us)
364		{
365		int timeout_ms;
366	155252x	if (task_interrupted_)
367	136656x	timeout_ms = 0;
368	18596x	else if (timeout_us < 0)
369	17958x	timeout_ms = -1;
370		else
371	638x	timeout_ms = static_cast<int>((timeout_us + 999) / 1000);
372
373	155252x	if (lock.owns_lock())
374	18600x	lock.unlock();
375
376	155252x	task_cleanup on_exit{this, &lock, ctx};
377
378		// Flush deferred timerfd programming before blocking
379	155252x	if (timerfd_stale_.exchange(false, std::memory_order_acquire))
380	4231x	update_timerfd();
381
382	155251x	int nfds = ::epoll_wait(
383	155251x	epoll_fd_, event_buffer_.data(), static_cast<int>(event_buffer_.size()),
384		timeout_ms);
385
386	155251x	if (nfds < 0 && errno != EINTR)
387	1x	detail::throw_system_error(make_err(errno), "epoll_wait");
388
389	155250x	bool check_timers = false;
390	155250x	ready_queue local_ops;
391
392	219120x	for (int i = 0; i < nfds; ++i)
393		{
394	63870x	if (event_buffer_[i].data.ptr == nullptr)
395		{
396		std::uint64_t val;
397		// NOLINTNEXTLINE(clang-analyzer-unix.BlockInCriticalSection)
398	8433x	[[maybe_unused]] auto r = ::read(event_fd_, &val, sizeof(val));
399	8433x	eventfd_armed_.store(false, std::memory_order_relaxed);
400	8433x	continue;
401	8433x	}
402
403	55437x	if (event_buffer_[i].data.ptr == &timer_fd_)
404		{
405		std::uint64_t expirations;
406		// NOLINTNEXTLINE(clang-analyzer-unix.BlockInCriticalSection)
407		[[maybe_unused]] auto r =
408	8205x	::read(timer_fd_, &expirations, sizeof(expirations));
409	8205x	check_timers = true;
410	8205x	continue;
411	8205x	}
412
413		auto* desc =
414	47232x	static_cast<reactor_descriptor_state*>(event_buffer_[i].data.ptr);
415
416		// EPOLLHUP maps to no reactor_event_* bit, so a HUP the widening
417		// below did not translate would take no branch in
418		// invoke_deferred_io() and, the registration being
419		// edge-triggered, never get another chance.
420		//
421		// HUP without OUT means a non-socket: a pipe or tty whose peer
422		// closed, with or without data still unread (HUP alone, or
423		// IN\|HUP). Treat it as readable, writable and faulted, the way
424		// poll() and io_uring report it, so a parked error wait
425		// completes too. Sockets carry at least OUT with HUP (a fresh
426		// unconnected stream socket reports HUP\|OUT), so they take the
427		// plain widening, where forcing IN costs at most one spurious
428		// EAGAIN.
429	47232x	std::uint32_t ev = event_buffer_[i].events;
430	47232x	if ((ev & (EPOLLHUP \| EPOLLOUT)) == EPOLLHUP)
431	8x	ev \|= EPOLLIN \| EPOLLOUT \| EPOLLERR;
432	47224x	else if (ev & EPOLLHUP)
433	2798x	ev \|= EPOLLIN \| EPOLLOUT;
434	47232x	desc->add_ready_events(ev);
435
436	47232x	bool expected = false;
437	47232x	if (desc->is_enqueued_.compare_exchange_strong(
438		expected, true, std::memory_order_release,
439		std::memory_order_relaxed))
440		{
441	47228x	local_ops.push(desc);
442		}
443		}
444
445	155250x	if (check_timers)
446		{
447	8205x	timer_svc_->process_expired();
448	8205x	update_timerfd();
449		}
450
451	155250x	lock.lock();
452
453	155250x	completed_ops_.splice(local_ops);
454	155252x	}
455
456		} // namespace boost::corosio::detail
457
458		#endif // BOOST_COROSIO_HAS_EPOLL
459
460		#endif // BOOST_COROSIO_NATIVE_DETAIL_EPOLL_EPOLL_SCHEDULER_HPP
461