. /include/boost/capy /io /any_read_source.hpp
0.00% Lines (0/0) 0.00% Functions (0/0)
[ ] prev/next file n p prev/next uncovered line
TLA Baseline Branch
Line Hits Code Line Hits Code
1 - //  
2 - // Copyright (c) 2025 Vinnie Falco (vinnie.falco@gmail.com)  
3 - //  
4 - // Distributed under the Boost Software License, Version 1.0. (See accompanying  
5 - // file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)  
6 - //  
7 - // Official repository: https://github.com/cppalliance/capy  
8 - //  
9 -  
10 - #ifndef BOOST_CAPY_IO_ANY_READ_SOURCE_HPP  
11 - #define BOOST_CAPY_IO_ANY_READ_SOURCE_HPP  
12 -  
13 - #include <boost/capy/detail/config.hpp>  
14 - #include <boost/capy/detail/await_suspend_helper.hpp>  
15 - #include <boost/capy/buffers.hpp>  
16 - #include <boost/capy/detail/buffer_array.hpp>  
17 - #include <boost/capy/buffers/buffer_param.hpp>  
18 - #include <boost/capy/concept/io_awaitable.hpp>  
19 - #include <boost/capy/concept/read_source.hpp>  
20 - #include <boost/capy/ex/io_env.hpp>  
21 - #include <boost/capy/io_result.hpp>  
22 - #include <boost/capy/io_task.hpp>  
23 -  
24 - #include <concepts>  
25 - #include <coroutine>  
26 - #include <cstddef>  
27 - #include <exception>  
28 - #include <new>  
29 - #include <span>  
30 - #include <stop_token>  
31 - #include <system_error>  
32 - #include <utility>  
33 -  
34 - namespace boost {  
35 - namespace capy {  
36 -  
37 - /** Type-erased wrapper for any ReadSource.  
38 -  
39 - This class provides type erasure for any type satisfying the  
40 - @ref ReadSource concept, enabling runtime polymorphism for  
41 - source read operations. It uses cached awaitable storage to achieve  
42 - zero steady-state allocation after construction.  
43 -  
44 - The wrapper supports two construction modes:  
45 - - **Owning**: Pass by value to transfer ownership. The wrapper  
46 - allocates storage and owns the source.  
47 - - **Reference**: Pass a pointer to wrap without ownership. The  
48 - pointed-to source must outlive this wrapper.  
49 -  
50 - @par Awaitable Preallocation  
51 - The constructor preallocates storage for the type-erased awaitable.  
52 - This reserves all virtual address space at server startup  
53 - so memory usage can be measured up front, rather than  
54 - allocating piecemeal as traffic arrives.  
55 -  
56 - @par Immediate Completion  
57 - Operations complete immediately without suspending when the  
58 - buffer sequence is empty, or when the underlying source's  
59 - awaitable reports readiness via `await_ready`.  
60 -  
61 - @par Thread Safety  
62 - Not thread-safe. Concurrent operations on the same wrapper  
63 - are undefined behavior.  
64 -  
65 - @par Example  
66 - @code  
67 - // Owning - takes ownership of the source  
68 - any_read_source rs(some_source{args...});  
69 -  
70 - // Reference - wraps without ownership  
71 - some_source source;  
72 - any_read_source rs(&source);  
73 -  
74 - mutable_buffer buf(data, size);  
75 - auto [ec, n] = co_await rs.read(std::span(&buf, 1));  
76 - @endcode  
77 -  
78 - @see any_read_stream, ReadSource  
79 - */  
80 - class any_read_source  
81 - {  
82 - struct vtable;  
83 - struct awaitable_ops;  
84 -  
85 - template<ReadSource S>  
86 - struct vtable_for_impl;  
87 -  
88 - void* source_ = nullptr;  
89 - vtable const* vt_ = nullptr;  
90 - void* cached_awaitable_ = nullptr;  
91 - void* storage_ = nullptr;  
92 - awaitable_ops const* active_ops_ = nullptr;  
93 -  
94 - public:  
95 - /** Destructor.  
96 -  
97 - Destroys the owned source (if any) and releases the cached  
98 - awaitable storage.  
99 - */  
100 - ~any_read_source();  
101 -  
102 - /** Construct a default instance.  
103 -  
104 - Constructs an empty wrapper. Operations on a default-constructed  
105 - wrapper result in undefined behavior.  
106 - */  
107 - any_read_source() = default;  
108 -  
109 - /** Non-copyable.  
110 -  
111 - The awaitable cache is per-instance and cannot be shared.  
112 - */  
113 - any_read_source(any_read_source const&) = delete;  
114 - any_read_source& operator=(any_read_source const&) = delete;  
115 -  
116 - /** Construct by moving.  
117 -  
118 - Transfers ownership of the wrapped source (if owned) and  
119 - cached awaitable storage from `other`. After the move, `other` is  
120 - in a default-constructed state.  
121 -  
122 - @param other The wrapper to move from.  
123 - */  
DCB 124 - 1 any_read_source(any_read_source&& other) noexcept  
DCB 125 - 1 : source_(std::exchange(other.source_, nullptr))  
DCB 126 - 1 , vt_(std::exchange(other.vt_, nullptr))  
DCB 127 - 1 , cached_awaitable_(std::exchange(other.cached_awaitable_, nullptr))  
DCB 128 - 1 , storage_(std::exchange(other.storage_, nullptr))  
DCB 129 - 1 , active_ops_(std::exchange(other.active_ops_, nullptr))  
130 - {  
DCB 131 - 1 }  
132 -  
133 - /** Assign by moving.  
134 -  
135 - Destroys any owned source and releases existing resources,  
136 - then transfers ownership from `other`.  
137 -  
138 - @param other The wrapper to move from.  
139 - @return Reference to this wrapper.  
140 - */  
141 - any_read_source&  
142 - operator=(any_read_source&& other) noexcept;  
143 -  
144 - /** Construct by taking ownership of a ReadSource.  
145 -  
146 - Allocates storage and moves the source into this wrapper.  
147 - The wrapper owns the source and will destroy it.  
148 -  
149 - @param s The source to take ownership of.  
150 - */  
151 - template<ReadSource S>  
152 - requires (!std::same_as<std::decay_t<S>, any_read_source>)  
153 - any_read_source(S s);  
154 -  
155 - /** Construct by wrapping a ReadSource without ownership.  
156 -  
157 - Wraps the given source by pointer. The source must remain  
158 - valid for the lifetime of this wrapper.  
159 -  
160 - @param s Pointer to the source to wrap.  
161 - */  
162 - template<ReadSource S>  
163 - any_read_source(S* s);  
164 -  
165 - /** Check if the wrapper contains a valid source.  
166 -  
167 - @return `true` if wrapping a source, `false` if default-constructed  
168 - or moved-from.  
169 - */  
170 - bool  
DCB 171 - 30 has_value() const noexcept  
172 - {  
DCB 173 - 30 return source_ != nullptr;  
174 - }  
175 -  
176 - /** Check if the wrapper contains a valid source.  
177 -  
178 - @return `true` if wrapping a source, `false` if default-constructed  
179 - or moved-from.  
180 - */  
181 - explicit  
DCB 182 - 8 operator bool() const noexcept  
183 - {  
DCB 184 - 8 return has_value();  
185 - }  
186 -  
187 - /** Initiate a partial read operation.  
188 -  
189 - Attempt to read up to `buffer_size( buffers )` bytes into  
190 - the provided buffer sequence. May fill less than the  
191 - full sequence.  
192 -  
193 - @param buffers The buffer sequence to read into.  
194 -  
195 - @return An awaitable that await-returns `(error_code,std::size_t)`.  
196 -  
197 - @par Immediate Completion  
198 - The operation completes immediately without suspending  
199 - the calling coroutine when:  
200 - @li The buffer sequence is empty, returning `{error_code{}, 0}`.  
201 - @li The underlying source's awaitable reports immediate  
202 - readiness via `await_ready`.  
203 -  
204 - @note This is a partial operation and may not process the  
205 - entire buffer sequence. Use @ref read for guaranteed  
206 - complete transfer.  
207 -  
208 - @par Preconditions  
209 - The wrapper must contain a valid source (`has_value() == true`).  
210 - The caller must not call this function again after a prior  
211 - call returned an error (including EOF).  
212 - */  
213 - template<MutableBufferSequence MB>  
214 - auto  
215 - read_some(MB buffers);  
216 -  
217 - /** Initiate a complete read operation.  
218 -  
219 - Reads data into the provided buffer sequence by forwarding  
220 - to the underlying source's `read` operation. Large buffer  
221 - sequences are processed in windows, with each window  
222 - forwarded as a separate `read` call to the underlying source.  
223 - The operation completes when the entire buffer sequence is  
224 - filled, end-of-file is reached, or an error occurs.  
225 -  
226 - @param buffers The buffer sequence to read into.  
227 -  
228 - @return An awaitable that await-returns `(error_code,std::size_t)`.  
229 -  
230 - @par Immediate Completion  
231 - The operation completes immediately without suspending  
232 - the calling coroutine when:  
233 - @li The buffer sequence is empty, returning `{error_code{}, 0}`.  
234 - @li The underlying source's `read` awaitable reports  
235 - immediate readiness via `await_ready`.  
236 -  
237 - @par Postconditions  
238 - Exactly one of the following is true on return:  
239 - @li **Success**: `!ec` and `n == buffer_size(buffers)`.  
240 - The entire buffer was filled.  
241 - @li **End-of-stream or Error**: `ec` and `n` indicates  
242 - the number of bytes transferred before the failure.  
243 -  
244 - @par Preconditions  
245 - The wrapper must contain a valid source (`has_value() == true`).  
246 - The caller must not call this function again after a prior  
247 - call returned an error (including EOF).  
248 - */  
249 - template<MutableBufferSequence MB>  
250 - io_task<std::size_t>  
251 - read(MB buffers);  
252 -  
253 - protected:  
254 - /** Rebind to a new source after move.  
255 -  
256 - Updates the internal pointer to reference a new source object.  
257 - Used by owning wrappers after move assignment when the owned  
258 - object has moved to a new location.  
259 -  
260 - @param new_source The new source to bind to. Must be the same  
261 - type as the original source.  
262 -  
263 - @note Terminates if called with a source of different type  
264 - than the original.  
265 - */  
266 - template<ReadSource S>  
267 - void  
268 - rebind(S& new_source) noexcept  
269 - {  
270 - if(vt_ != &vtable_for_impl<S>::value)  
271 - std::terminate();  
272 - source_ = &new_source;  
273 - }  
274 -  
275 - private:  
276 - auto  
277 - read_(std::span<mutable_buffer const> buffers);  
278 - };  
279 -  
280 - // ordered by call sequence for cache line coherence  
281 - struct any_read_source::awaitable_ops  
282 - {  
283 - bool (*await_ready)(void*);  
284 - std::coroutine_handle<> (*await_suspend)(void*, std::coroutine_handle<>, io_env const*);  
285 - io_result<std::size_t> (*await_resume)(void*);  
286 - void (*destroy)(void*) noexcept;  
287 - };  
288 -  
289 - // ordered by call frequency for cache line coherence  
290 - struct any_read_source::vtable  
291 - {  
292 - awaitable_ops const* (*construct_read_some_awaitable)(  
293 - void* source,  
294 - void* storage,  
295 - std::span<mutable_buffer const> buffers);  
296 - awaitable_ops const* (*construct_read_awaitable)(  
297 - void* source,  
298 - void* storage,  
299 - std::span<mutable_buffer const> buffers);  
300 - std::size_t awaitable_size;  
301 - std::size_t awaitable_align;  
302 - void (*destroy)(void*) noexcept;  
303 - };  
304 -  
305 - template<ReadSource S>  
306 - struct any_read_source::vtable_for_impl  
307 - {  
308 - using ReadSomeAwaitable = decltype(std::declval<S&>().read_some(  
309 - std::span<mutable_buffer const>{}));  
310 - using ReadAwaitable = decltype(std::declval<S&>().read(  
311 - std::span<mutable_buffer const>{}));  
312 -  
313 - static void  
DCB 314 - 8 do_destroy_impl(void* source) noexcept  
315 - {  
DCB 316 - 8 static_cast<S*>(source)->~S();  
DCB 317 - 8 }  
318 -  
319 - static awaitable_ops const*  
DCB 320 - 52 construct_read_some_awaitable_impl(  
321 - void* source,  
322 - void* storage,  
323 - std::span<mutable_buffer const> buffers)  
324 - {  
DCB 325 - 52 auto& s = *static_cast<S*>(source);  
DCB 326 - 52 ::new(storage) ReadSomeAwaitable(s.read_some(buffers));  
327 -  
328 - static constexpr awaitable_ops ops = {  
DCB 329 - 52 +[](void* p) {  
DCB 330 - 52 return static_cast<ReadSomeAwaitable*>(p)->await_ready();  
331 - },  
DCB 332 - 52 +[](void* p, std::coroutine_handle<> h, io_env const* env) {  
DCB 333 - 52 return detail::call_await_suspend(  
DCB 334 - 52 static_cast<ReadSomeAwaitable*>(p), h, env);  
335 - },  
DCB 336 - 50 +[](void* p) {  
DCB 337 - 50 return static_cast<ReadSomeAwaitable*>(p)->await_resume();  
338 - },  
DCB 339 - 54 +[](void* p) noexcept {  
DCB 340 - 2 static_cast<ReadSomeAwaitable*>(p)->~ReadSomeAwaitable();  
341 - }  
342 - };  
DCB 343 - 52 return &ops;  
344 - }  
345 -  
346 - static awaitable_ops const*  
DCB 347 - 117 construct_read_awaitable_impl(  
348 - void* source,  
349 - void* storage,  
350 - std::span<mutable_buffer const> buffers)  
351 - {  
DCB 352 - 117 auto& s = *static_cast<S*>(source);  
DCB 353 - 117 ::new(storage) ReadAwaitable(s.read(buffers));  
354 -  
355 - static constexpr awaitable_ops ops = {  
DCB 356 - 117 +[](void* p) {  
DCB 357 - 117 return static_cast<ReadAwaitable*>(p)->await_ready();  
358 - },  
DCB 359 - 117 +[](void* p, std::coroutine_handle<> h, io_env const* env) {  
DCB 360 - 117 return detail::call_await_suspend(  
DCB 361 - 117 static_cast<ReadAwaitable*>(p), h, env);  
362 - },  
DCB 363 - 117 +[](void* p) {  
DCB 364 - 117 return static_cast<ReadAwaitable*>(p)->await_resume();  
365 - },  
DCB 366 - 117 +[](void* p) noexcept {  
367 - static_cast<ReadAwaitable*>(p)->~ReadAwaitable(); // LCOV_EXCL_LINE runs via destroy tests; gcov miscounts fn-ptr thunk  
368 - }  
369 - };  
DCB 370 - 117 return &ops;  
371 - }  
372 -  
373 - static constexpr std::size_t max_awaitable_size =  
374 - sizeof(ReadSomeAwaitable) > sizeof(ReadAwaitable)  
375 - ? sizeof(ReadSomeAwaitable)  
376 - : sizeof(ReadAwaitable);  
377 - static constexpr std::size_t max_awaitable_align =  
378 - alignof(ReadSomeAwaitable) > alignof(ReadAwaitable)  
379 - ? alignof(ReadSomeAwaitable)  
380 - : alignof(ReadAwaitable);  
381 -  
382 - static constexpr vtable value = {  
383 - &construct_read_some_awaitable_impl,  
384 - &construct_read_awaitable_impl,  
385 - max_awaitable_size,  
386 - max_awaitable_align,  
387 - &do_destroy_impl  
388 - };  
389 - };  
390 -  
391 - inline  
DCB 392 - 148 any_read_source::~any_read_source()  
393 - {  
DCB 394 - 148 if(storage_)  
395 - {  
DCB 396 - 7 vt_->destroy(source_);  
DCB 397 - 7 ::operator delete(storage_);  
398 - }  
DCB 399 - 148 if(cached_awaitable_)  
400 - {  
DCB 401 - 141 if(active_ops_)  
DCB 402 - 1 active_ops_->destroy(cached_awaitable_);  
DCB 403 - 141 ::operator delete(cached_awaitable_);  
404 - }  
DCB 405 - 148 }  
406 -  
407 - inline any_read_source&  
DCB 408 - 5 any_read_source::operator=(any_read_source&& other) noexcept  
409 - {  
DCB 410 - 5 if(this != &other)  
411 - {  
DCB 412 - 4 if(storage_)  
413 - {  
DCB 414 - 1 vt_->destroy(source_);  
DCB 415 - 1 ::operator delete(storage_);  
416 - }  
DCB 417 - 4 if(cached_awaitable_)  
418 - {  
DCB 419 - 3 if(active_ops_)  
DCB 420 - 1 active_ops_->destroy(cached_awaitable_);  
DCB 421 - 3 ::operator delete(cached_awaitable_);  
422 - }  
DCB 423 - 4 source_ = std::exchange(other.source_, nullptr);  
DCB 424 - 4 vt_ = std::exchange(other.vt_, nullptr);  
DCB 425 - 4 cached_awaitable_ = std::exchange(other.cached_awaitable_, nullptr);  
DCB 426 - 4 storage_ = std::exchange(other.storage_, nullptr);  
DCB 427 - 4 active_ops_ = std::exchange(other.active_ops_, nullptr);  
428 - }  
DCB 429 - 5 return *this;  
430 - }  
431 -  
432 - template<ReadSource S>  
433 - requires (!std::same_as<std::decay_t<S>, any_read_source>)  
DCB 434 - 9 any_read_source::any_read_source(S s)  
DCB 435 - 9 : vt_(&vtable_for_impl<S>::value)  
436 - {  
437 - struct guard {  
438 - any_read_source* self;  
439 - bool committed = false;  
DCB 440 - 9 ~guard() {  
DCB 441 - 9 if(!committed && self->storage_) {  
DCB 442 - 1 if(self->source_)  
443 - self->vt_->destroy(self->source_); // LCOV_EXCL_LINE OOM rollback: only when the cached-awaitable allocation throws  
DCB 444 - 1 ::operator delete(self->storage_);  
DCB 445 - 1 self->storage_ = nullptr;  
DCB 446 - 1 self->source_ = nullptr;  
447 - }  
DCB 448 - 9 }  
DCB 449 - 9 } g{this};  
450 -  
DCB 451 - 9 storage_ = ::operator new(sizeof(S));  
DCB 452 - 9 source_ = ::new(storage_) S(std::move(s));  
453 -  
454 - // Preallocate the awaitable storage  
DCB 455 - 8 cached_awaitable_ = ::operator new(vt_->awaitable_size);  
456 -  
DCB 457 - 8 g.committed = true;  
DCB 458 - 9 }  
459 -  
460 - template<ReadSource S>  
DCB 461 - 136 any_read_source::any_read_source(S* s)  
DCB 462 - 136 : source_(s)  
DCB 463 - 136 , vt_(&vtable_for_impl<S>::value)  
464 - {  
465 - // Preallocate the awaitable storage  
DCB 466 - 136 cached_awaitable_ = ::operator new(vt_->awaitable_size);  
DCB 467 - 136 }  
468 -  
469 - template<MutableBufferSequence MB>  
470 - auto  
DCB 471 - 54 any_read_source::read_some(MB buffers)  
472 - {  
473 - struct awaitable  
474 - {  
475 - any_read_source* self_;  
476 - detail::mutable_buffer_array<detail::max_iovec_> ba_;  
477 -  
DCB 478 - 54 awaitable(any_read_source* self, MB const& buffers)  
DCB 479 - 54 : self_(self)  
DCB 480 - 54 , ba_(buffers)  
481 - {  
DCB 482 - 54 }  
483 -  
484 - bool  
DCB 485 - 54 await_ready() const noexcept  
486 - {  
DCB 487 - 54 return ba_.to_span().empty();  
488 - }  
489 -  
490 - std::coroutine_handle<>  
DCB 491 - 52 await_suspend(std::coroutine_handle<> h, io_env const* env)  
492 - {  
DCB 493 - 52 self_->active_ops_ = self_->vt_->construct_read_some_awaitable(  
DCB 494 - 52 self_->source_,  
DCB 495 - 52 self_->cached_awaitable_,  
DCB 496 - 52 ba_.to_span());  
497 -  
DCB 498 - 52 if(self_->active_ops_->await_ready(self_->cached_awaitable_))  
DUB 499 - ✗ return h;  
500 -  
DCB 501 - 52 return self_->active_ops_->await_suspend(  
DCB 502 - 52 self_->cached_awaitable_, h, env);  
503 - }  
504 -  
505 - io_result<std::size_t>  
DCB 506 - 52 await_resume()  
507 - {  
DCB 508 - 52 if(ba_.to_span().empty())  
DCB 509 - 2 return {{}, 0};  
510 -  
511 - struct guard {  
512 - any_read_source* self;  
DCB 513 - 50 ~guard() {  
DCB 514 - 50 self->active_ops_->destroy(self->cached_awaitable_);  
DCB 515 - 50 self->active_ops_ = nullptr;  
DCB 516 - 50 }  
DCB 517 - 50 } g{self_};  
DCB 518 - 50 return self_->active_ops_->await_resume(  
DCB 519 - 50 self_->cached_awaitable_);  
DCB 520 - 50 }  
521 - };  
DCB 522 - 54 return awaitable(this, buffers);  
523 - }  
524 -  
525 - inline auto  
DCB 526 - 117 any_read_source::read_(std::span<mutable_buffer const> buffers)  
527 - {  
528 - struct awaitable  
529 - {  
530 - any_read_source* self_;  
531 - std::span<mutable_buffer const> buffers_;  
532 -  
533 - bool  
DCB 534 - 117 await_ready() const noexcept  
535 - {  
DCB 536 - 117 return false;  
537 - }  
538 -  
539 - std::coroutine_handle<>  
DCB 540 - 117 await_suspend(std::coroutine_handle<> h, io_env const* env)  
541 - {  
DCB 542 - 234 self_->active_ops_ = self_->vt_->construct_read_awaitable(  
DCB 543 - 117 self_->source_,  
DCB 544 - 117 self_->cached_awaitable_,  
545 - buffers_);  
546 -  
DCB 547 - 117 if(self_->active_ops_->await_ready(self_->cached_awaitable_))  
DUB 548 - ✗ return h;  
549 -  
DCB 550 - 117 return self_->active_ops_->await_suspend(  
DCB 551 - 117 self_->cached_awaitable_, h, env);  
552 - }  
553 -  
554 - io_result<std::size_t>  
DCB 555 - 117 await_resume()  
556 - {  
557 - struct guard {  
558 - any_read_source* self;  
DCB 559 - 117 ~guard() {  
DCB 560 - 117 self->active_ops_->destroy(self->cached_awaitable_);  
DCB 561 - 117 self->active_ops_ = nullptr;  
DCB 562 - 117 }  
DCB 563 - 117 } g{self_};  
DCB 564 - 117 return self_->active_ops_->await_resume(  
DCB 565 - 202 self_->cached_awaitable_);  
DCB 566 - 117 }  
567 - };  
DCB 568 - 117 return awaitable{this, buffers};  
569 - }  
570 -  
571 - template<MutableBufferSequence MB>  
572 - io_task<std::size_t>  
DCB 573 - 111 any_read_source::read(MB buffers)  
574 - {  
575 - buffer_param bp(buffers);  
576 - std::size_t total = 0;  
577 -  
578 - for(;;)  
579 - {  
580 - auto bufs = bp.data();  
581 - if(bufs.empty())  
582 - break;  
583 -  
584 - auto [ec, n] = co_await read_(bufs);  
585 - total += n;  
586 - if(ec)  
587 - co_return {ec, total};  
588 - bp.consume(n);  
589 - }  
590 -  
591 - co_return {{}, total};  
DCB 592 - 222 }  
593 -  
594 - } // namespace capy  
595 - } // namespace boost  
596 -  
597 - #endif