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