include/boost/capy/io/any_read_stream.hpp

91.7% Lines (77 / 84, 43 excl) 63.7% Functions (58 / 91, 7 excl)

[ ] prev/next file n p prev/next uncovered line

any_read_stream.hpp

f(x) Functions (98)

Function Calls Lines Blocks

Line	Hits	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	220x	disown() noexcept
89		{
90	220x	if(owner_)
91	8x	*std::exchange(owner_, nullptr) = false;
92	220x	}
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	10x	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	2x	any_read_stream(any_read_stream&& other) noexcept
137	2x	: stream_(std::exchange(other.stream_, nullptr))
138	2x	, vt_(std::exchange(other.vt_, nullptr))
139	2x	, cached_awaitable_(std::exchange(other.cached_awaitable_, nullptr))
140	2x	, storage_(std::exchange(other.storage_, nullptr))
141	2x	, awaitable_active_(std::exchange(other.awaitable_active_, false))
142		{
143	2x	other.disown();
144	2x	}
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	31x	has_value() const noexcept
185		{
186	31x	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	3x	operator bool() const noexcept
196		{
197	3x	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	6x	do_destroy_impl(void* stream) noexcept
290		{
291	6x	static_cast<S*>(stream)->~S();
292	6x	}
293
294		static void
295	108x	construct_awaitable_impl(
296		void* stream,
297		void* storage,
298		std::span<mutable_buffer const> buffers)
299		{
300	108x	auto& s = static_cast<S>(stream);
301	108x	::new(storage) Awaitable(s.read_some(buffers));
302	108x	}
303
304		static constexpr vtable value = {
305		&construct_awaitable_impl,
306	108x	+[](void* p) {
307	108x	return static_cast<Awaitable*>(p)->await_ready();
308		},
309	84x	+[](void* p, std::coroutine_handle<> h, await_context const* env) {
310	84x	return detail::call_await_suspend(
311	84x	static_cast<Awaitable*>(p), h, env);
312		},
313	97x	+[](void* p) {
314	97x	return static_cast<Awaitable*>(p)->await_resume();
315		},
316	125x	+[](void* p) noexcept {
317	17x	static_cast<Awaitable*>(p)->~Awaitable();
318		},
319		sizeof(Awaitable),
320		alignof(Awaitable),
321		&do_destroy_impl
322		};
323		};
324
325		inline void
326	198x	any_read_stream::end_operation() noexcept
327		{
328	198x	disown();
329	198x	if(awaitable_active_)
330		{
331	8x	vt_->destroy_awaitable(cached_awaitable_);
332	8x	awaitable_active_ = false;
333		}
334	198x	}
335
336		inline
337	135x	any_read_stream::~any_read_stream()
338		{
339	135x	end_operation();
340	135x	if(cached_awaitable_)
341	110x	::operator delete(cached_awaitable_);
342	135x	if(storage_)
343		{
344	4x	vt_->destroy(stream_);
345	4x	::operator delete(storage_);
346		}
347	135x	}
348
349		inline any_read_stream&
350	20x	any_read_stream::operator=(any_read_stream&& other) noexcept
351		{
352	20x	if(this != &other)
353		{
354	20x	end_operation();
355	20x	other.disown();
356	20x	if(cached_awaitable_)
357	8x	::operator delete(cached_awaitable_);
358	20x	if(storage_)
359		{
360	2x	vt_->destroy(stream_);
361	2x	::operator delete(storage_);
362		}
363	20x	stream_ = std::exchange(other.stream_, nullptr);
364	20x	vt_ = std::exchange(other.vt_, nullptr);
365	20x	cached_awaitable_ = std::exchange(other.cached_awaitable_, nullptr);
366	20x	storage_ = std::exchange(other.storage_, nullptr);
367	20x	awaitable_active_ = std::exchange(other.awaitable_active_, false);
368		}
369	20x	return *this;
370		}
371
372		template<ReadStream S>
373		requires (!std::same_as<std::decay_t<S>, any_read_stream>)
374	7x	any_read_stream::any_read_stream(S s)
375	7x	: vt_(&vtable_for_impl<S>::value)
376		{
377		struct guard {
378		any_read_stream* self;
379		bool committed = false;
380	✗	~guard() {
381	✗	if(!committed && self->storage_) {
382	✗	if(self->stream_)
383	−	self->vt_->destroy(self->stream_); // LCOV_EXCL_LINE OOM rollback: only when the cached-awaitable allocation throws
384	✗	::operator delete(self->storage_);
385	✗	self->storage_ = nullptr;
386	✗	self->stream_ = nullptr;
387		}
388	✗	}
389	7x	} g{this};
390
391	7x	storage_ = ::operator new(sizeof(S));
392	7x	stream_ = ::new(storage_) S(std::move(s));
393
394		// Preallocate the awaitable storage
395	6x	cached_awaitable_ = ::operator new(vt_->awaitable_size);
396
397	6x	g.committed = true;
398	7x	}
399
400		template<ReadStream S>
401	112x	any_read_stream::any_read_stream(S* s)
402	112x	: stream_(s)
403	112x	, vt_(&vtable_for_impl<S>::value)
404		{
405		// Preallocate the awaitable storage
406	112x	cached_awaitable_ = ::operator new(vt_->awaitable_size);
407	112x	}
408
409		template<MutableBufferSequence MB>
410		auto
411	109x	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	109x	detail::mutable_buffer_array<detail::max_iovec_>(buffers)};
496		}
497
498		} // namespace capy
499		} // namespace boost
500
501		#endif
502