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