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