TLA Line data 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_NATIVE_LOCAL_STREAM_SOCKET_HPP
12 : #define BOOST_COROSIO_NATIVE_NATIVE_LOCAL_STREAM_SOCKET_HPP
13 :
14 : #include <boost/corosio/local_stream_socket.hpp>
15 : #include <boost/corosio/backend.hpp>
16 :
17 : #ifndef BOOST_COROSIO_MRDOCS
18 : #if BOOST_COROSIO_HAS_EPOLL
19 : #include <boost/corosio/native/detail/epoll/epoll_types.hpp>
20 : #endif
21 :
22 : #if BOOST_COROSIO_HAS_SELECT
23 : #include <boost/corosio/native/detail/select/select_types.hpp>
24 : #endif
25 :
26 : #if BOOST_COROSIO_HAS_KQUEUE
27 : #include <boost/corosio/native/detail/kqueue/kqueue_types.hpp>
28 : #endif
29 :
30 : #if BOOST_COROSIO_HAS_IO_URING
31 : #include <boost/corosio/native/detail/io_uring/io_uring_types.hpp>
32 : #endif
33 :
34 : #if BOOST_COROSIO_HAS_IOCP
35 : #include <boost/corosio/native/detail/iocp/win_local_stream_service.hpp>
36 : #endif
37 : #endif // !BOOST_COROSIO_MRDOCS
38 :
39 : namespace boost::corosio {
40 :
41 : /** An asynchronous Unix stream socket with devirtualized I/O operations.
42 :
43 : This class template inherits from @ref local_stream_socket and
44 : shadows the async operations (`read_some`, `write_some`,
45 : `connect`) with versions that call the backend implementation
46 : directly, allowing the compiler to inline through the entire
47 : call chain.
48 :
49 : Non-async operations (`open`, `close`, `cancel`, socket options)
50 : remain unchanged and dispatch through the compiled library.
51 :
52 : A `native_local_stream_socket` IS-A `local_stream_socket` and
53 : can be passed to any function expecting `local_stream_socket&`
54 : or `io_stream&`, in which case virtual dispatch is used
55 : transparently.
56 :
57 : @tparam Backend A backend tag value (e.g., `epoll`) whose type
58 : provides the concrete implementation types.
59 :
60 : @par Thread Safety
61 : Same as @ref local_stream_socket.
62 :
63 : @par Example
64 : @par !example connect
65 :
66 : @see local_stream_socket, epoll_t, iocp_t
67 : */
68 : template<auto Backend>
69 : class native_local_stream_socket : public local_stream_socket
70 : {
71 : using backend_type = decltype(Backend);
72 : using impl_type = typename backend_type::local_stream_socket_type;
73 : using service_type = typename backend_type::local_stream_service_type;
74 :
75 HIT 34 : impl_type& get_impl() noexcept
76 : {
77 34 : return *static_cast<impl_type*>(h_.get());
78 : }
79 :
80 : template<class MutableBufferSequence>
81 : struct native_read_awaitable
82 : {
83 : native_local_stream_socket& self_;
84 : MutableBufferSequence buffers_;
85 : std::stop_token token_;
86 : mutable std::error_code ec_;
87 : mutable std::size_t bytes_transferred_ = 0;
88 :
89 8 : native_read_awaitable(
90 : native_local_stream_socket& self,
91 : MutableBufferSequence buffers) noexcept
92 8 : : self_(self)
93 8 : , buffers_(std::move(buffers))
94 : {
95 8 : }
96 :
97 8 : bool await_ready() const noexcept
98 : {
99 : // A pre-set ec_ means the initiator failed before
100 : // dispatch (e.g. a closed object).
101 8 : return static_cast<bool>(ec_) || token_.stop_requested();
102 : }
103 :
104 8 : [[nodiscard]] capy::io_result<std::size_t> await_resume() const noexcept
105 : {
106 8 : if (token_.stop_requested())
107 2 : return {make_error_code(std::errc::operation_canceled), 0};
108 6 : return {ec_, bytes_transferred_};
109 : }
110 :
111 8 : auto await_suspend(std::coroutine_handle<> h, capy::io_env const* env)
112 : -> std::coroutine_handle<>
113 : {
114 8 : token_ = env->stop_token;
115 24 : return self_.get_impl().read_some(
116 24 : h, env->executor, buffers_, token_, &ec_, &bytes_transferred_);
117 : }
118 : };
119 :
120 : template<class ConstBufferSequence>
121 : struct native_write_awaitable
122 : {
123 : native_local_stream_socket& self_;
124 : ConstBufferSequence buffers_;
125 : std::stop_token token_;
126 : mutable std::error_code ec_;
127 : mutable std::size_t bytes_transferred_ = 0;
128 :
129 8 : native_write_awaitable(
130 : native_local_stream_socket& self,
131 : ConstBufferSequence buffers) noexcept
132 8 : : self_(self)
133 8 : , buffers_(std::move(buffers))
134 : {
135 8 : }
136 :
137 8 : bool await_ready() const noexcept
138 : {
139 : // A pre-set ec_ means the initiator failed before
140 : // dispatch (e.g. a closed object).
141 8 : return static_cast<bool>(ec_) || token_.stop_requested();
142 : }
143 :
144 8 : [[nodiscard]] capy::io_result<std::size_t> await_resume() const noexcept
145 : {
146 8 : if (token_.stop_requested())
147 2 : return {make_error_code(std::errc::operation_canceled), 0};
148 6 : return {ec_, bytes_transferred_};
149 : }
150 :
151 8 : auto await_suspend(std::coroutine_handle<> h, capy::io_env const* env)
152 : -> std::coroutine_handle<>
153 : {
154 8 : token_ = env->stop_token;
155 24 : return self_.get_impl().write_some(
156 24 : h, env->executor, buffers_, token_, &ec_, &bytes_transferred_);
157 : }
158 : };
159 :
160 : struct native_wait_awaitable
161 : {
162 : native_local_stream_socket& self_;
163 : wait_type w_;
164 : std::stop_token token_;
165 : mutable std::error_code ec_;
166 :
167 6 : native_wait_awaitable(
168 : native_local_stream_socket& self, wait_type w) noexcept
169 6 : : self_(self)
170 6 : , w_(w)
171 : {
172 6 : }
173 :
174 6 : bool await_ready() const noexcept
175 : {
176 : // A pre-set ec_ means the initiator failed before
177 : // dispatch (e.g. auto-open).
178 6 : return static_cast<bool>(ec_) || token_.stop_requested();
179 : }
180 :
181 6 : [[nodiscard]] capy::io_result<> await_resume() const noexcept
182 : {
183 6 : if (token_.stop_requested())
184 2 : return {make_error_code(std::errc::operation_canceled)};
185 4 : return {ec_};
186 : }
187 :
188 6 : auto await_suspend(std::coroutine_handle<> h, capy::io_env const* env)
189 : -> std::coroutine_handle<>
190 : {
191 6 : token_ = env->stop_token;
192 18 : return self_.get_impl().wait(
193 18 : h, env->executor, w_, token_, &ec_);
194 : }
195 : };
196 :
197 : struct native_connect_awaitable
198 : {
199 : native_local_stream_socket& self_;
200 : corosio::local_endpoint endpoint_;
201 : std::stop_token token_;
202 : mutable std::error_code ec_;
203 :
204 12 : native_connect_awaitable(
205 : native_local_stream_socket& self,
206 : corosio::local_endpoint ep) noexcept
207 12 : : self_(self)
208 12 : , endpoint_(ep)
209 : {
210 12 : }
211 :
212 12 : bool await_ready() const noexcept
213 : {
214 : // A pre-set ec_ means the initiator failed before
215 : // dispatch (e.g. a closed object).
216 12 : return static_cast<bool>(ec_) || token_.stop_requested();
217 : }
218 :
219 12 : [[nodiscard]] capy::io_result<> await_resume() const noexcept
220 : {
221 12 : if (token_.stop_requested())
222 2 : return {make_error_code(std::errc::operation_canceled)};
223 10 : return {ec_};
224 : }
225 :
226 12 : auto await_suspend(std::coroutine_handle<> h, capy::io_env const* env)
227 : -> std::coroutine_handle<>
228 : {
229 12 : token_ = env->stop_token;
230 36 : return self_.get_impl().connect(
231 36 : h, env->executor, endpoint_, token_, &ec_);
232 : }
233 : };
234 :
235 : public:
236 : /** Construct a native socket from an execution context.
237 :
238 : @param ctx The execution context that will own this socket.
239 : */
240 40 : explicit native_local_stream_socket(capy::execution_context& ctx)
241 40 : : io_object(create_handle<service_type>(ctx))
242 : {
243 40 : }
244 :
245 : /** Construct a native socket from an executor.
246 :
247 : @param ex The executor whose context will own the socket.
248 : */
249 : template<class Ex>
250 : requires(!std::same_as<
251 : std::remove_cvref_t<Ex>,
252 : native_local_stream_socket>) &&
253 : capy::Executor<Ex>
254 : explicit native_local_stream_socket(Ex const& ex)
255 : : native_local_stream_socket(ex.context())
256 : {
257 : }
258 :
259 : /// Move construct.
260 6 : native_local_stream_socket(native_local_stream_socket&&) noexcept = default;
261 :
262 : /// Move assign.
263 : native_local_stream_socket&
264 : operator=(native_local_stream_socket&&) noexcept = default;
265 :
266 : native_local_stream_socket(native_local_stream_socket const&) = delete;
267 : native_local_stream_socket&
268 : operator=(native_local_stream_socket const&) = delete;
269 :
270 : /** Asynchronously read data from the socket.
271 :
272 : Calls the backend implementation directly, bypassing virtual
273 : dispatch. Otherwise identical to @ref io_stream::read_some.
274 :
275 : @param buffers The buffer sequence to read into.
276 :
277 : @return An awaitable yielding `(error_code, std::size_t)`.
278 : */
279 : template<capy::MutableBufferSequence MB>
280 8 : [[nodiscard]] auto read_some(MB const& buffers)
281 : {
282 8 : return native_read_awaitable<MB>(*this, buffers);
283 : }
284 :
285 : /** Asynchronously write data to the socket.
286 :
287 : Calls the backend implementation directly, bypassing virtual
288 : dispatch. Otherwise identical to @ref io_stream::write_some.
289 :
290 : @param buffers The buffer sequence to write from.
291 :
292 : @return An awaitable yielding `(error_code, std::size_t)`.
293 : */
294 : template<capy::ConstBufferSequence CB>
295 8 : [[nodiscard]] auto write_some(CB const& buffers)
296 : {
297 8 : return native_write_awaitable<CB>(*this, buffers);
298 : }
299 :
300 : /** Asynchronously connect to a remote endpoint.
301 :
302 : Calls the backend implementation directly, bypassing virtual
303 : dispatch. Otherwise identical to @ref local_stream_socket::connect.
304 :
305 : If the socket is not already open, it is opened automatically.
306 :
307 : @param ep The local endpoint (path) to connect to.
308 :
309 : @return An awaitable yielding `io_result<>`.
310 :
311 : If the socket needs to be opened and the open fails, the
312 : awaitable completes immediately with that error.
313 : */
314 12 : [[nodiscard]] auto connect(corosio::local_endpoint ep)
315 : {
316 12 : native_connect_awaitable aw(*this, ep);
317 12 : if (!is_open())
318 10 : aw.ec_ = open();
319 12 : return aw;
320 : }
321 :
322 : /** Asynchronously wait for the socket to be ready.
323 :
324 : Calls the backend implementation directly, bypassing virtual
325 : dispatch. Otherwise identical to @ref local_stream_socket::wait.
326 :
327 : @param w The wait direction (read, write, or error).
328 :
329 : @return An awaitable yielding `io_result<>`.
330 : */
331 6 : [[nodiscard]] auto wait(wait_type w)
332 : {
333 6 : return native_wait_awaitable(*this, w);
334 : }
335 : };
336 :
337 : } // namespace boost::corosio
338 :
339 : #endif // BOOST_COROSIO_NATIVE_NATIVE_LOCAL_STREAM_SOCKET_HPP
|
