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