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