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