libs/corosio/src/corosio/src/detail/epoll/op.hpp

84.3% Lines (97/115) 81.0% Functions (17/21) 64.1% Branches (25/39)

Line	Branch	Hits	Source Code
1			//
2			// Copyright (c) 2026 Steve Gerbino
3			//
4			// Distributed under the Boost Software License, Version 1.0. (See accompanying
5			// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
6			//
7			// Official repository: https://github.com/cppalliance/corosio
8			//
9
10			#ifndef BOOST_COROSIO_DETAIL_EPOLL_OP_HPP
11			#define BOOST_COROSIO_DETAIL_EPOLL_OP_HPP
12
13			#include <boost/corosio/detail/platform.hpp>
14
15			#if BOOST_COROSIO_HAS_EPOLL
16
17			#include <boost/corosio/detail/config.hpp>
18			#include <boost/corosio/io_object.hpp>
19			#include <boost/corosio/endpoint.hpp>
20			#include <boost/capy/ex/executor_ref.hpp>
21			#include <boost/capy/coro.hpp>
22			#include <boost/capy/error.hpp>
23			#include <system_error>
24
25			#include "src/detail/make_err.hpp"
26			#include "src/detail/resume_coro.hpp"
27			#include "src/detail/scheduler_op.hpp"
28			#include "src/detail/endpoint_convert.hpp"
29
30			#include <unistd.h>
31			#include <errno.h>
32
33			#include <atomic>
34			#include <cstddef>
35			#include <memory>
36			#include <mutex>
37			#include <optional>
38			#include <stop_token>
39
40			#include <netinet/in.h>
41			#include <sys/socket.h>
42			#include <sys/uio.h>
43
44			/*
45			epoll Operation State
46			=====================
47
48			Each async I/O operation has a corresponding epoll_op-derived struct that
49			holds the operation's state while it's in flight. The socket impl owns
50			fixed slots for each operation type (conn_, rd_, wr_), so only one
51			operation of each type can be pending per socket at a time.
52
53			Persistent Registration
54			-----------------------
55			File descriptors are registered with epoll once (via descriptor_state) and
56			stay registered until closed. The descriptor_state tracks which operations
57			are pending (read_op, write_op, connect_op). When an event arrives, the
58			reactor dispatches to the appropriate pending operation.
59
60			Impl Lifetime Management
61			------------------------
62			When cancel() posts an op to the scheduler's ready queue, the socket impl
63			might be destroyed before the scheduler processes the op. The `impl_ptr`
64			member holds a shared_ptr to the impl, keeping it alive until the op
65			completes. This is set by cancel() and cleared in operator() after the
66			coroutine is resumed.
67
68			EOF Detection
69			-------------
70			For reads, 0 bytes with no error means EOF. But an empty user buffer also
71			returns 0 bytes. The `empty_buffer_read` flag distinguishes these cases.
72
73			SIGPIPE Prevention
74			------------------
75			Writes use sendmsg() with MSG_NOSIGNAL instead of writev() to prevent
76			SIGPIPE when the peer has closed.
77			*/
78
79			namespace boost::corosio::detail {
80
81			// Forward declarations
82			class epoll_socket_impl;
83			class epoll_acceptor_impl;
84			struct epoll_op;
85
86			// Forward declaration
87			class epoll_scheduler;
88
89			/** Per-descriptor state for persistent epoll registration.
90
91			Tracks pending operations for a file descriptor. The fd is registered
92			once with epoll and stays registered until closed.
93
94			This struct extends scheduler_op to support deferred I/O processing.
95			When epoll events arrive, the reactor sets ready_events and queues
96			this descriptor for processing. When popped from the scheduler queue,
97			operator() performs the actual I/O and queues completion handlers.
98
99			@par Deferred I/O Model
100			The reactor no longer performs I/O directly. Instead:
101			1. Reactor sets ready_events and queues descriptor_state
102			2. Scheduler pops descriptor_state and calls operator()
103			3. operator() performs I/O under mutex and queues completions
104
105			This eliminates per-descriptor mutex locking from the reactor hot path.
106
107			@par Thread Safety
108			The mutex protects operation pointers and ready flags during I/O.
109			ready_events_ and is_enqueued_ are atomic for lock-free reactor access.
110			*/
111			struct descriptor_state : scheduler_op
112			{
113			std::mutex mutex;
114
115			// Protected by mutex
116			epoll_op* read_op = nullptr;
117			epoll_op* write_op = nullptr;
118			epoll_op* connect_op = nullptr;
119
120			// Caches edge events that arrived before an op was registered
121			bool read_ready = false;
122			bool write_ready = false;
123
124			// Set during registration only (no mutex needed)
125			std::uint32_t registered_events = 0;
126			int fd = -1;
127
128			// For deferred I/O - set by reactor, read by scheduler
129			std::atomic<std::uint32_t> ready_events_{0};
130			std::atomic<bool> is_enqueued_{false};
131			epoll_scheduler const* scheduler_ = nullptr;
132
133			// Prevents impl destruction while this descriptor_state is queued.
134			// Set by close_socket() when is_enqueued_ is true, cleared by operator().
135			std::shared_ptr<void> impl_ref_;
136
137			/// Add ready events atomically.
138		79209	void add_ready_events(std::uint32_t ev) noexcept
139			{
140		79209	ready_events_.fetch_or(ev, std::memory_order_relaxed);
141		79209	}
142
143			/// Perform deferred I/O and queue completions.
144			void operator()() override;
145
146			/// Destroy without invoking.
147		✗	void destroy() override {}
148			};
149
150			struct epoll_op : scheduler_op
151			{
152			struct canceller
153			{
154			epoll_op* op;
155			void operator()() const noexcept;
156			};
157
158			capy::coro h;
159			capy::executor_ref ex;
160			std::error_code* ec_out = nullptr;
161			std::size_t* bytes_out = nullptr;
162
163			int fd = -1;
164			int errn = 0;
165			std::size_t bytes_transferred = 0;
166
167			std::atomic<bool> cancelled{false};
168			std::optional<std::stop_callback<canceller>> stop_cb;
169
170			// Prevents use-after-free when socket is closed with pending ops.
171			// See "Impl Lifetime Management" in file header.
172			std::shared_ptr<void> impl_ptr;
173
174			// For stop_token cancellation - pointer to owning socket/acceptor impl.
175			// When stop is requested, we call back to the impl to perform actual I/O cancellation.
176			epoll_socket_impl* socket_impl_ = nullptr;
177			epoll_acceptor_impl* acceptor_impl_ = nullptr;
178
179		15793	epoll_op() = default;
180
181		153161	void reset() noexcept
182			{
183		153161	fd = -1;
184		153161	errn = 0;
185		153161	bytes_transferred = 0;
186		153161	cancelled.store(false, std::memory_order_relaxed);
187		153161	impl_ptr.reset();
188		153161	socket_impl_ = nullptr;
189		153161	acceptor_impl_ = nullptr;
190		153161	}
191
192		147920	void operator()() override
193			{
194		147920	stop_cb.reset();
195
196	1/2 ✓ Branch 0 taken 147920 times. ✗ Branch 1 not taken.	147920	if (ec_out)
197			{
198	2/2 ✓ Branch 1 taken 201 times. ✓ Branch 2 taken 147719 times.	147920	if (cancelled.load(std::memory_order_acquire))
199		201	*ec_out = capy::error::canceled;
200	2/2 ✓ Branch 0 taken 1 time. ✓ Branch 1 taken 147718 times.	147719	else if (errn != 0)
201		1	*ec_out = make_err(errn);
202	6/6 ✓ Branch 1 taken 73822 times. ✓ Branch 2 taken 73896 times. ✓ Branch 3 taken 5 times. ✓ Branch 4 taken 73817 times. ✓ Branch 5 taken 5 times. ✓ Branch 6 taken 147713 times.	147718	else if (is_read_operation() && bytes_transferred == 0)
203		5	*ec_out = capy::error::eof;
204			else
205		147713	*ec_out = {};
206			}
207
208	1/2 ✓ Branch 0 taken 147920 times. ✗ Branch 1 not taken.	147920	if (bytes_out)
209		147920	*bytes_out = bytes_transferred;
210
211			// Move to stack before resuming coroutine. The coroutine might close
212			// the socket, releasing the last wrapper ref. If impl_ptr were the
213			// last ref and we destroyed it while still in operator(), we'd have
214			// use-after-free. Moving to local ensures destruction happens at
215			// function exit, after all member accesses are complete.
216		147920	capy::executor_ref saved_ex( std::move( ex ) );
217		147920	capy::coro saved_h( std::move( h ) );
218		147920	auto prevent_premature_destruction = std::move(impl_ptr);
219	1/1 ✓ Branch 1 taken 147920 times.	147920	resume_coro(saved_ex, saved_h);
220		147920	}
221
222		73895	virtual bool is_read_operation() const noexcept { return false; }
223			virtual void cancel() noexcept = 0;
224
225		✗	void destroy() override
226			{
227		✗	stop_cb.reset();
228		✗	impl_ptr.reset();
229		✗	}
230
231		24128	void request_cancel() noexcept
232			{
233		24128	cancelled.store(true, std::memory_order_release);
234		24128	}
235
236		150537	void start(std::stop_token token, epoll_socket_impl* impl)
237			{
238		150537	cancelled.store(false, std::memory_order_release);
239		150537	stop_cb.reset();
240		150537	socket_impl_ = impl;
241		150537	acceptor_impl_ = nullptr;
242
243	2/2 ✓ Branch 1 taken 105 times. ✓ Branch 2 taken 150432 times.	150537	if (token.stop_possible())
244		105	stop_cb.emplace(token, canceller{this});
245		150537	}
246
247		2624	void start(std::stop_token token, epoll_acceptor_impl* impl)
248			{
249		2624	cancelled.store(false, std::memory_order_release);
250		2624	stop_cb.reset();
251		2624	socket_impl_ = nullptr;
252		2624	acceptor_impl_ = impl;
253
254	2/2 ✓ Branch 1 taken 9 times. ✓ Branch 2 taken 2615 times.	2624	if (token.stop_possible())
255		9	stop_cb.emplace(token, canceller{this});
256		2624	}
257
258		153112	void complete(int err, std::size_t bytes) noexcept
259			{
260		153112	errn = err;
261		153112	bytes_transferred = bytes;
262		153112	}
263
264		✗	virtual void perform_io() noexcept {}
265			};
266
267
268			struct epoll_connect_op : epoll_op
269			{
270			endpoint target_endpoint;
271
272		2617	void reset() noexcept
273			{
274		2617	epoll_op::reset();
275		2617	target_endpoint = endpoint{};
276		2617	}
277
278		2616	void perform_io() noexcept override
279			{
280			// connect() completion status is retrieved via SO_ERROR, not return value
281		2616	int err = 0;
282		2616	socklen_t len = sizeof(err);
283	1/2 ✗ Branch 1 not taken. ✓ Branch 2 taken 2616 times.	2616	if (::getsockopt(fd, SOL_SOCKET, SO_ERROR, &err, &len) < 0)
284		✗	err = errno;
285		2616	complete(err, 0);
286		2616	}
287
288			// Defined in sockets.cpp where epoll_socket_impl is complete
289			void operator()() override;
290			void cancel() noexcept override;
291			};
292
293
294			struct epoll_read_op : epoll_op
295			{
296			static constexpr std::size_t max_buffers = 16;
297			iovec iovecs[max_buffers];
298			int iovec_count = 0;
299			bool empty_buffer_read = false;
300
301		73823	bool is_read_operation() const noexcept override
302			{
303		73823	return !empty_buffer_read;
304			}
305
306		74020	void reset() noexcept
307			{
308		74020	epoll_op::reset();
309		74020	iovec_count = 0;
310		74020	empty_buffer_read = false;
311		74020	}
312
313		127	void perform_io() noexcept override
314			{
315			ssize_t n;
316			do {
317		127	n = ::readv(fd, iovecs, iovec_count);
318	3/4 ✓ Branch 0 taken 76 times. ✓ Branch 1 taken 51 times. ✗ Branch 2 not taken. ✓ Branch 3 taken 76 times.	127	} while (n < 0 && errno == EINTR);
319
320	2/2 ✓ Branch 0 taken 51 times. ✓ Branch 1 taken 76 times.	127	if (n >= 0)
321		51	complete(0, static_cast<std::size_t>(n));
322			else
323		76	complete(errno, 0);
324		127	}
325
326			void cancel() noexcept override;
327			};
328
329
330			struct epoll_write_op : epoll_op
331			{
332			static constexpr std::size_t max_buffers = 16;
333			iovec iovecs[max_buffers];
334			int iovec_count = 0;
335
336		73900	void reset() noexcept
337			{
338		73900	epoll_op::reset();
339		73900	iovec_count = 0;
340		73900	}
341
342		✗	void perform_io() noexcept override
343			{
344		✗	msghdr msg{};
345		✗	msg.msg_iov = iovecs;
346		✗	msg.msg_iovlen = static_cast<std::size_t>(iovec_count);
347
348			ssize_t n;
349			do {
350		✗	n = ::sendmsg(fd, &msg, MSG_NOSIGNAL);
351		✗	} while (n < 0 && errno == EINTR);
352
353		✗	if (n >= 0)
354		✗	complete(0, static_cast<std::size_t>(n));
355			else
356		✗	complete(errno, 0);
357		✗	}
358
359			void cancel() noexcept override;
360			};
361
362
363			struct epoll_accept_op : epoll_op
364			{
365			int accepted_fd = -1;
366			io_object::io_object_impl* peer_impl = nullptr;
367			io_object::io_object_impl** impl_out = nullptr;
368
369		2624	void reset() noexcept
370			{
371		2624	epoll_op::reset();
372		2624	accepted_fd = -1;
373		2624	peer_impl = nullptr;
374		2624	impl_out = nullptr;
375		2624	}
376
377		2613	void perform_io() noexcept override
378			{
379		2613	sockaddr_in addr{};
380		2613	socklen_t addrlen = sizeof(addr);
381			int new_fd;
382			do {
383		2613	new_fd = ::accept4(fd, reinterpret_cast<sockaddr*>(&addr),
384			&addrlen, SOCK_NONBLOCK \| SOCK_CLOEXEC);
385	1/4 ✗ Branch 0 not taken. ✓ Branch 1 taken 2613 times. ✗ Branch 2 not taken. ✗ Branch 3 not taken.	2613	} while (new_fd < 0 && errno == EINTR);
386
387	1/2 ✓ Branch 0 taken 2613 times. ✗ Branch 1 not taken.	2613	if (new_fd >= 0)
388			{
389		2613	accepted_fd = new_fd;
390		2613	complete(0, 0);
391			}
392			else
393			{
394		✗	complete(errno, 0);
395			}
396		2613	}
397
398			// Defined in acceptors.cpp where epoll_acceptor_impl is complete
399			void operator()() override;
400			void cancel() noexcept override;
401			};
402
403			} // namespace boost::corosio::detail
404
405			#endif // BOOST_COROSIO_HAS_EPOLL
406
407			#endif // BOOST_COROSIO_DETAIL_EPOLL_OP_HPP
408