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

86.7% Lines (52/60) 93.3% List of functions (14/15)

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

epoll_tcp_service.hpp

f(x) Functions (15)

Function Calls Lines Blocks

Line	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_NATIVE_DETAIL_EPOLL_EPOLL_TCP_SERVICE_HPP
11		#define BOOST_COROSIO_NATIVE_DETAIL_EPOLL_EPOLL_TCP_SERVICE_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/detail/tcp_service.hpp>
19
20		#include <boost/corosio/native/detail/epoll/epoll_tcp_socket.hpp>
21		#include <boost/corosio/native/detail/epoll/epoll_scheduler.hpp>
22		#include <boost/corosio/native/detail/reactor/reactor_socket_service.hpp>
23
24		#include <boost/corosio/native/detail/reactor/reactor_op_complete.hpp>
25
26		#include <coroutine>
27
28		#include <errno.h>
29		#include <netinet/in.h>
30		#include <netinet/tcp.h>
31		#include <sys/epoll.h>
32		#include <sys/socket.h>
33		#include <unistd.h>
34
35		/*
36		epoll Socket Implementation
37		===========================
38
39		Each I/O operation follows the same pattern:
40		1. Try the syscall immediately (non-blocking socket)
41		2. If it succeeds or fails with a real error, post to completion queue
42		3. If EAGAIN/EWOULDBLOCK, register with epoll and wait
43
44		This "try first" approach avoids unnecessary epoll round-trips for
45		operations that can complete immediately (common for small reads/writes
46		on fast local connections).
47
48		One-Shot Registration
49		---------------------
50		We use one-shot epoll registration: each operation registers, waits for
51		one event, then unregisters. This simplifies the state machine since we
52		don't need to track whether an fd is currently registered or handle
53		re-arming. The tradeoff is slightly more epoll_ctl calls, but the
54		simplicity is worth it.
55
56		Cancellation
57		------------
58		See op.hpp for the completion/cancellation race handling via the
59		`registered` atomic. cancel() must complete pending operations (post
60		them with cancelled flag) so coroutines waiting on them can resume.
61		close_socket() calls cancel() first to ensure this.
62
63		Impl Lifetime with shared_ptr
64		-----------------------------
65		Socket impls use enable_shared_from_this. The service owns impls via
66		shared_ptr maps (impl_ptrs_) keyed by raw pointer for O(1) lookup and
67		removal. When a user calls close(), we call cancel() which posts pending
68		ops to the scheduler.
69
70		CRITICAL: The posted ops must keep the impl alive until they complete.
71		Otherwise the scheduler would process a freed op (use-after-free). The
72		cancel() method captures shared_from_this() into op.impl_ptr before
73		posting. When the op completes, impl_ptr is cleared, allowing the impl
74		to be destroyed if no other references exist.
75
76		Service Ownership
77		-----------------
78		epoll_tcp_service owns all socket impls. destroy_impl() removes the
79		shared_ptr from the map, but the impl may survive if ops still hold
80		impl_ptr refs. shutdown() closes all sockets and clears the map; any
81		in-flight ops will complete and release their refs.
82		*/
83
84		namespace boost::corosio::detail {
85
86		/** epoll TCP service implementation.
87
88		Inherits from tcp_service to enable runtime polymorphism.
89		Uses key_type = tcp_service for service lookup.
90		*/
91		class BOOST_COROSIO_DECL epoll_tcp_service final
92		: public reactor_socket_service<
93		epoll_tcp_service,
94		tcp_service,
95		epoll_scheduler,
96		epoll_tcp_socket>
97		{
98		public:
99	401x	explicit epoll_tcp_service(capy::execution_context& ctx)
100	401x	: reactor_socket_service(ctx)
101		{
102	401x	}
103
104		std::error_code open_socket(
105		tcp_socket::implementation& impl,
106		int family,
107		int type,
108		int protocol) override;
109
110		std::error_code
111		bind_socket(tcp_socket::implementation& impl, endpoint ep) override;
112		};
113
114		inline void
115	✗	epoll_connect_op::cancel() noexcept
116		{
117	✗	if (socket_impl_)
118	✗	socket_impl_->cancel_single_op(*this);
119		else
120	✗	request_cancel();
121	✗	}
122
123		inline void
124	792x	epoll_read_op::cancel() noexcept
125		{
126	792x	if (socket_impl_)
127	792x	socket_impl_->cancel_single_op(*this);
128		else
129	✗	request_cancel();
130	792x	}
131
132		inline void
133	1x	epoll_write_op::cancel() noexcept
134		{
135	1x	if (socket_impl_)
136	1x	socket_impl_->cancel_single_op(*this);
137		else
138	✗	request_cancel();
139	1x	}
140
141		inline void
142	63822x	epoll_op::operator()()
143		{
144	63822x	complete_io_op(*this);
145	63822x	}
146
147		inline void
148	4904x	epoll_connect_op::operator()()
149		{
150	4904x	complete_connect_op(*this);
151	4904x	}
152
153	14816x	inline epoll_tcp_socket::epoll_tcp_socket(epoll_tcp_service& svc) noexcept
154	14816x	: reactor_stream_socket(svc)
155		{
156	14816x	}
157
158	14816x	inline epoll_tcp_socket::~epoll_tcp_socket() = default;
159
160		inline std::coroutine_handle<>
161	4904x	epoll_tcp_socket::connect(
162		std::coroutine_handle<> h,
163		capy::executor_ref ex,
164		endpoint ep,
165		std::stop_token token,
166		std::error_code* ec)
167		{
168	4904x	return do_connect(h, ex, ep, token, ec);
169		}
170
171		inline std::coroutine_handle<>
172	115740x	epoll_tcp_socket::read_some(
173		std::coroutine_handle<> h,
174		capy::executor_ref ex,
175		buffer_param param,
176		std::stop_token token,
177		std::error_code* ec,
178		std::size_t* bytes_out)
179		{
180	115740x	return do_read_some(h, ex, param, token, ec, bytes_out);
181		}
182
183		inline std::coroutine_handle<>
184	186117x	epoll_tcp_socket::write_some(
185		std::coroutine_handle<> h,
186		capy::executor_ref ex,
187		buffer_param param,
188		std::stop_token token,
189		std::error_code* ec,
190		std::size_t* bytes_out)
191		{
192	186117x	return do_write_some(h, ex, param, token, ec, bytes_out);
193		}
194
195		inline void
196	123x	epoll_tcp_socket::cancel() noexcept
197		{
198	123x	do_cancel();
199	123x	}
200
201		inline void
202	44382x	epoll_tcp_socket::close_socket() noexcept
203		{
204	44382x	do_close_socket();
205	44382x	}
206
207		inline std::error_code
208	4924x	epoll_tcp_service::open_socket(
209		tcp_socket::implementation& impl, int family, int type, int protocol)
210		{
211	4924x	auto* epoll_impl = static_cast<epoll_tcp_socket*>(&impl);
212	4924x	epoll_impl->close_socket();
213
214	4924x	int fd = ::socket(family, type \| SOCK_NONBLOCK \| SOCK_CLOEXEC, protocol);
215	4924x	if (fd < 0)
216	✗	return make_err(errno);
217
218	4924x	if (family == AF_INET6)
219		{
220	6x	int one = 1;
221	6x	::setsockopt(fd, IPPROTO_IPV6, IPV6_V6ONLY, &one, sizeof(one));
222		}
223
224	4924x	epoll_impl->fd_ = fd;
225
226		// Register fd with epoll (edge-triggered mode)
227	4924x	epoll_impl->desc_state_.fd = fd;
228		{
229	4924x	std::lock_guard lock(epoll_impl->desc_state_.mutex);
230	4924x	epoll_impl->desc_state_.read_op = nullptr;
231	4924x	epoll_impl->desc_state_.write_op = nullptr;
232	4924x	epoll_impl->desc_state_.connect_op = nullptr;
233	4924x	}
234	4924x	scheduler().register_descriptor(fd, &epoll_impl->desc_state_);
235
236	4924x	return {};
237		}
238
239		inline std::error_code
240	6x	epoll_tcp_service::bind_socket(
241		tcp_socket::implementation& impl, endpoint ep)
242		{
243	6x	return static_cast<epoll_tcp_socket*>(&impl)->do_bind(ep);
244		}
245
246		} // namespace boost::corosio::detail
247
248		#endif // BOOST_COROSIO_HAS_EPOLL
249
250		#endif // BOOST_COROSIO_NATIVE_DETAIL_EPOLL_EPOLL_TCP_SERVICE_HPP
251