. /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 - s.open();  
68   native_local_stream_socket<epoll> s(ctx); 67   native_local_stream_socket<epoll> s(ctx);
69   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)
  70 + co_return;
70   @endcode 71   @endcode
71   72  
72   @see local_stream_socket, epoll_t, iocp_t 73   @see local_stream_socket, epoll_t, iocp_t
73   */ 74   */
74   template<auto Backend> 75   template<auto Backend>
75   class native_local_stream_socket : public local_stream_socket 76   class native_local_stream_socket : public local_stream_socket
76   { 77   {
77   using backend_type = decltype(Backend); 78   using backend_type = decltype(Backend);
78   using impl_type = typename backend_type::local_stream_socket_type; 79   using impl_type = typename backend_type::local_stream_socket_type;
79   using service_type = typename backend_type::local_stream_service_type; 80   using service_type = typename backend_type::local_stream_service_type;
80   81  
HITCBC 81   20 impl_type& get_impl() noexcept 82   26 impl_type& get_impl() noexcept
82   { 83   {
HITCBC 83   20 return *static_cast<impl_type*>(h_.get()); 84   26 return *static_cast<impl_type*>(h_.get());
84   } 85   }
85   86  
86   template<class MutableBufferSequence> 87   template<class MutableBufferSequence>
87   struct native_read_awaitable 88   struct native_read_awaitable
88   { 89   {
89   native_local_stream_socket& self_; 90   native_local_stream_socket& self_;
90   MutableBufferSequence buffers_; 91   MutableBufferSequence buffers_;
91   std::stop_token token_; 92   std::stop_token token_;
92   mutable std::error_code ec_; 93   mutable std::error_code ec_;
93   mutable std::size_t bytes_transferred_ = 0; 94   mutable std::size_t bytes_transferred_ = 0;
94   95  
HITCBC 95   4 native_read_awaitable( 96   6 native_read_awaitable(
96   native_local_stream_socket& self, 97   native_local_stream_socket& self,
97   MutableBufferSequence buffers) noexcept 98   MutableBufferSequence buffers) noexcept
HITCBC 98   4 : self_(self) 99   6 : self_(self)
HITCBC 99   4 , buffers_(std::move(buffers)) 100   6 , buffers_(std::move(buffers))
100   { 101   {
HITCBC 101   4 } 102   6 }
102   103  
HITCBC 103   4 bool await_ready() const noexcept 104   6 bool await_ready() const noexcept
104   { 105   {
ECB 105 - 4 return token_.stop_requested(); 106 + // A pre-set ec_ means the initiator failed before
  107 + // dispatch (e.g. a closed object).
HITGNC   108 + 6 return static_cast<bool>(ec_) || token_.stop_requested();
106   } 109   }
107   110  
HITCBC 108   4 [[nodiscard]] capy::io_result<std::size_t> await_resume() const noexcept 111   6 [[nodiscard]] capy::io_result<std::size_t> await_resume() const noexcept
109   { 112   {
HITCBC 110   4 if (token_.stop_requested()) 113   6 if (token_.stop_requested())
MISUBC 111   ✗ return {make_error_code(std::errc::operation_canceled), 0}; 114   ✗ return {make_error_code(std::errc::operation_canceled), 0};
HITCBC 112   4 return {ec_, bytes_transferred_}; 115   6 return {ec_, bytes_transferred_};
113   } 116   }
114   117  
HITCBC 115   4 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)
116   -> std::coroutine_handle<> 119   -> std::coroutine_handle<>
117   { 120   {
HITCBC 118   4 token_ = env->stop_token; 121   6 token_ = env->stop_token;
HITCBC 119   12 return self_.get_impl().read_some( 122   18 return self_.get_impl().read_some(
HITCBC 120   12 h, env->executor, buffers_, token_, &ec_, &bytes_transferred_); 123   18 h, env->executor, buffers_, token_, &ec_, &bytes_transferred_);
121   } 124   }
122   }; 125   };
123   126  
124   template<class ConstBufferSequence> 127   template<class ConstBufferSequence>
125   struct native_write_awaitable 128   struct native_write_awaitable
126   { 129   {
127   native_local_stream_socket& self_; 130   native_local_stream_socket& self_;
128   ConstBufferSequence buffers_; 131   ConstBufferSequence buffers_;
129   std::stop_token token_; 132   std::stop_token token_;
130   mutable std::error_code ec_; 133   mutable std::error_code ec_;
131   mutable std::size_t bytes_transferred_ = 0; 134   mutable std::size_t bytes_transferred_ = 0;
132   135  
HITCBC 133   4 native_write_awaitable( 136   6 native_write_awaitable(
134   native_local_stream_socket& self, 137   native_local_stream_socket& self,
135   ConstBufferSequence buffers) noexcept 138   ConstBufferSequence buffers) noexcept
HITCBC 136   4 : self_(self) 139   6 : self_(self)
HITCBC 137   4 , buffers_(std::move(buffers)) 140   6 , buffers_(std::move(buffers))
138   { 141   {
HITCBC 139   4 } 142   6 }
140   143  
HITCBC 141   4 bool await_ready() const noexcept 144   6 bool await_ready() const noexcept
142   { 145   {
ECB 143 - 4 return token_.stop_requested(); 146 + // A pre-set ec_ means the initiator failed before
  147 + // dispatch (e.g. a closed object).
HITGNC   148 + 6 return static_cast<bool>(ec_) || token_.stop_requested();
144   } 149   }
145   150  
HITCBC 146   4 [[nodiscard]] capy::io_result<std::size_t> await_resume() const noexcept 151   6 [[nodiscard]] capy::io_result<std::size_t> await_resume() const noexcept
147   { 152   {
HITCBC 148   4 if (token_.stop_requested()) 153   6 if (token_.stop_requested())
MISUBC 149   ✗ return {make_error_code(std::errc::operation_canceled), 0}; 154   ✗ return {make_error_code(std::errc::operation_canceled), 0};
HITCBC 150   4 return {ec_, bytes_transferred_}; 155   6 return {ec_, bytes_transferred_};
151   } 156   }
152   157  
HITCBC 153   4 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)
154   -> std::coroutine_handle<> 159   -> std::coroutine_handle<>
155   { 160   {
HITCBC 156   4 token_ = env->stop_token; 161   6 token_ = env->stop_token;
HITCBC 157   12 return self_.get_impl().write_some( 162   18 return self_.get_impl().write_some(
HITCBC 158   12 h, env->executor, buffers_, token_, &ec_, &bytes_transferred_); 163   18 h, env->executor, buffers_, token_, &ec_, &bytes_transferred_);
159   } 164   }
160   }; 165   };
161   166  
162   struct native_wait_awaitable 167   struct native_wait_awaitable
163   { 168   {
164   native_local_stream_socket& self_; 169   native_local_stream_socket& self_;
165   wait_type w_; 170   wait_type w_;
166   std::stop_token token_; 171   std::stop_token token_;
167   mutable std::error_code ec_; 172   mutable std::error_code ec_;
168   173  
HITCBC 169   2 native_wait_awaitable( 174   4 native_wait_awaitable(
170   native_local_stream_socket& self, wait_type w) noexcept 175   native_local_stream_socket& self, wait_type w) noexcept
HITCBC 171   2 : self_(self) 176   4 : self_(self)
HITCBC 172   2 , w_(w) 177   4 , w_(w)
173   { 178   {
HITCBC 174   2 } 179   4 }
175   180  
HITCBC 176   2 bool await_ready() const noexcept 181   4 bool await_ready() const noexcept
177   { 182   {
ECB 178 - 2 return token_.stop_requested(); 183 + // A pre-set ec_ means the initiator failed before
  184 + // dispatch (e.g. auto-open).
HITGNC   185 + 4 return static_cast<bool>(ec_) || token_.stop_requested();
179   } 186   }
180   187  
HITCBC 181   2 [[nodiscard]] capy::io_result<> await_resume() const noexcept 188   4 [[nodiscard]] capy::io_result<> await_resume() const noexcept
182   { 189   {
HITCBC 183   2 if (token_.stop_requested()) 190   4 if (token_.stop_requested())
MISUBC 184   ✗ return {make_error_code(std::errc::operation_canceled)}; 191   ✗ return {make_error_code(std::errc::operation_canceled)};
HITCBC 185   2 return {ec_}; 192   4 return {ec_};
186   } 193   }
187   194  
HITCBC 188   2 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)
189   -> std::coroutine_handle<> 196   -> std::coroutine_handle<>
190   { 197   {
HITCBC 191   2 token_ = env->stop_token; 198   4 token_ = env->stop_token;
HITCBC 192   6 return self_.get_impl().wait( 199   12 return self_.get_impl().wait(
HITCBC 193   6 h, env->executor, w_, token_, &ec_); 200   12 h, env->executor, w_, token_, &ec_);
194   } 201   }
195   }; 202   };
196   203  
197   struct native_connect_awaitable 204   struct native_connect_awaitable
198   { 205   {
199   native_local_stream_socket& self_; 206   native_local_stream_socket& self_;
200   corosio::local_endpoint endpoint_; 207   corosio::local_endpoint endpoint_;
201   std::stop_token token_; 208   std::stop_token token_;
202   mutable std::error_code ec_; 209   mutable std::error_code ec_;
203   210  
HITCBC 204   10 native_connect_awaitable( 211   10 native_connect_awaitable(
205   native_local_stream_socket& self, 212   native_local_stream_socket& self,
206   corosio::local_endpoint ep) noexcept 213   corosio::local_endpoint ep) noexcept
HITCBC 207   10 : self_(self) 214   10 : self_(self)
HITCBC 208   10 , endpoint_(ep) 215   10 , endpoint_(ep)
209   { 216   {
HITCBC 210   10 } 217   10 }
211   218  
HITCBC 212   10 bool await_ready() const noexcept 219   10 bool await_ready() const noexcept
213   { 220   {
ECB 214 - 10 return token_.stop_requested(); 221 + // A pre-set ec_ means the initiator failed before
  222 + // dispatch (e.g. a closed object).
HITGNC   223 + 10 return static_cast<bool>(ec_) || token_.stop_requested();
215   } 224   }
216   225  
HITCBC 217   10 [[nodiscard]] capy::io_result<> await_resume() const noexcept 226   10 [[nodiscard]] capy::io_result<> await_resume() const noexcept
218   { 227   {
HITCBC 219   10 if (token_.stop_requested()) 228   10 if (token_.stop_requested())
MISUBC 220   ✗ return {make_error_code(std::errc::operation_canceled)}; 229   ✗ return {make_error_code(std::errc::operation_canceled)};
HITCBC 221   10 return {ec_}; 230   10 return {ec_};
222   } 231   }
223   232  
HITCBC 224   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)
225   -> std::coroutine_handle<> 234   -> std::coroutine_handle<>
226   { 235   {
HITCBC 227   10 token_ = env->stop_token; 236   10 token_ = env->stop_token;
HITCBC 228   30 return self_.get_impl().connect( 237   30 return self_.get_impl().connect(
HITCBC 229   30 h, env->executor, endpoint_, token_, &ec_); 238   30 h, env->executor, endpoint_, token_, &ec_);
230   } 239   }
231   }; 240   };
232   241  
233   public: 242   public:
234   /** Construct a native socket from an execution context. 243   /** Construct a native socket from an execution context.
235   244  
236   @param ctx The execution context that will own this socket. 245   @param ctx The execution context that will own this socket.
237   */ 246   */
HITCBC 238   28 explicit native_local_stream_socket(capy::execution_context& ctx) 247   32 explicit native_local_stream_socket(capy::execution_context& ctx)
HITCBC 239   28 : io_object(create_handle<service_type>(ctx)) 248   32 : io_object(create_handle<service_type>(ctx))
240   { 249   {
HITCBC 241   28 } 250   32 }
242   251  
243   /** Construct a native socket from an executor. 252   /** Construct a native socket from an executor.
244   253  
245   @param ex The executor whose context will own the socket. 254   @param ex The executor whose context will own the socket.
246   */ 255   */
247   template<class Ex> 256   template<class Ex>
248   requires(!std::same_as< 257   requires(!std::same_as<
249   std::remove_cvref_t<Ex>, 258   std::remove_cvref_t<Ex>,
250   native_local_stream_socket>) && 259   native_local_stream_socket>) &&
251   capy::Executor<Ex> 260   capy::Executor<Ex>
252   explicit native_local_stream_socket(Ex const& ex) 261   explicit native_local_stream_socket(Ex const& ex)
253   : native_local_stream_socket(ex.context()) 262   : native_local_stream_socket(ex.context())
254   { 263   {
255   } 264   }
256   265  
257   /// Move construct. 266   /// Move construct.
HITCBC 258   2 native_local_stream_socket(native_local_stream_socket&&) noexcept = default; 267   4 native_local_stream_socket(native_local_stream_socket&&) noexcept = default;
259   268  
260   /// Move assign. 269   /// Move assign.
261   native_local_stream_socket& 270   native_local_stream_socket&
262   operator=(native_local_stream_socket&&) noexcept = default; 271   operator=(native_local_stream_socket&&) noexcept = default;
263   272  
264   native_local_stream_socket(native_local_stream_socket const&) = delete; 273   native_local_stream_socket(native_local_stream_socket const&) = delete;
265   native_local_stream_socket& 274   native_local_stream_socket&
266   operator=(native_local_stream_socket const&) = delete; 275   operator=(native_local_stream_socket const&) = delete;
267   276  
268   /** Asynchronously read data from the socket. 277   /** Asynchronously read data from the socket.
269   278  
270   Calls the backend implementation directly, bypassing virtual 279   Calls the backend implementation directly, bypassing virtual
271   dispatch. Otherwise identical to @ref io_stream::read_some. 280   dispatch. Otherwise identical to @ref io_stream::read_some.
272   281  
273   @param buffers The buffer sequence to read into. 282   @param buffers The buffer sequence to read into.
274   283  
275   @return An awaitable yielding `(error_code, std::size_t)`. 284   @return An awaitable yielding `(error_code, std::size_t)`.
276   */ 285   */
277   template<capy::MutableBufferSequence MB> 286   template<capy::MutableBufferSequence MB>
HITCBC 278 - 4 auto read_some(MB const& buffers) 287 + 6 [[nodiscard]] auto read_some(MB const& buffers)
279   { 288   {
HITCBC 280   4 return native_read_awaitable<MB>(*this, buffers); 289   6 return native_read_awaitable<MB>(*this, buffers);
281   } 290   }
282   291  
283   /** Asynchronously write data to the socket. 292   /** Asynchronously write data to the socket.
284   293  
285   Calls the backend implementation directly, bypassing virtual 294   Calls the backend implementation directly, bypassing virtual
286   dispatch. Otherwise identical to @ref io_stream::write_some. 295   dispatch. Otherwise identical to @ref io_stream::write_some.
287   296  
288   @param buffers The buffer sequence to write from. 297   @param buffers The buffer sequence to write from.
289   298  
290   @return An awaitable yielding `(error_code, std::size_t)`. 299   @return An awaitable yielding `(error_code, std::size_t)`.
291   */ 300   */
292   template<capy::ConstBufferSequence CB> 301   template<capy::ConstBufferSequence CB>
HITCBC 293 - 4 auto write_some(CB const& buffers) 302 + 6 [[nodiscard]] auto write_some(CB const& buffers)
294   { 303   {
HITCBC 295   4 return native_write_awaitable<CB>(*this, buffers); 304   6 return native_write_awaitable<CB>(*this, buffers);
296   } 305   }
297   306  
298   /** Asynchronously connect to a remote endpoint. 307   /** Asynchronously connect to a remote endpoint.
299   308  
300   Calls the backend implementation directly, bypassing virtual 309   Calls the backend implementation directly, bypassing virtual
301   dispatch. Otherwise identical to @ref local_stream_socket::connect. 310   dispatch. Otherwise identical to @ref local_stream_socket::connect.
302   311  
303   If the socket is not already open, it is opened automatically. 312   If the socket is not already open, it is opened automatically.
304   313  
305   @param ep The local endpoint (path) to connect to. 314   @param ep The local endpoint (path) to connect to.
306   315  
307   @return An awaitable yielding `io_result<>`. 316   @return An awaitable yielding `io_result<>`.
308   317  
309 - @throws std::system_error if the socket needs to be opened 318 + If the socket needs to be opened and the open fails, the
310 - and the open fails. 319 + awaitable completes immediately with that error.
311   */ 320   */
HITCBC 312 - 10 auto connect(corosio::local_endpoint ep) 321 + 10 [[nodiscard]] auto connect(corosio::local_endpoint ep)
313   { 322   {
HITGNC   323 + 10 native_connect_awaitable aw(*this, ep);
HITCBC 314   10 if (!is_open()) 324   10 if (!is_open())
HITCBC 315 - 10 open(); 325 + 10 aw.ec_ = open();
HITCBC 316 - 10 return native_connect_awaitable(*this, ep); 326 + 10 return aw;
317   } 327   }
318   328  
319   /** Asynchronously wait for the socket to be ready. 329   /** Asynchronously wait for the socket to be ready.
320   330  
321   Calls the backend implementation directly, bypassing virtual 331   Calls the backend implementation directly, bypassing virtual
322   dispatch. Otherwise identical to @ref local_stream_socket::wait. 332   dispatch. Otherwise identical to @ref local_stream_socket::wait.
323   333  
324   @param w The wait direction (read, write, or error). 334   @param w The wait direction (read, write, or error).
325   335  
326   @return An awaitable yielding `io_result<>`. 336   @return An awaitable yielding `io_result<>`.
327   */ 337   */
HITCBC 328   2 [[nodiscard]] auto wait(wait_type w) 338   4 [[nodiscard]] auto wait(wait_type w)
329   { 339   {
HITCBC 330   2 return native_wait_awaitable(*this, w); 340   4 return native_wait_awaitable(*this, w);
331   } 341   }
332   }; 342   };
333   343  
334   } // namespace boost::corosio 344   } // namespace boost::corosio
335   345  
336   #endif // BOOST_COROSIO_NATIVE_NATIVE_LOCAL_STREAM_SOCKET_HPP 346   #endif // BOOST_COROSIO_NATIVE_NATIVE_LOCAL_STREAM_SOCKET_HPP