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_IO_ANY_READ_STREAM_HPP
12 : #define BOOST_CAPY_IO_ANY_READ_STREAM_HPP
13 :
14 : #include <boost/capy/detail/config.hpp>
15 : #include <boost/capy/detail/await_suspend_helper.hpp>
16 : #include <boost/capy/buffers.hpp>
17 : #include <boost/capy/detail/buffer_array.hpp>
18 : #include <boost/capy/concept/context_awaitable.hpp>
19 : #include <boost/capy/concept/read_stream.hpp>
20 : #include <boost/capy/ex/await_context.hpp>
21 : #include <boost/capy/results.hpp>
22 :
23 : #include <concepts>
24 : #include <coroutine>
25 : #include <cstddef>
26 : #include <exception>
27 : #include <new>
28 : #include <span>
29 : #include <stop_token>
30 : #include <system_error>
31 : #include <utility>
32 :
33 : namespace boost {
34 : namespace capy {
35 :
36 : /** Dispatches `read_some` through a type-erased vtable, using preallocated awaitable storage.
37 :
38 : This class provides type erasure for any type satisfying the
39 : @ref ReadStream concept, enabling runtime polymorphism for
40 : read operations. It uses cached awaitable storage to achieve
41 : zero steady-state allocation after construction.
42 :
43 : The wrapper supports two construction modes:
44 : - **Owning**: Pass by value to transfer ownership. The wrapper
45 : allocates storage and owns the stream.
46 : - **Reference**: Pass a pointer to wrap without ownership. The
47 : pointed-to stream must outlive this wrapper.
48 :
49 : @par Awaitable Preallocation
50 : The constructor preallocates storage for the type-erased awaitable.
51 : This reserves all virtual address space at server startup
52 : so memory usage can be measured up front, rather than
53 : allocating piecemeal as traffic arrives.
54 :
55 : @par Immediate Completion
56 : When the underlying stream's awaitable reports ready immediately
57 : (e.g. buffered data already available), the wrapper skips
58 : coroutine suspension entirely and returns the result inline.
59 :
60 : @par Thread Safety
61 : Not thread-safe. Concurrent operations on the same wrapper
62 : are undefined behavior.
63 :
64 : @par Example
65 : @par !example example
66 :
67 :
68 : @see any_write_stream, any_stream, ReadStream
69 : */
70 : class any_read_stream
71 : {
72 : struct vtable;
73 :
74 : template<ReadStream S>
75 : struct vtable_for_impl;
76 :
77 : // ordered for cache line coherence
78 : void* stream_ = nullptr;
79 : vtable const* vt_ = nullptr;
80 : void* cached_awaitable_ = nullptr;
81 : void* storage_ = nullptr;
82 : bool awaitable_active_ = false;
83 : // The flag of the awaitable whose operation is active, cleared
84 : // when the stream ends that operation itself.
85 : bool* owner_ = nullptr;
86 :
87 : void
88 HIT 220 : disown() noexcept
89 : {
90 220 : if(owner_)
91 8 : *std::exchange(owner_, nullptr) = false;
92 220 : }
93 :
94 : public:
95 : /** Destructor.
96 :
97 : Ends any operation in flight, then destroys the owned stream
98 : (if any) and releases the cached awaitable storage.
99 : */
100 : ~any_read_stream();
101 :
102 : /** Construct a default instance.
103 :
104 : Constructs an empty wrapper. @ref has_value and `operator bool`
105 : report the empty state; calling @ref read_some before the
106 : wrapper holds a stream is undefined behavior.
107 : */
108 10 : any_read_stream() = default;
109 :
110 : /** Non-copyable.
111 :
112 : The awaitable cache is per-instance and cannot be shared.
113 :
114 : @param other The wrapper that would be copied.
115 : */
116 : any_read_stream(any_read_stream const& other) = delete;
117 :
118 : /** Copy assignment is disabled.
119 :
120 : The awaitable cache is per-instance and cannot be shared.
121 :
122 : @param other The wrapper that would be assigned from.
123 :
124 : @return A reference to `*this`.
125 : */
126 : any_read_stream& operator=(any_read_stream const& other) = delete;
127 :
128 : /** Construct by moving.
129 :
130 : Transfers ownership of the wrapped stream (if owned) and
131 : cached awaitable storage from `other`. After the move, `other` is
132 : in a default-constructed state.
133 :
134 : @param other The wrapper to move from.
135 : */
136 2 : any_read_stream(any_read_stream&& other) noexcept
137 2 : : stream_(std::exchange(other.stream_, nullptr))
138 2 : , vt_(std::exchange(other.vt_, nullptr))
139 2 : , cached_awaitable_(std::exchange(other.cached_awaitable_, nullptr))
140 2 : , storage_(std::exchange(other.storage_, nullptr))
141 2 : , awaitable_active_(std::exchange(other.awaitable_active_, false))
142 : {
143 2 : other.disown();
144 2 : }
145 :
146 : /** Assign by moving.
147 :
148 : Destroys any owned stream and releases existing resources,
149 : then transfers ownership from `other`.
150 :
151 : @param other The wrapper to move from.
152 : @return Reference to this wrapper.
153 : */
154 : any_read_stream&
155 : operator=(any_read_stream&& other) noexcept;
156 :
157 : /** Construct by taking ownership of a ReadStream.
158 :
159 : Allocates storage and moves the stream into this wrapper.
160 : The wrapper owns the stream and destroys it.
161 :
162 : @param s The stream to take ownership of.
163 : */
164 : template<ReadStream S>
165 : requires (!std::same_as<std::decay_t<S>, any_read_stream>)
166 : any_read_stream(S s);
167 :
168 : /** Construct by wrapping a ReadStream without ownership.
169 :
170 : Wraps the given stream by pointer. The stream must remain
171 : valid for the lifetime of this wrapper.
172 :
173 : @param s Pointer to the stream to wrap.
174 : */
175 : template<ReadStream S>
176 : any_read_stream(S* s);
177 :
178 : /** Check if the wrapper contains a valid stream.
179 :
180 : @return `true` if wrapping a stream, `false` if default-constructed
181 : or moved-from.
182 : */
183 : bool
184 31 : has_value() const noexcept
185 : {
186 31 : return stream_ != nullptr;
187 : }
188 :
189 : /** Check if the wrapper contains a valid stream.
190 :
191 : @return `true` if wrapping a stream, `false` if default-constructed
192 : or moved-from.
193 : */
194 : explicit
195 3 : operator bool() const noexcept
196 : {
197 3 : return has_value();
198 : }
199 :
200 : /** Initiate an asynchronous read operation.
201 :
202 : Reads data into the provided buffer sequence. The operation
203 : completes when at least one byte is read, or an error
204 : occurs.
205 :
206 : @param buffers The buffer sequence to read into. Passed by
207 : value to ensure the sequence lives in the coroutine frame
208 : across suspension points.
209 :
210 : @return An awaitable that await-returns `(error_code,std::size_t)`.
211 :
212 : @par Immediate Completion
213 : The operation completes immediately without suspending
214 : the calling coroutine when the underlying stream's
215 : awaitable reports immediate readiness via `await_ready`.
216 :
217 : @note This is a partial operation and may not process the
218 : entire buffer sequence. Use the composed @ref read algorithm
219 : for guaranteed complete transfer.
220 :
221 : @par Preconditions
222 : The wrapper must contain a valid stream (`has_value() == true`).
223 :
224 : @par After an Error
225 : A subsequent call is permitted. The wrapper forwards directly
226 : to the underlying stream, imposing no stricter rule than
227 : @ref ReadStream.
228 : */
229 : template<MutableBufferSequence MB>
230 : auto
231 : read_some(MB buffers);
232 :
233 : protected:
234 : /** End the operation in flight, if any.
235 :
236 : Destroys the wrapped stream's active operation, which may refer
237 : to the stream, so an owner must call this before destroying the
238 : stream. The awaitable that started the operation no longer ends
239 : it.
240 : */
241 : void
242 : end_operation() noexcept;
243 :
244 : /** Rebind to a new stream after move.
245 :
246 : Updates the internal pointer to reference a new stream object.
247 : Used by owning wrappers after move assignment when the owned
248 : object has moved to a new location.
249 :
250 : @param new_stream The new stream to bind to. Must be the same
251 : type as the original stream.
252 :
253 : @note Terminates if called with a stream of different type
254 : than the original.
255 : */
256 : template<ReadStream S>
257 : void
258 : rebind(S& new_stream) noexcept
259 : {
260 : if(vt_ != &vtable_for_impl<S>::value)
261 : std::terminate();
262 : stream_ = &new_stream;
263 : }
264 : };
265 :
266 : struct any_read_stream::vtable
267 : {
268 : // ordered by call frequency for cache line coherence
269 : void (*construct_awaitable)(
270 : void* stream,
271 : void* storage,
272 : std::span<mutable_buffer const> buffers);
273 : bool (*await_ready)(void*);
274 : std::coroutine_handle<> (*await_suspend)(void*, std::coroutine_handle<>, await_context const*);
275 : results<std::size_t> (*await_resume)(void*);
276 : void (*destroy_awaitable)(void*) noexcept;
277 : std::size_t awaitable_size;
278 : std::size_t awaitable_align;
279 : void (*destroy)(void*) noexcept;
280 : };
281 :
282 : template<ReadStream S>
283 : struct any_read_stream::vtable_for_impl
284 : {
285 : using Awaitable = decltype(std::declval<S&>().read_some(
286 : std::span<mutable_buffer const>{}));
287 :
288 : static void
289 6 : do_destroy_impl(void* stream) noexcept
290 : {
291 6 : static_cast<S*>(stream)->~S();
292 6 : }
293 :
294 : static void
295 108 : construct_awaitable_impl(
296 : void* stream,
297 : void* storage,
298 : std::span<mutable_buffer const> buffers)
299 : {
300 108 : auto& s = *static_cast<S*>(stream);
301 108 : ::new(storage) Awaitable(s.read_some(buffers));
302 108 : }
303 :
304 : static constexpr vtable value = {
305 : &construct_awaitable_impl,
306 108 : +[](void* p) {
307 108 : return static_cast<Awaitable*>(p)->await_ready();
308 : },
309 84 : +[](void* p, std::coroutine_handle<> h, await_context const* env) {
310 84 : return detail::call_await_suspend(
311 84 : static_cast<Awaitable*>(p), h, env);
312 : },
313 97 : +[](void* p) {
314 97 : return static_cast<Awaitable*>(p)->await_resume();
315 : },
316 125 : +[](void* p) noexcept {
317 17 : static_cast<Awaitable*>(p)->~Awaitable();
318 : },
319 : sizeof(Awaitable),
320 : alignof(Awaitable),
321 : &do_destroy_impl
322 : };
323 : };
324 :
325 : inline void
326 198 : any_read_stream::end_operation() noexcept
327 : {
328 198 : disown();
329 198 : if(awaitable_active_)
330 : {
331 8 : vt_->destroy_awaitable(cached_awaitable_);
332 8 : awaitable_active_ = false;
333 : }
334 198 : }
335 :
336 : inline
337 135 : any_read_stream::~any_read_stream()
338 : {
339 135 : end_operation();
340 135 : if(cached_awaitable_)
341 110 : ::operator delete(cached_awaitable_);
342 135 : if(storage_)
343 : {
344 4 : vt_->destroy(stream_);
345 4 : ::operator delete(storage_);
346 : }
347 135 : }
348 :
349 : inline any_read_stream&
350 20 : any_read_stream::operator=(any_read_stream&& other) noexcept
351 : {
352 20 : if(this != &other)
353 : {
354 20 : end_operation();
355 20 : other.disown();
356 20 : if(cached_awaitable_)
357 8 : ::operator delete(cached_awaitable_);
358 20 : if(storage_)
359 : {
360 2 : vt_->destroy(stream_);
361 2 : ::operator delete(storage_);
362 : }
363 20 : stream_ = std::exchange(other.stream_, nullptr);
364 20 : vt_ = std::exchange(other.vt_, nullptr);
365 20 : cached_awaitable_ = std::exchange(other.cached_awaitable_, nullptr);
366 20 : storage_ = std::exchange(other.storage_, nullptr);
367 20 : awaitable_active_ = std::exchange(other.awaitable_active_, false);
368 : }
369 20 : return *this;
370 : }
371 :
372 : template<ReadStream S>
373 : requires (!std::same_as<std::decay_t<S>, any_read_stream>)
374 7 : any_read_stream::any_read_stream(S s)
375 7 : : vt_(&vtable_for_impl<S>::value)
376 : {
377 : struct guard {
378 : any_read_stream* self;
379 : bool committed = false;
380 7 : ~guard() {
381 7 : if(!committed && self->storage_) {
382 1 : if(self->stream_)
383 : self->vt_->destroy(self->stream_); // LCOV_EXCL_LINE OOM rollback: only when the cached-awaitable allocation throws
384 1 : ::operator delete(self->storage_);
385 1 : self->storage_ = nullptr;
386 1 : self->stream_ = nullptr;
387 : }
388 7 : }
389 7 : } g{this};
390 :
391 7 : storage_ = ::operator new(sizeof(S));
392 7 : stream_ = ::new(storage_) S(std::move(s));
393 :
394 : // Preallocate the awaitable storage
395 6 : cached_awaitable_ = ::operator new(vt_->awaitable_size);
396 :
397 6 : g.committed = true;
398 7 : }
399 :
400 : template<ReadStream S>
401 112 : any_read_stream::any_read_stream(S* s)
402 112 : : stream_(s)
403 112 : , vt_(&vtable_for_impl<S>::value)
404 : {
405 : // Preallocate the awaitable storage
406 112 : cached_awaitable_ = ::operator new(vt_->awaitable_size);
407 112 : }
408 :
409 : template<MutableBufferSequence MB>
410 : auto
411 109 : any_read_stream::read_some(MB buffers)
412 : {
413 : // VFALCO in theory, we could use if constexpr to detect a
414 : // span and then pass that through to read_some without the array
415 : // LCOV_EXCL_START read_some awaitable: exercised by tests, but the
416 : // coverage tooling reports its templated body uncovered per-instantiation
417 : struct awaitable
418 : {
419 : any_read_stream* self_;
420 : detail::mutable_buffer_array<detail::max_iovec_> ba_;
421 : // Set while this awaitable, not another on the same stream,
422 : // holds the stream's inner awaitable.
423 : bool owns_ = false;
424 :
425 : // A chain unwound while suspended here never reaches
426 : // await_resume, so the inner awaitable is destroyed here.
427 : ~awaitable()
428 : {
429 : if(owns_)
430 : {
431 : self_->vt_->destroy_awaitable(self_->cached_awaitable_);
432 : self_->awaitable_active_ = false;
433 : self_->owner_ = nullptr;
434 : }
435 : }
436 :
437 : awaitable(
438 : any_read_stream* self,
439 : detail::mutable_buffer_array<detail::max_iovec_> ba) noexcept
440 : : self_(self)
441 : , ba_(ba)
442 : {
443 : }
444 :
445 : awaitable(awaitable&& other) noexcept
446 : : self_(std::exchange(other.self_, nullptr))
447 : , ba_(other.ba_)
448 : , owns_(std::exchange(other.owns_, false))
449 : {
450 : if(owns_)
451 : self_->owner_ = &owns_;
452 : }
453 :
454 : bool
455 : await_ready()
456 : {
457 : self_->vt_->construct_awaitable(
458 : self_->stream_,
459 : self_->cached_awaitable_,
460 : ba_.to_span());
461 : self_->awaitable_active_ = true;
462 : owns_ = true;
463 : self_->owner_ = &owns_;
464 :
465 : return self_->vt_->await_ready(
466 : self_->cached_awaitable_);
467 : }
468 :
469 : std::coroutine_handle<>
470 : await_suspend(std::coroutine_handle<> h, await_context const* env)
471 : {
472 : return self_->vt_->await_suspend(
473 : self_->cached_awaitable_, h, env);
474 : }
475 :
476 : [[nodiscard]] results<std::size_t>
477 : await_resume()
478 : {
479 : struct guard {
480 : awaitable* a;
481 : ~guard() {
482 : a->self_->vt_->destroy_awaitable(
483 : a->self_->cached_awaitable_);
484 : a->self_->awaitable_active_ = false;
485 : a->self_->owner_ = nullptr;
486 : a->owns_ = false;
487 : }
488 : } g{this};
489 : return self_->vt_->await_resume(
490 : self_->cached_awaitable_);
491 : }
492 : };
493 : // LCOV_EXCL_STOP
494 : return awaitable{this,
495 109 : detail::mutable_buffer_array<detail::max_iovec_>(buffers)};
496 : }
497 :
498 : } // namespace capy
499 : } // namespace boost
500 :
501 : #endif
|
