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