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