TLA Line data 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 HIT 333 : explicit epoll_tcp_service(capy::execution_context& ctx)
100 333 : : reactor_socket_service(ctx)
101 : {
102 333 : }
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 MIS 0 : epoll_connect_op::cancel() noexcept
116 : {
117 0 : if (socket_impl_)
118 0 : socket_impl_->cancel_single_op(*this);
119 : else
120 0 : request_cancel();
121 0 : }
122 :
123 : inline void
124 HIT 98 : epoll_read_op::cancel() noexcept
125 : {
126 98 : if (socket_impl_)
127 98 : socket_impl_->cancel_single_op(*this);
128 : else
129 MIS 0 : request_cancel();
130 HIT 98 : }
131 :
132 : inline void
133 MIS 0 : epoll_write_op::cancel() noexcept
134 : {
135 0 : if (socket_impl_)
136 0 : socket_impl_->cancel_single_op(*this);
137 : else
138 0 : request_cancel();
139 0 : }
140 :
141 : inline void
142 HIT 44510 : epoll_op::operator()()
143 : {
144 44510 : complete_io_op(*this);
145 44510 : }
146 :
147 : inline void
148 2402 : epoll_connect_op::operator()()
149 : {
150 2402 : complete_connect_op(*this);
151 2402 : }
152 :
153 7267 : inline epoll_tcp_socket::epoll_tcp_socket(epoll_tcp_service& svc) noexcept
154 7267 : : reactor_stream_socket(svc)
155 : {
156 7267 : }
157 :
158 7267 : inline epoll_tcp_socket::~epoll_tcp_socket() = default;
159 :
160 : inline std::coroutine_handle<>
161 2402 : 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 2402 : return do_connect(h, ex, ep, token, ec);
169 : }
170 :
171 : inline std::coroutine_handle<>
172 111016 : 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 111016 : return do_read_some(h, ex, param, token, ec, bytes_out);
181 : }
182 :
183 : inline std::coroutine_handle<>
184 110864 : 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 110864 : return do_write_some(h, ex, param, token, ec, bytes_out);
193 : }
194 :
195 : inline void
196 95 : epoll_tcp_socket::cancel() noexcept
197 : {
198 95 : do_cancel();
199 95 : }
200 :
201 : inline void
202 21777 : epoll_tcp_socket::close_socket() noexcept
203 : {
204 21777 : do_close_socket();
205 21777 : }
206 :
207 : inline std::error_code
208 2422 : epoll_tcp_service::open_socket(
209 : tcp_socket::implementation& impl, int family, int type, int protocol)
210 : {
211 2422 : auto* epoll_impl = static_cast<epoll_tcp_socket*>(&impl);
212 2422 : epoll_impl->close_socket();
213 :
214 2422 : int fd = ::socket(family, type | SOCK_NONBLOCK | SOCK_CLOEXEC, protocol);
215 2422 : if (fd < 0)
216 MIS 0 : return make_err(errno);
217 :
218 HIT 2422 : if (family == AF_INET6)
219 : {
220 6 : int one = 1;
221 6 : ::setsockopt(fd, IPPROTO_IPV6, IPV6_V6ONLY, &one, sizeof(one));
222 : }
223 :
224 2422 : epoll_impl->fd_ = fd;
225 :
226 : // Register fd with epoll (edge-triggered mode)
227 2422 : epoll_impl->desc_state_.fd = fd;
228 : {
229 2422 : std::lock_guard lock(epoll_impl->desc_state_.mutex);
230 2422 : epoll_impl->desc_state_.read_op = nullptr;
231 2422 : epoll_impl->desc_state_.write_op = nullptr;
232 2422 : epoll_impl->desc_state_.connect_op = nullptr;
233 2422 : }
234 2422 : scheduler().register_descriptor(fd, &epoll_impl->desc_state_);
235 :
236 2422 : return {};
237 : }
238 :
239 : inline std::error_code
240 6 : epoll_tcp_service::bind_socket(
241 : tcp_socket::implementation& impl, endpoint ep)
242 : {
243 6 : 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
|
