. /include/boost/capy /test /stream.hpp
99.39% Lines (163/164) 100.00% Functions (26/26)
[ ] prev/next file n p prev/next uncovered line
TLA Baseline Branch
Line Hits Code Line Hits Code
1   // 1   //
2   // Copyright (c) 2025 Vinnie Falco (vinnie.falco@gmail.com) 2   // Copyright (c) 2025 Vinnie Falco (vinnie.falco@gmail.com)
3   // Copyright (c) 2026 Michael Vandeberg 3   // Copyright (c) 2026 Michael Vandeberg
4   // 4   //
5   // Distributed under the Boost Software License, Version 1.0. (See accompanying 5   // Distributed under the Boost Software License, Version 1.0. (See accompanying
6   // 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)
7   // 7   //
8   // Official repository: https://github.com/cppalliance/capy 8   // Official repository: https://github.com/cppalliance/capy
9   // 9   //
10   10  
11   #ifndef BOOST_CAPY_TEST_STREAM_HPP 11   #ifndef BOOST_CAPY_TEST_STREAM_HPP
12   #define BOOST_CAPY_TEST_STREAM_HPP 12   #define BOOST_CAPY_TEST_STREAM_HPP
13   13  
14   #include <boost/capy/detail/config.hpp> 14   #include <boost/capy/detail/config.hpp>
15   #include <boost/capy/buffers.hpp> 15   #include <boost/capy/buffers.hpp>
16   #include <boost/capy/buffers/buffer_copy.hpp> 16   #include <boost/capy/buffers/buffer_copy.hpp>
17   #include <boost/capy/buffers/make_buffer.hpp> 17   #include <boost/capy/buffers/make_buffer.hpp>
18   #include <boost/capy/continuation.hpp> 18   #include <boost/capy/continuation.hpp>
19   #include <coroutine> 19   #include <coroutine>
20 - #include <boost/capy/ex/io_env.hpp> 20 + #include <boost/capy/ex/await_context.hpp>
21 - #include <boost/capy/io_result.hpp> 21 + #include <boost/capy/results.hpp>
22   #include <boost/capy/error.hpp> 22   #include <boost/capy/error.hpp>
23   #include <boost/capy/read.hpp> 23   #include <boost/capy/read.hpp>
24   #include <boost/capy/task.hpp> 24   #include <boost/capy/task.hpp>
25   #include <boost/capy/test/fuse.hpp> 25   #include <boost/capy/test/fuse.hpp>
26   #include <boost/capy/test/run_blocking.hpp> 26   #include <boost/capy/test/run_blocking.hpp>
27   27  
28   #include <atomic> 28   #include <atomic>
29   #include <memory> 29   #include <memory>
30   #include <new> 30   #include <new>
31   #include <stop_token> 31   #include <stop_token>
32   #include <string> 32   #include <string>
33   #include <string_view> 33   #include <string_view>
34   #include <utility> 34   #include <utility>
35   35  
36   namespace boost { 36   namespace boost {
37   namespace capy { 37   namespace capy {
38   namespace test { 38   namespace test {
39   39  
40   /** Suspends a reader until its paired end writes, or the shared fuse injects an error. 40   /** Suspends a reader until its paired end writes, or the shared fuse injects an error.
41   41  
42   Streams are created in pairs via @ref make_stream_pair. 42   Streams are created in pairs via @ref make_stream_pair.
43   Data written to one end becomes available for reading on 43   Data written to one end becomes available for reading on
44   the other. If no data is available when @ref read_some 44   the other. If no data is available when @ref read_some
45   is called, the calling coroutine suspends until the peer 45   is called, the calling coroutine suspends until the peer
46   calls @ref write_some. The shared @ref fuse enables error 46   calls @ref write_some. The shared @ref fuse enables error
47   injection at controlled points in both directions. 47   injection at controlled points in both directions.
48   48  
49   When the fuse injects an error or throws on one end, the 49   When the fuse injects an error or throws on one end, the
50   pair is automatically closed. Any suspended reader on 50   pair is automatically closed. Any suspended reader on
51   either end is resumed with `error::eof`, and subsequent 51   either end is resumed with `error::eof`, and subsequent
52   operations on both ends return `error::eof`. Calling 52   operations on both ends return `error::eof`. Calling
53   @ref close on one end signals eof to the peer's reads 53   @ref close on one end signals eof to the peer's reads
54   after draining any buffered data, while the peer may 54   after draining any buffered data, while the peer may
55   still write. 55   still write.
56   56  
57   @par Thread Safety 57   @par Thread Safety
58   Single-threaded only. Both ends of the pair must be 58   Single-threaded only. Both ends of the pair must be
59   accessed from the same thread. Concurrent access is 59   accessed from the same thread. Concurrent access is
60   undefined behavior. 60   undefined behavior.
61   61  
62   @par Example 62   @par Example
63   @par !example example 63   @par !example example
64   64  
65   65  
66   @see make_stream_pair, fuse 66   @see make_stream_pair, fuse
67   */ 67   */
68   class stream 68   class stream
69   { 69   {
70   // Single-threaded only. No concurrent access to either 70   // Single-threaded only. No concurrent access to either
71   // end of the pair. Both streams and all operations must 71   // end of the pair. Both streams and all operations must
72   // run on the same thread. 72   // run on the same thread.
73   73  
74   struct half 74   struct half
75   { 75   {
76   std::string buf; 76   std::string buf;
77   std::size_t max_read_size = std::size_t(-1); 77   std::size_t max_read_size = std::size_t(-1);
78   // The parked read's own continuation: an executor may queue it 78   // The parked read's own continuation: an executor may queue it
79   // in place, so it is never copied or changed once posted. 79   // in place, so it is never copied or changed once posted.
80   continuation* pending_cont_ = nullptr; 80   continuation* pending_cont_ = nullptr;
81   executor_ref pending_ex; 81   executor_ref pending_ex;
82   // Points at the suspended reader's claim flag (owned by the 82   // Points at the suspended reader's claim flag (owned by the
83   // read awaitable). Lets a peer wake coordinate with a stop 83   // read awaitable). Lets a peer wake coordinate with a stop
84   // callback so the parked read is resumed exactly once. 84   // callback so the parked read is resumed exactly once.
85   std::atomic<bool>* pending_claimed = nullptr; 85   std::atomic<bool>* pending_claimed = nullptr;
86   bool eof = false; 86   bool eof = false;
87   }; 87   };
88   88  
89   struct state 89   struct state
90   { 90   {
91   fuse f; 91   fuse f;
92   bool closed = false; 92   bool closed = false;
93   half sides[2]; 93   half sides[2];
94   94  
HITCBC 95   308 explicit state(fuse f_) noexcept 95   308 explicit state(fuse f_) noexcept
HITCBC 96   924 : f(std::move(f_)) 96   924 : f(std::move(f_))
97   { 97   {
HITCBC 98   308 } 98   308 }
99   99  
100   // Resume a suspended reader on this side, if any. Claims the 100   // Resume a suspended reader on this side, if any. Claims the
101   // reader's atomic so it is never double-resumed by a racing 101   // reader's atomic so it is never double-resumed by a racing
102   // stop callback; the loser of the race skips the post. 102   // stop callback; the loser of the race skips the post.
HITCBC 103   696 static void wake(half& side) 103   696 static void wake(half& side)
104   { 104   {
HITCBC 105   696 auto* c = std::exchange(side.pending_cont_, nullptr); 105   696 auto* c = std::exchange(side.pending_cont_, nullptr);
HITCBC 106   696 if(! c) 106   696 if(! c)
HITCBC 107   670 return; 107   670 return;
HITCBC 108   26 auto ex = std::exchange(side.pending_ex, {}); 108   26 auto ex = std::exchange(side.pending_ex, {});
HITCBC 109   26 auto* claimed = std::exchange(side.pending_claimed, nullptr); 109   26 auto* claimed = std::exchange(side.pending_claimed, nullptr);
HITCBC 110   52 if(! claimed || 110   52 if(! claimed ||
HITCBC 111   26 ! claimed->exchange(true, std::memory_order_acq_rel)) 111   26 ! claimed->exchange(true, std::memory_order_acq_rel))
112   { 112   {
HITCBC 113   26 ex.post(*c); 113   26 ex.post(*c);
114   } 114   }
115   } 115   }
116   116  
117   // Set closed and resume any suspended readers 117   // Set closed and resume any suspended readers
118   // with eof on both sides. 118   // with eof on both sides.
HITCBC 119   214 void close() 119   214 void close()
120   { 120   {
HITCBC 121   214 closed = true; 121   214 closed = true;
HITCBC 122   642 for(auto& side : sides) 122   642 for(auto& side : sides)
HITCBC 123   428 wake(side); 123   428 wake(side);
HITCBC 124   214 } 124   214 }
125   }; 125   };
126   126  
127   // Wraps the maybe_fail() call. If the guard is 127   // Wraps the maybe_fail() call. If the guard is
128   // not disarmed before destruction (fuse returned 128   // not disarmed before destruction (fuse returned
129   // an error, or threw an exception), closes both 129   // an error, or threw an exception), closes both
130   // ends so any suspended peer gets eof. 130   // ends so any suspended peer gets eof.
131   struct close_guard 131   struct close_guard
132   { 132   {
133   state* st; 133   state* st;
134   bool armed = true; 134   bool armed = true;
HITCBC 135   324 void disarm() noexcept { armed = false; } 135   324 void disarm() noexcept { armed = false; }
HITCBC 136   538 ~close_guard() noexcept(false) { if(armed) st->close(); } 136   538 ~close_guard() noexcept(false) { if(armed) st->close(); }
137   }; 137   };
138   138  
139   std::shared_ptr<state> state_; 139   std::shared_ptr<state> state_;
140   int index_; 140   int index_;
141   141  
HITCBC 142   616 stream( 142   616 stream(
143   std::shared_ptr<state> sp, 143   std::shared_ptr<state> sp,
144   int index) noexcept 144   int index) noexcept
HITCBC 145   616 : state_(std::move(sp)) 145   616 : state_(std::move(sp))
HITCBC 146   616 , index_(index) 146   616 , index_(index)
147   { 147   {
HITCBC 148   616 } 148   616 }
149   149  
150   friend std::pair<stream, stream> 150   friend std::pair<stream, stream>
151   make_stream_pair(fuse); 151   make_stream_pair(fuse);
152   152  
153   public: 153   public:
154   /** Copy construction is disabled; a stream end is move-only. 154   /** Copy construction is disabled; a stream end is move-only.
155   155  
156   @param other The stream end that would be copied. 156   @param other The stream end that would be copied.
157   */ 157   */
158   stream(stream const& other) = delete; 158   stream(stream const& other) = delete;
159   159  
160   /** Copy assignment is disabled; a stream end is move-only. 160   /** Copy assignment is disabled; a stream end is move-only.
161   161  
162   @param other The stream end that would be assigned from. 162   @param other The stream end that would be assigned from.
163   163  
164   @return A reference to `*this`. 164   @return A reference to `*this`.
165   */ 165   */
166   stream& operator=(stream const& other) = delete; 166   stream& operator=(stream const& other) = delete;
167   167  
168   /** Move constructor. 168   /** Move constructor.
169   169  
170   @param other The stream end to move from. 170   @param other The stream end to move from.
171   */ 171   */
HITCBC 172   718 stream(stream&& other) = default; 172   718 stream(stream&& other) = default;
173   173  
174   /** Move assignment. 174   /** Move assignment.
175   175  
176   @param other The stream end to move from. 176   @param other The stream end to move from.
177   177  
178   @return A reference to `*this`. 178   @return A reference to `*this`.
179   */ 179   */
180   stream& operator=(stream&& other) = default; 180   stream& operator=(stream&& other) = default;
181   181  
182   /** Signal end-of-stream to the peer. 182   /** Signal end-of-stream to the peer.
183   183  
184   Marks the peer's read direction as closed. 184   Marks the peer's read direction as closed.
185   If the peer is suspended in @ref read_some, 185   If the peer is suspended in @ref read_some,
186   it is resumed. The peer drains any buffered 186   it is resumed. The peer drains any buffered
187   data before receiving `error::eof`. Writes 187   data before receiving `error::eof`. Writes
188   from the peer are unaffected. 188   from the peer are unaffected.
189   */ 189   */
190   void 190   void
HITCBC 191   6 close() 191   6 close()
192   { 192   {
HITCBC 193   6 int peer = 1 - index_; 193   6 int peer = 1 - index_;
HITCBC 194   6 auto& side = state_->sides[peer]; 194   6 auto& side = state_->sides[peer];
HITCBC 195   6 side.eof = true; 195   6 side.eof = true;
HITCBC 196   6 state::wake(side); 196   6 state::wake(side);
HITCBC 197   6 } 197   6 }
198   198  
199   /** Set the maximum bytes returned per read. 199   /** Set the maximum bytes returned per read.
200   200  
201   Limits how many bytes @ref read_some returns in 201   Limits how many bytes @ref read_some returns in
202   a single call, simulating chunked network delivery. 202   a single call, simulating chunked network delivery.
203   The default is unlimited. 203   The default is unlimited.
204   204  
205   @param n Maximum bytes per read. 205   @param n Maximum bytes per read.
206   */ 206   */
207   void 207   void
HITCBC 208   55 set_max_read_size(std::size_t n) noexcept 208   55 set_max_read_size(std::size_t n) noexcept
209   { 209   {
HITCBC 210   55 state_->sides[index_].max_read_size = n; 210   55 state_->sides[index_].max_read_size = n;
HITCBC 211   55 } 211   55 }
212   212  
213   /** Asynchronously read data from the stream. 213   /** Asynchronously read data from the stream.
214   214  
215   Transfers up to `buffer_size(buffers)` bytes from 215   Transfers up to `buffer_size(buffers)` bytes from
216   data written by the peer. If no data is available, 216   data written by the peer. If no data is available,
217   the calling coroutine suspends until the peer calls 217   the calling coroutine suspends until the peer calls
218   @ref write_some. Before every read, the attached 218   @ref write_some. Before every read, the attached
219   @ref fuse is consulted to possibly inject an error. 219   @ref fuse is consulted to possibly inject an error.
220   If the fuse fires, the pair is automatically closed. 220   If the fuse fires, the pair is automatically closed.
221   If the stream is closed, returns `error::eof`. 221   If the stream is closed, returns `error::eof`.
222   The returned `std::size_t` is the number of bytes 222   The returned `std::size_t` is the number of bytes
223   transferred. 223   transferred.
224   224  
225   @param buffers The mutable buffer sequence to receive data. 225   @param buffers The mutable buffer sequence to receive data.
226   226  
227   @return An awaitable that await-returns `(error_code,std::size_t)`. 227   @return An awaitable that await-returns `(error_code,std::size_t)`.
228   228  
229   @par Cancellation 229   @par Cancellation
230   Cancellation applies only to a read that would otherwise suspend. 230   Cancellation applies only to a read that would otherwise suspend.
231   If no data is available and the environment's stop token is 231   If no data is available and the environment's stop token is
232   requested, before or during the wait, the read resumes with 232   requested, before or during the wait, the read resumes with
233   `error::canceled`. A read that can complete immediately from 233   `error::canceled`. A read that can complete immediately from
234   buffered data is unaffected by the stop token. 234   buffered data is unaffected by the stop token.
235   235  
236   @see fuse, close 236   @see fuse, close
237   */ 237   */
238   template<MutableBufferSequence MB> 238   template<MutableBufferSequence MB>
239   auto 239   auto
HITCBC 240   299 read_some(MB buffers) 240   299 read_some(MB buffers)
241   { 241   {
242   // The read suspends when no data is available, parking its 242   // The read suspends when no data is available, parking its
243   // continuation on the side until the peer writes/closes. To 243   // continuation on the side until the peer writes/closes. To
244   // support cancellation it follows the same pattern as 244   // support cancellation it follows the same pattern as
245   // async_waker::wait_awaiter: a stop callback claims the resume 245   // async_waker::wait_awaiter: a stop callback claims the resume
246   // (racing the peer wake via an atomic) and posts the continuation 246   // (racing the peer wake via an atomic) and posts the continuation
247   // through the executor. Because it owns a std::atomic and a 247   // through the executor. Because it owns a std::atomic and a
248   // std::stop_callback, the awaitable needs explicit move and 248   // std::stop_callback, the awaitable needs explicit move and
249   // destruction (the task promise moves it into its 249   // destruction (the task promise moves it into its
250   // transform_awaiter before awaiting). 250   // transform_awaiter before awaiting).
251   struct awaitable 251   struct awaitable
252   { 252   {
253   stream* self_; 253   stream* self_;
254   MB buffers_; 254   MB buffers_;
255   255  
256   // Declared before stop_cb_buf_: the stop callback reads 256   // Declared before stop_cb_buf_: the stop callback reads
257   // these, so they must outlive a blocking stop_cb_ destructor. 257   // these, so they must outlive a blocking stop_cb_ destructor.
258   continuation cont_; 258   continuation cont_;
259   executor_ref ex_; 259   executor_ref ex_;
260   half* side_ = nullptr; 260   half* side_ = nullptr;
261   std::atomic<bool> claimed_{false}; 261   std::atomic<bool> claimed_{false};
262   bool canceled_ = false; 262   bool canceled_ = false;
263   bool stop_cb_active_ = false; 263   bool stop_cb_active_ = false;
264   264  
265   struct cancel_fn 265   struct cancel_fn
266   { 266   {
267   awaitable* self_; 267   awaitable* self_;
268   268  
HITCBC 269   14 void operator()() const noexcept 269   14 void operator()() const noexcept
270   { 270   {
HITCBC 271   14 if(! self_->claimed_.exchange( 271   14 if(! self_->claimed_.exchange(
272   true, std::memory_order_acq_rel)) 272   true, std::memory_order_acq_rel))
273   { 273   {
HITCBC 274   2 self_->canceled_ = true; 274   2 self_->canceled_ = true;
HITCBC 275   2 self_->ex_.post(self_->cont_); 275   2 self_->ex_.post(self_->cont_);
276   } 276   }
HITCBC 277   14 } 277   14 }
278   }; 278   };
279   279  
280   using stop_cb_t = std::stop_callback<cancel_fn>; 280   using stop_cb_t = std::stop_callback<cancel_fn>;
281   281  
282   // Declared last: its destructor may block while the callback 282   // Declared last: its destructor may block while the callback
283   // accesses the members above. A union gives correct alignment 283   // accesses the members above. A union gives correct alignment
284   // for stop_cb_t without an alignas specifier, which avoids 284   // for stop_cb_t without an alignas specifier, which avoids
285   // MSVC's C4324 padding warning on this function-local class 285   // MSVC's C4324 padding warning on this function-local class
286   // (the member-level pragma used by async_waker::wait_awaiter 286   // (the member-level pragma used by async_waker::wait_awaiter
287   // does not suppress it here). Lifetime is managed manually: 287   // does not suppress it here). Lifetime is managed manually:
288   // placement new in await_suspend, explicit destruction once done. 288   // placement new in await_suspend, explicit destruction once done.
289   union { stop_cb_t stop_cb_; }; 289   union { stop_cb_t stop_cb_; };
290   290  
HITCBC 291   299 awaitable(stream* self, MB buffers) noexcept 291   299 awaitable(stream* self, MB buffers) noexcept
HITCBC 292   299 : self_(self) 292   299 : self_(self)
HITCBC 293   299 , buffers_(buffers) 293   299 , buffers_(buffers)
294   { 294   {
HITCBC 295   299 } 295   299 }
296   296  
297   /// @pre Not yet awaited (no active stop callback). 297   /// @pre Not yet awaited (no active stop callback).
HITCBC 298   293 awaitable(awaitable&& o) noexcept 298   293 awaitable(awaitable&& o) noexcept
HITCBC 299   293 : self_(o.self_) 299   293 : self_(o.self_)
HITCBC 300   293 , buffers_(o.buffers_) 300   293 , buffers_(o.buffers_)
HITCBC 301   293 , cont_(o.cont_) 301   293 , cont_(o.cont_)
HITCBC 302   293 , ex_(o.ex_) 302   293 , ex_(o.ex_)
HITCBC 303   293 , side_(o.side_) 303   293 , side_(o.side_)
HITCBC 304   293 , claimed_(o.claimed_.load(std::memory_order_relaxed)) 304   293 , claimed_(o.claimed_.load(std::memory_order_relaxed))
HITCBC 305   293 , canceled_(o.canceled_) 305   293 , canceled_(o.canceled_)
HITCBC 306   293 , stop_cb_active_(std::exchange(o.stop_cb_active_, false)) 306   293 , stop_cb_active_(std::exchange(o.stop_cb_active_, false))
307   { 307   {
HITCBC 308   293 } 308   293 }
309   309  
HITCBC 310   592 ~awaitable() 310   592 ~awaitable()
311   { 311   {
HITCBC 312   592 if(stop_cb_active_) 312   592 if(stop_cb_active_)
HITCBC 313   1 stop_cb_.~stop_cb_t(); 313   1 stop_cb_.~stop_cb_t();
314   // Unlink from the side if still parked (e.g. the 314   // Unlink from the side if still parked (e.g. the
315   // coroutine was destroyed while suspended), so a later 315   // coroutine was destroyed while suspended), so a later
316   // peer wake does not dereference a freed claim flag. 316   // peer wake does not dereference a freed claim flag.
HITCBC 317   592 if(side_ && side_->pending_claimed == &claimed_) 317   592 if(side_ && side_->pending_claimed == &claimed_)
318   { 318   {
HITCBC 319   1 side_->pending_cont_ = nullptr; 319   1 side_->pending_cont_ = nullptr;
HITCBC 320   1 side_->pending_ex = {}; 320   1 side_->pending_ex = {};
HITCBC 321   1 side_->pending_claimed = nullptr; 321   1 side_->pending_claimed = nullptr;
322   } 322   }
HITCBC 323   592 } 323   592 }
324   324  
325   awaitable(awaitable const&) = delete; 325   awaitable(awaitable const&) = delete;
326   awaitable& operator=(awaitable const&) = delete; 326   awaitable& operator=(awaitable const&) = delete;
327   awaitable& operator=(awaitable&&) = delete; 327   awaitable& operator=(awaitable&&) = delete;
328   328  
HITCBC 329   299 bool await_ready() const noexcept 329   299 bool await_ready() const noexcept
330   { 330   {
HITCBC 331   299 if(buffer_empty(buffers_)) 331   299 if(buffer_empty(buffers_))
HITCBC 332   8 return true; 332   8 return true;
HITCBC 333   291 auto* st = self_->state_.get(); 333   291 auto* st = self_->state_.get();
HITCBC 334   291 auto& side = st->sides[self_->index_]; 334   291 auto& side = st->sides[self_->index_];
HITCBC 335   574 return st->closed || side.eof || 335   574 return st->closed || side.eof ||
HITCBC 336   574 !side.buf.empty(); 336   574 !side.buf.empty();
337   } 337   }
338   338  
HITCBC 339   31 bool await_suspend( 339   31 bool await_suspend(
340   std::coroutine_handle<> h, 340   std::coroutine_handle<> h,
341 - io_env const* env) noexcept 341 + await_context const* env) noexcept
342   { 342   {
343   // A callback registered after stop was requested would 343   // A callback registered after stop was requested would
344   // fire inline and post this coroutine, which could then be 344   // fire inline and post this coroutine, which could then be
345   // resumed and finished before the registration returns. 345   // resumed and finished before the registration returns.
HITCBC 346   31 if(env->stop_token.stop_requested()) 346   31 if(env->stop_token.stop_requested())
347   { 347   {
HITCBC 348   2 canceled_ = true; 348   2 canceled_ = true;
HITCBC 349   2 return false; 349   2 return false;
350   } 350   }
HITCBC 351   29 auto& side = self_->state_->sides[ 351   29 auto& side = self_->state_->sides[
HITCBC 352   29 self_->index_]; 352   29 self_->index_];
HITCBC 353   29 cont_.h = h; 353   29 cont_.h = h;
HITCBC 354   29 cont_.owner = env->owner; 354   29 cont_.owner = env->owner;
HITCBC 355   29 ex_ = env->executor; 355   29 ex_ = env->executor;
HITCBC 356   29 side_ = &side; 356   29 side_ = &side;
HITCBC 357   29 side.pending_cont_ = &cont_; 357   29 side.pending_cont_ = &cont_;
HITCBC 358   29 side.pending_ex = env->executor; 358   29 side.pending_ex = env->executor;
HITCBC 359   29 side.pending_claimed = &claimed_; 359   29 side.pending_claimed = &claimed_;
360   360  
HITCBC 361   29 ::new(static_cast<void*>(&stop_cb_)) stop_cb_t( 361   29 ::new(static_cast<void*>(&stop_cb_)) stop_cb_t(
HITCBC 362   29 env->stop_token, cancel_fn{this}); 362   29 env->stop_token, cancel_fn{this});
HITCBC 363   29 stop_cb_active_ = true; 363   29 stop_cb_active_ = true;
364   364  
HITCBC 365   29 return true; 365   29 return true;
366   } 366   }
367   367  
368 - [[nodiscard]] io_result<std::size_t> 368 + [[nodiscard]] results<std::size_t>
HITCBC 369   298 await_resume() 369   298 await_resume()
370   { 370   {
HITCBC 371   298 if(stop_cb_active_) 371   298 if(stop_cb_active_)
372   { 372   {
HITCBC 373   28 stop_cb_.~stop_cb_t(); 373   28 stop_cb_.~stop_cb_t();
HITCBC 374   28 stop_cb_active_ = false; 374   28 stop_cb_active_ = false;
375   } 375   }
376   376  
HITCBC 377   298 if(buffer_empty(buffers_)) 377   298 if(buffer_empty(buffers_))
HITCBC 378   8 return {std::error_code(), 0}; 378   8 return {std::error_code(), 0};
379   379  
HITCBC 380   290 if(canceled_) 380   290 if(canceled_)
381   { 381   {
382   // The stop callback posted us but left the side 382   // The stop callback posted us but left the side
383   // untouched; unlink if a peer wake has not already. 383   // untouched; unlink if a peer wake has not already.
HITCBC 384   4 if(side_ && side_->pending_claimed == &claimed_) 384   4 if(side_ && side_->pending_claimed == &claimed_)
385   { 385   {
HITCBC 386   2 side_->pending_cont_ = nullptr; 386   2 side_->pending_cont_ = nullptr;
HITCBC 387   2 side_->pending_ex = {}; 387   2 side_->pending_ex = {};
HITCBC 388   2 side_->pending_claimed = nullptr; 388   2 side_->pending_claimed = nullptr;
389   } 389   }
HITCBC 390   4 return {error::canceled, 0}; 390   4 return {error::canceled, 0};
391   } 391   }
392   392  
HITCBC 393   286 auto* st = self_->state_.get(); 393   286 auto* st = self_->state_.get();
HITCBC 394   286 auto& side = st->sides[ 394   286 auto& side = st->sides[
HITCBC 395   286 self_->index_]; 395   286 self_->index_];
396   396  
HITCBC 397   286 if(st->closed) 397   286 if(st->closed)
HITCBC 398   12 return {error::eof, 0}; 398   12 return {error::eof, 0};
399   399  
HITCBC 400   274 if(side.eof && side.buf.empty()) 400   274 if(side.eof && side.buf.empty())
HITCBC 401   6 return {error::eof, 0}; 401   6 return {error::eof, 0};
402   402  
HITCBC 403   268 if(!side.eof) 403   268 if(!side.eof)
404   { 404   {
HITCBC 405   265 close_guard g{st}; 405   265 close_guard g{st};
HITCBC 406   265 auto ec = st->f.maybe_fail(); 406   265 auto ec = st->f.maybe_fail();
HITCBC 407   211 if(ec) 407   211 if(ec)
HITCBC 408   54 return {ec, 0}; 408   54 return {ec, 0};
HITCBC 409   157 g.disarm(); 409   157 g.disarm();
HITCBC 410   265 } 410   265 }
411   411  
HITCBC 412   480 std::size_t const n = buffer_copy( 412   480 std::size_t const n = buffer_copy(
HITCBC 413   160 buffers_, make_buffer(side.buf), 413   160 buffers_, make_buffer(side.buf),
414   side.max_read_size); 414   side.max_read_size);
HITCBC 415   160 side.buf.erase(0, n); 415   160 side.buf.erase(0, n);
HITCBC 416   160 return {std::error_code(), n}; 416   160 return {std::error_code(), n};
417   } 417   }
418   }; 418   };
HITCBC 419   299 return awaitable{this, buffers}; 419   299 return awaitable{this, buffers};
420   } 420   }
421   421  
422   /** Asynchronously write data to the stream. 422   /** Asynchronously write data to the stream.
423   423  
424   Transfers up to `buffer_size(buffers)` bytes to the 424   Transfers up to `buffer_size(buffers)` bytes to the
425   peer's incoming buffer. If the peer is suspended in 425   peer's incoming buffer. If the peer is suspended in
426   @ref read_some, it is resumed. Before every write, 426   @ref read_some, it is resumed. Before every write,
427   the attached @ref fuse is consulted to possibly inject 427   the attached @ref fuse is consulted to possibly inject
428   an error. If the fuse fires, the pair is automatically 428   an error. If the fuse fires, the pair is automatically
429   closed. If the stream is closed, returns `error::eof`. 429   closed. If the stream is closed, returns `error::eof`.
430   The returned `std::size_t` is the number of bytes 430   The returned `std::size_t` is the number of bytes
431   transferred. 431   transferred.
432   432  
433   @param buffers The const buffer sequence containing 433   @param buffers The const buffer sequence containing
434   data to write. 434   data to write.
435   435  
436   @return An awaitable that await-returns `(error_code,std::size_t)`. 436   @return An awaitable that await-returns `(error_code,std::size_t)`.
437   437  
438   @par Cancellation 438   @par Cancellation
439   If the environment's stop token is requested, the write 439   If the environment's stop token is requested, the write
440   completes immediately with `error::canceled` and transfers no 440   completes immediately with `error::canceled` and transfers no
441   data. An empty buffer sequence is a no-op that completes 441   data. An empty buffer sequence is a no-op that completes
442   successfully regardless of the stop token. 442   successfully regardless of the stop token.
443   443  
444   @see fuse, close 444   @see fuse, close
445   */ 445   */
446   template<ConstBufferSequence CB> 446   template<ConstBufferSequence CB>
447   auto 447   auto
HITCBC 448   278 write_some(CB buffers) 448   278 write_some(CB buffers)
449   { 449   {
450   struct awaitable 450   struct awaitable
451   { 451   {
452   stream* self_; 452   stream* self_;
453   CB buffers_; 453   CB buffers_;
454   bool canceled_ = false; 454   bool canceled_ = false;
455   455  
HITCBC 456   278 bool await_ready() const noexcept { return false; } 456   278 bool await_ready() const noexcept { return false; }
457   457  
458   // The write completes synchronously; await_suspend is only 458   // The write completes synchronously; await_suspend is only
459   // used to observe the environment's stop token. Returning 459   // used to observe the environment's stop token. Returning
460   // false means the coroutine does not actually suspend. 460   // false means the coroutine does not actually suspend.
461   bool 461   bool
HITCBC 462   278 await_suspend( 462   278 await_suspend(
463   std::coroutine_handle<>, 463   std::coroutine_handle<>,
464 - io_env const* env) noexcept 464 + await_context const* env) noexcept
465   { 465   {
HITCBC 466   278 canceled_ = env->stop_token.stop_requested(); 466   278 canceled_ = env->stop_token.stop_requested();
HITCBC 467   278 return false; 467   278 return false;
468   } 468   }
469   469  
470 - [[nodiscard]] io_result<std::size_t> 470 + [[nodiscard]] results<std::size_t>
HITCBC 471   278 await_resume() 471   278 await_resume()
472   { 472   {
HITCBC 473   278 std::size_t n = buffer_size(buffers_); 473   278 std::size_t n = buffer_size(buffers_);
HITCBC 474   278 if(n == 0) 474   278 if(n == 0)
HITCBC 475   4 return {std::error_code(), 0}; 475   4 return {std::error_code(), 0};
476   476  
HITCBC 477   274 if(canceled_) 477   274 if(canceled_)
HITCBC 478   1 return {error::canceled, 0}; 478   1 return {error::canceled, 0};
479   479  
HITCBC 480   273 auto* st = self_->state_.get(); 480   273 auto* st = self_->state_.get();
481   481  
HITCBC 482   273 if(st->closed) 482   273 if(st->closed)
MISUBC 483   ✗ return {error::eof, 0}; 483   ✗ return {error::eof, 0};
484   484  
HITCBC 485   273 close_guard g{st}; 485   273 close_guard g{st};
HITCBC 486   273 auto ec = st->f.maybe_fail(); 486   273 auto ec = st->f.maybe_fail();
HITCBC 487   220 if(ec) 487   220 if(ec)
HITCBC 488   53 return {ec, 0}; 488   53 return {ec, 0};
HITCBC 489   167 g.disarm(); 489   167 g.disarm();
490   490  
HITCBC 491   167 int peer = 1 - self_->index_; 491   167 int peer = 1 - self_->index_;
HITCBC 492   167 auto& side = st->sides[peer]; 492   167 auto& side = st->sides[peer];
493   493  
HITCBC 494   167 std::size_t const old_size = side.buf.size(); 494   167 std::size_t const old_size = side.buf.size();
HITCBC 495   167 side.buf.resize(old_size + n); 495   167 side.buf.resize(old_size + n);
HITCBC 496   167 buffer_copy(make_buffer( 496   167 buffer_copy(make_buffer(
HITCBC 497   167 side.buf.data() + old_size, n), 497   167 side.buf.data() + old_size, n),
HITCBC 498   167 buffers_, n); 498   167 buffers_, n);
499   499  
HITCBC 500   167 state::wake(side); 500   167 state::wake(side);
501   501  
HITCBC 502   167 return {std::error_code(), n}; 502   167 return {std::error_code(), n};
HITCBC 503   273 } 503   273 }
504   }; 504   };
HITCBC 505   278 return awaitable{this, buffers}; 505   278 return awaitable{this, buffers};
506   } 506   }
507   507  
508   /** Inject data into this stream's peer for reading. 508   /** Inject data into this stream's peer for reading.
509   509  
510   Appends data directly to the peer's incoming buffer, 510   Appends data directly to the peer's incoming buffer,
511   bypassing the fuse. If the peer is suspended in 511   bypassing the fuse. If the peer is suspended in
512   @ref read_some, it is resumed. This is test setup, 512   @ref read_some, it is resumed. This is test setup,
513   not an operation under test. 513   not an operation under test.
514   514  
515   @param sv The data to inject. 515   @param sv The data to inject.
516   516  
517   @see make_stream_pair 517   @see make_stream_pair
518   */ 518   */
519   void 519   void
HITCBC 520   95 provide(std::string_view sv) 520   95 provide(std::string_view sv)
521   { 521   {
HITCBC 522   95 int peer = 1 - index_; 522   95 int peer = 1 - index_;
HITCBC 523   95 auto& side = state_->sides[peer]; 523   95 auto& side = state_->sides[peer];
HITCBC 524   95 side.buf.append(sv); 524   95 side.buf.append(sv);
HITCBC 525   95 state::wake(side); 525   95 state::wake(side);
HITCBC 526   95 } 526   95 }
527   527  
528   /** Read from this stream and verify the content. 528   /** Read from this stream and verify the content.
529   529  
530   Reads exactly `expected.size()` bytes from the stream 530   Reads exactly `expected.size()` bytes from the stream
531   and compares against the expected string. The read goes 531   and compares against the expected string. The read goes
532   through the normal path including the fuse. 532   through the normal path including the fuse.
533   533  
534   @param expected The expected content. 534   @param expected The expected content.
535   535  
536   @return A pair of `(error_code, bool)`. The error_code 536   @return A pair of `(error_code, bool)`. The error_code
537   is set if a read error occurs (e.g. fuse injection). 537   is set if a read error occurs (e.g. fuse injection).
538   The bool is true if the data matches. 538   The bool is true if the data matches.
539   539  
540   @see provide 540   @see provide
541   */ 541   */
542   std::pair<std::error_code, bool> 542   std::pair<std::error_code, bool>
HITCBC 543   38 expect(std::string_view expected) 543   38 expect(std::string_view expected)
544   { 544   {
HITCBC 545   38 std::error_code result; 545   38 std::error_code result;
HITCBC 546   38 bool match = false; 546   38 bool match = false;
HITCBC 547   141 run_blocking()([]( 547   141 run_blocking()([](
548   stream& self, 548   stream& self,
549   std::string_view expected, 549   std::string_view expected,
550   std::error_code& result, 550   std::error_code& result,
551   bool& match) -> task<> 551   bool& match) -> task<>
552   { 552   {
553   std::string buf(expected.size(), '\0'); 553   std::string buf(expected.size(), '\0');
554   auto [ec, n] = co_await read( 554   auto [ec, n] = co_await read(
555   self, mutable_buffer( 555   self, mutable_buffer(
556   buf.data(), buf.size())); 556   buf.data(), buf.size()));
557   if(ec) 557   if(ec)
558   { 558   {
559   result = ec; 559   result = ec;
560   co_return; 560   co_return;
561   } 561   }
562   match = (std::string_view( 562   match = (std::string_view(
563   buf.data(), n) == expected); 563   buf.data(), n) == expected);
HITCBC 564   161 }(*this, expected, result, match)); 564   161 }(*this, expected, result, match));
HITCBC 565   58 return {result, match}; 565   58 return {result, match};
566   } 566   }
567   567  
568   /** Return the stream's pending read data. 568   /** Return the stream's pending read data.
569   569  
570   Returns a view of the data waiting to be read 570   Returns a view of the data waiting to be read
571   from this stream. This is a direct peek at the 571   from this stream. This is a direct peek at the
572   internal buffer, bypassing the fuse. 572   internal buffer, bypassing the fuse.
573   573  
574   @return A view of the pending data. 574   @return A view of the pending data.
575   575  
576   @see provide, expect 576   @see provide, expect
577   */ 577   */
578   std::string_view 578   std::string_view
HITCBC 579   5 data() const noexcept 579   5 data() const noexcept
580   { 580   {
HITCBC 581   5 return state_->sides[index_].buf; 581   5 return state_->sides[index_].buf;
582   } 582   }
583   }; 583   };
584   584  
585   /** Create a connected pair of test streams. 585   /** Create a connected pair of test streams.
586   586  
587   Data written to one stream becomes readable on the other. 587   Data written to one stream becomes readable on the other.
588   If a coroutine calls @ref stream::read_some when no data 588   If a coroutine calls @ref stream::read_some when no data
589   is available, it suspends until the peer writes. Before 589   is available, it suspends until the peer writes. Before
590   every read or write, the @ref fuse is consulted to 590   every read or write, the @ref fuse is consulted to
591   possibly inject an error for testing fault scenarios. 591   possibly inject an error for testing fault scenarios.
592   When the fuse fires, the pair is automatically closed. 592   When the fuse fires, the pair is automatically closed.
593   593  
594   @param f The fuse used to inject errors during operations. 594   @param f The fuse used to inject errors during operations.
595   595  
596   @return A pair of connected streams. 596   @return A pair of connected streams.
597   597  
598   @see stream, fuse 598   @see stream, fuse
599   */ 599   */
600   inline std::pair<stream, stream> 600   inline std::pair<stream, stream>
HITCBC 601   308 make_stream_pair(fuse f = {}) 601   308 make_stream_pair(fuse f = {})
602   { 602   {
HITCBC 603   308 auto sp = std::make_shared<stream::state>(std::move(f)); 603   308 auto sp = std::make_shared<stream::state>(std::move(f));
HITCBC 604   616 return {stream(sp, 0), stream(sp, 1)}; 604   616 return {stream(sp, 0), stream(sp, 1)};
HITCBC 605   308 } 605   308 }
606   606  
607   } // test 607   } // test
608   } // capy 608   } // capy
609   } // boost 609   } // boost
610   610  
611   #endif 611   #endif