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/io_awaitable.hpp>
19 : #include <boost/capy/concept/read_stream.hpp>
20 : #include <boost/capy/ex/io_env.hpp>
21 : #include <boost/capy/io_result.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 : @code
66 : // Owning - takes ownership of the stream
67 : any_read_stream owning_stream(socket{ioc});
68 :
69 : // Reference - wraps without ownership
70 : socket sock(ioc);
71 : any_read_stream ref_stream(&sock);
72 :
73 : char data[1024];
74 : mutable_buffer buf(data, sizeof(data));
75 : auto [ec, n] = co_await owning_stream.read_some(buf);
76 : @endcode
77 :
78 : @see any_write_stream, any_stream, ReadStream
79 : */
80 : class any_read_stream
81 : {
82 : struct vtable;
83 :
84 : template<ReadStream S>
85 : struct vtable_for_impl;
86 :
87 : // ordered for cache line coherence
88 : void* stream_ = nullptr;
89 : vtable const* vt_ = nullptr;
90 : void* cached_awaitable_ = nullptr;
91 : void* storage_ = nullptr;
92 : bool awaitable_active_ = false;
93 :
94 : public:
95 : /** Destructor.
96 :
97 : Destroys the owned stream (if any) and releases the cached
98 : 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 HIT 4 : 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 4 : any_read_stream(any_read_stream&& other) noexcept
137 4 : : stream_(std::exchange(other.stream_, nullptr))
138 4 : , vt_(std::exchange(other.vt_, nullptr))
139 4 : , cached_awaitable_(std::exchange(other.cached_awaitable_, nullptr))
140 4 : , storage_(std::exchange(other.storage_, nullptr))
141 4 : , awaitable_active_(std::exchange(other.awaitable_active_, false))
142 : {
143 4 : }
144 :
145 : /** Assign by moving.
146 :
147 : Destroys any owned stream and releases existing resources,
148 : then transfers ownership from `other`.
149 :
150 : @param other The wrapper to move from.
151 : @return Reference to this wrapper.
152 : */
153 : any_read_stream&
154 : operator=(any_read_stream&& other) noexcept;
155 :
156 : /** Construct by taking ownership of a ReadStream.
157 :
158 : Allocates storage and moves the stream into this wrapper.
159 : The wrapper owns the stream and destroys it.
160 :
161 : @param s The stream to take ownership of.
162 : */
163 : template<ReadStream S>
164 : requires (!std::same_as<std::decay_t<S>, any_read_stream>)
165 : any_read_stream(S s);
166 :
167 : /** Construct by wrapping a ReadStream without ownership.
168 :
169 : Wraps the given stream by pointer. The stream must remain
170 : valid for the lifetime of this wrapper.
171 :
172 : @param s Pointer to the stream to wrap.
173 : */
174 : template<ReadStream S>
175 : any_read_stream(S* s);
176 :
177 : /** Check if the wrapper contains a valid stream.
178 :
179 : @return `true` if wrapping a stream, `false` if default-constructed
180 : or moved-from.
181 : */
182 : bool
183 31 : has_value() const noexcept
184 : {
185 31 : return stream_ != nullptr;
186 : }
187 :
188 : /** Check if the wrapper contains a valid stream.
189 :
190 : @return `true` if wrapping a stream, `false` if default-constructed
191 : or moved-from.
192 : */
193 : explicit
194 3 : operator bool() const noexcept
195 : {
196 3 : return has_value();
197 : }
198 :
199 : /** Initiate an asynchronous read operation.
200 :
201 : Reads data into the provided buffer sequence. The operation
202 : completes when at least one byte is read, or an error
203 : occurs.
204 :
205 : @param buffers The buffer sequence to read into. Passed by
206 : value to ensure the sequence lives in the coroutine frame
207 : across suspension points.
208 :
209 : @return An awaitable that await-returns `(error_code,std::size_t)`.
210 :
211 : @par Immediate Completion
212 : The operation completes immediately without suspending
213 : the calling coroutine when the underlying stream's
214 : awaitable reports immediate readiness via `await_ready`.
215 :
216 : @note This is a partial operation and may not process the
217 : entire buffer sequence. Use the composed @ref read algorithm
218 : for guaranteed complete transfer.
219 :
220 : @par Preconditions
221 : The wrapper must contain a valid stream (`has_value() == true`).
222 :
223 : @par After an Error
224 : A subsequent call is permitted. The wrapper forwards directly
225 : to the underlying stream, imposing no stricter rule than
226 : @ref ReadStream.
227 : */
228 : template<MutableBufferSequence MB>
229 : auto
230 : read_some(MB buffers);
231 :
232 : protected:
233 : /** Rebind to a new stream after move.
234 :
235 : Updates the internal pointer to reference a new stream object.
236 : Used by owning wrappers after move assignment when the owned
237 : object has moved to a new location.
238 :
239 : @param new_stream The new stream to bind to. Must be the same
240 : type as the original stream.
241 :
242 : @note Terminates if called with a stream of different type
243 : than the original.
244 : */
245 : template<ReadStream S>
246 : void
247 : rebind(S& new_stream) noexcept
248 : {
249 : if(vt_ != &vtable_for_impl<S>::value)
250 : std::terminate();
251 : stream_ = &new_stream;
252 : }
253 : };
254 :
255 : struct any_read_stream::vtable
256 : {
257 : // ordered by call frequency for cache line coherence
258 : void (*construct_awaitable)(
259 : void* stream,
260 : void* storage,
261 : std::span<mutable_buffer const> buffers);
262 : bool (*await_ready)(void*);
263 : std::coroutine_handle<> (*await_suspend)(void*, std::coroutine_handle<>, io_env const*);
264 : io_result<std::size_t> (*await_resume)(void*);
265 : void (*destroy_awaitable)(void*) noexcept;
266 : std::size_t awaitable_size;
267 : std::size_t awaitable_align;
268 : void (*destroy)(void*) noexcept;
269 : };
270 :
271 : template<ReadStream S>
272 : struct any_read_stream::vtable_for_impl
273 : {
274 : using Awaitable = decltype(std::declval<S&>().read_some(
275 : std::span<mutable_buffer const>{}));
276 :
277 : static void
278 4 : do_destroy_impl(void* stream) noexcept
279 : {
280 4 : static_cast<S*>(stream)->~S();
281 4 : }
282 :
283 : static void
284 103 : construct_awaitable_impl(
285 : void* stream,
286 : void* storage,
287 : std::span<mutable_buffer const> buffers)
288 : {
289 103 : auto& s = *static_cast<S*>(stream);
290 103 : ::new(storage) Awaitable(s.read_some(buffers));
291 103 : }
292 :
293 : static constexpr vtable value = {
294 : &construct_awaitable_impl,
295 103 : +[](void* p) {
296 103 : return static_cast<Awaitable*>(p)->await_ready();
297 : },
298 77 : +[](void* p, std::coroutine_handle<> h, io_env const* env) {
299 77 : return detail::call_await_suspend(
300 77 : static_cast<Awaitable*>(p), h, env);
301 : },
302 101 : +[](void* p) {
303 101 : return static_cast<Awaitable*>(p)->await_resume();
304 : },
305 115 : +[](void* p) noexcept {
306 26 : static_cast<Awaitable*>(p)->~Awaitable();
307 : },
308 : sizeof(Awaitable),
309 : alignof(Awaitable),
310 : &do_destroy_impl
311 : };
312 : };
313 :
314 : inline
315 123 : any_read_stream::~any_read_stream()
316 : {
317 123 : if(storage_)
318 : {
319 3 : vt_->destroy(stream_);
320 3 : ::operator delete(storage_);
321 : }
322 123 : if(cached_awaitable_)
323 : {
324 106 : if(awaitable_active_)
325 1 : vt_->destroy_awaitable(cached_awaitable_);
326 106 : ::operator delete(cached_awaitable_);
327 : }
328 123 : }
329 :
330 : inline any_read_stream&
331 10 : any_read_stream::operator=(any_read_stream&& other) noexcept
332 : {
333 10 : if(this != &other)
334 : {
335 10 : if(storage_)
336 : {
337 1 : vt_->destroy(stream_);
338 1 : ::operator delete(storage_);
339 : }
340 10 : if(cached_awaitable_)
341 : {
342 4 : if(awaitable_active_)
343 1 : vt_->destroy_awaitable(cached_awaitable_);
344 4 : ::operator delete(cached_awaitable_);
345 : }
346 10 : stream_ = std::exchange(other.stream_, nullptr);
347 10 : vt_ = std::exchange(other.vt_, nullptr);
348 10 : cached_awaitable_ = std::exchange(other.cached_awaitable_, nullptr);
349 10 : storage_ = std::exchange(other.storage_, nullptr);
350 10 : awaitable_active_ = std::exchange(other.awaitable_active_, false);
351 : }
352 10 : return *this;
353 : }
354 :
355 : template<ReadStream S>
356 : requires (!std::same_as<std::decay_t<S>, any_read_stream>)
357 5 : any_read_stream::any_read_stream(S s)
358 5 : : vt_(&vtable_for_impl<S>::value)
359 : {
360 : struct guard {
361 : any_read_stream* self;
362 : bool committed = false;
363 5 : ~guard() {
364 5 : if(!committed && self->storage_) {
365 1 : if(self->stream_)
366 : self->vt_->destroy(self->stream_); // LCOV_EXCL_LINE OOM rollback: only when the cached-awaitable allocation throws
367 1 : ::operator delete(self->storage_);
368 1 : self->storage_ = nullptr;
369 1 : self->stream_ = nullptr;
370 : }
371 5 : }
372 5 : } g{this};
373 :
374 5 : storage_ = ::operator new(sizeof(S));
375 5 : stream_ = ::new(storage_) S(std::move(s));
376 :
377 : // Preallocate the awaitable storage
378 4 : cached_awaitable_ = ::operator new(vt_->awaitable_size);
379 :
380 4 : g.committed = true;
381 5 : }
382 :
383 : template<ReadStream S>
384 106 : any_read_stream::any_read_stream(S* s)
385 106 : : stream_(s)
386 106 : , vt_(&vtable_for_impl<S>::value)
387 : {
388 : // Preallocate the awaitable storage
389 106 : cached_awaitable_ = ::operator new(vt_->awaitable_size);
390 106 : }
391 :
392 : template<MutableBufferSequence MB>
393 : auto
394 103 : any_read_stream::read_some(MB buffers)
395 : {
396 : // VFALCO in theory, we could use if constexpr to detect a
397 : // span and then pass that through to read_some without the array
398 : // LCOV_EXCL_START read_some awaitable: exercised by tests, but the
399 : // coverage tooling reports its templated body uncovered per-instantiation
400 : struct awaitable
401 : {
402 : any_read_stream* self_;
403 : detail::mutable_buffer_array<detail::max_iovec_> ba_;
404 :
405 : bool
406 : await_ready()
407 : {
408 : self_->vt_->construct_awaitable(
409 : self_->stream_,
410 : self_->cached_awaitable_,
411 : ba_.to_span());
412 : self_->awaitable_active_ = true;
413 :
414 : return self_->vt_->await_ready(
415 : self_->cached_awaitable_);
416 : }
417 :
418 : std::coroutine_handle<>
419 : await_suspend(std::coroutine_handle<> h, io_env const* env)
420 : {
421 : return self_->vt_->await_suspend(
422 : self_->cached_awaitable_, h, env);
423 : }
424 :
425 : io_result<std::size_t>
426 : await_resume()
427 : {
428 : struct guard {
429 : any_read_stream* self;
430 : ~guard() {
431 : self->vt_->destroy_awaitable(self->cached_awaitable_);
432 : self->awaitable_active_ = false;
433 : }
434 : } g{self_};
435 : return self_->vt_->await_resume(
436 : self_->cached_awaitable_);
437 : }
438 : };
439 : // LCOV_EXCL_STOP
440 : return awaitable{this,
441 103 : detail::mutable_buffer_array<detail::max_iovec_>(buffers)};
442 103 : }
443 :
444 : } // namespace capy
445 : } // namespace boost
446 :
447 : #endif
|
