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