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		123989	void add_ready_events(std::uint32_t ev) noexcept
139			{
140		123989	ready_events_.fetch_or(ev, std::memory_order_relaxed);
141		123989	}
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		26374	epoll_op() = default;
180
181		239196	void reset() noexcept
182			{
183		239196	fd = -1;
184		239196	errn = 0;
185		239196	bytes_transferred = 0;
186		239196	cancelled.store(false, std::memory_order_relaxed);
187		239196	impl_ptr.reset();
188		239196	socket_impl_ = nullptr;
189		239196	acceptor_impl_ = nullptr;
190		239196	}
191
192		230428	void operator()() override
193			{
194		230428	stop_cb.reset();
195
196	1/2 ✓ Branch 0 taken 230428 times. ✗ Branch 1 not taken.	230428	if (ec_out)
197			{
198	2/2 ✓ Branch 1 taken 205 times. ✓ Branch 2 taken 230223 times.	230428	if (cancelled.load(std::memory_order_acquire))
199		205	*ec_out = capy::error::canceled;
200	2/2 ✓ Branch 0 taken 1 time. ✓ Branch 1 taken 230222 times.	230223	else if (errn != 0)
201		1	*ec_out = make_err(errn);
202	6/6 ✓ Branch 1 taken 115073 times. ✓ Branch 2 taken 115149 times. ✓ Branch 3 taken 5 times. ✓ Branch 4 taken 115068 times. ✓ Branch 5 taken 5 times. ✓ Branch 6 taken 230217 times.	230222	else if (is_read_operation() && bytes_transferred == 0)
203		5	*ec_out = capy::error::eof;
204			else
205		230217	*ec_out = {};
206			}
207
208	1/2 ✓ Branch 0 taken 230428 times. ✗ Branch 1 not taken.	230428	if (bytes_out)
209		230428	*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		230428	capy::executor_ref saved_ex( std::move( ex ) );
217		230428	capy::coro saved_h( std::move( h ) );
218		230428	auto prevent_premature_destruction = std::move(impl_ptr);
219	1/1 ✓ Branch 1 taken 230428 times.	230428	resume_coro(saved_ex, saved_h);
220		230428	}
221
222		115148	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		40006	void request_cancel() noexcept
232			{
233		40006	cancelled.store(true, std::memory_order_release);
234		40006	}
235
236		234808	void start(std::stop_token token, epoll_socket_impl* impl)
237			{
238		234808	cancelled.store(false, std::memory_order_release);
239		234808	stop_cb.reset();
240		234808	socket_impl_ = impl;
241		234808	acceptor_impl_ = nullptr;
242
243	2/2 ✓ Branch 1 taken 107 times. ✓ Branch 2 taken 234701 times.	234808	if (token.stop_possible())
244		107	stop_cb.emplace(token, canceller{this});
245		234808	}
246
247		4388	void start(std::stop_token token, epoll_acceptor_impl* impl)
248			{
249		4388	cancelled.store(false, std::memory_order_release);
250		4388	stop_cb.reset();
251		4388	socket_impl_ = nullptr;
252		4388	acceptor_impl_ = impl;
253
254	2/2 ✓ Branch 1 taken 9 times. ✓ Branch 2 taken 4379 times.	4388	if (token.stop_possible())
255		9	stop_cb.emplace(token, canceller{this});
256		4388	}
257
258		239146	void complete(int err, std::size_t bytes) noexcept
259			{
260		239146	errn = err;
261		239146	bytes_transferred = bytes;
262		239146	}
263
264		✗	virtual void perform_io() noexcept {}
265			};
266
267
268			struct epoll_connect_op : epoll_op
269			{
270			endpoint target_endpoint;
271
272		4380	void reset() noexcept
273			{
274		4380	epoll_op::reset();
275		4380	target_endpoint = endpoint{};
276		4380	}
277
278		4380	void perform_io() noexcept override
279			{
280			// connect() completion status is retrieved via SO_ERROR, not return value
281		4380	int err = 0;
282		4380	socklen_t len = sizeof(err);
283	1/2 ✗ Branch 1 not taken. ✓ Branch 2 taken 4380 times.	4380	if (::getsockopt(fd, SOL_SOCKET, SO_ERROR, &err, &len) < 0)
284		✗	err = errno;
285		4380	complete(err, 0);
286		4380	}
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		115074	bool is_read_operation() const noexcept override
302			{
303		115074	return !empty_buffer_read;
304			}
305
306		115275	void reset() noexcept
307			{
308		115275	epoll_op::reset();
309		115275	iovec_count = 0;
310		115275	empty_buffer_read = false;
311		115275	}
312
313		130	void perform_io() noexcept override
314			{
315			ssize_t n;
316			do {
317		130	n = ::readv(fd, iovecs, iovec_count);
318	3/4 ✓ Branch 0 taken 78 times. ✓ Branch 1 taken 52 times. ✗ Branch 2 not taken. ✓ Branch 3 taken 78 times.	130	} while (n < 0 && errno == EINTR);
319
320	2/2 ✓ Branch 0 taken 52 times. ✓ Branch 1 taken 78 times.	130	if (n >= 0)
321		52	complete(0, static_cast<std::size_t>(n));
322			else
323		78	complete(errno, 0);
324		130	}
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		115153	void reset() noexcept
337			{
338		115153	epoll_op::reset();
339		115153	iovec_count = 0;
340		115153	}
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		4388	void reset() noexcept
370			{
371		4388	epoll_op::reset();
372		4388	accepted_fd = -1;
373		4388	peer_impl = nullptr;
374		4388	impl_out = nullptr;
375		4388	}
376
377		4377	void perform_io() noexcept override
378			{
379		4377	sockaddr_in addr{};
380		4377	socklen_t addrlen = sizeof(addr);
381			int new_fd;
382			do {
383		4377	new_fd = ::accept4(fd, reinterpret_cast<sockaddr*>(&addr),
384			&addrlen, SOCK_NONBLOCK \| SOCK_CLOEXEC);
385	1/4 ✗ Branch 0 not taken. ✓ Branch 1 taken 4377 times. ✗ Branch 2 not taken. ✗ Branch 3 not taken.	4377	} while (new_fd < 0 && errno == EINTR);
386
387	1/2 ✓ Branch 0 taken 4377 times. ✗ Branch 1 not taken.	4377	if (new_fd >= 0)
388			{
389		4377	accepted_fd = new_fd;
390		4377	complete(0, 0);
391			}
392			else
393			{
394		✗	complete(errno, 0);
395			}
396		4377	}
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