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