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