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