include/boost/capy/io/any_read_stream.hpp

94.7% Lines (71/21/96) 65.7% List of functions (46/8/78)

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

any_read_stream.hpp

f(x) Functions (78)

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/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	4x	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	4x	any_read_stream(any_read_stream&& other) noexcept
137	4x	: stream_(std::exchange(other.stream_, nullptr))
138	4x	, vt_(std::exchange(other.vt_, nullptr))
139	4x	, cached_awaitable_(std::exchange(other.cached_awaitable_, nullptr))
140	4x	, storage_(std::exchange(other.storage_, nullptr))
141	4x	, awaitable_active_(std::exchange(other.awaitable_active_, false))
142		{
143	4x	}
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	31x	has_value() const noexcept
184		{
185	31x	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	3x	operator bool() const noexcept
195		{
196	3x	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	4x	do_destroy_impl(void* stream) noexcept
279		{
280	4x	static_cast<S*>(stream)->~S();
281	4x	}
282
283		static void
284	103x	construct_awaitable_impl(
285		void* stream,
286		void* storage,
287		std::span<mutable_buffer const> buffers)
288		{
289	103x	auto& s = static_cast<S>(stream);
290	103x	::new(storage) Awaitable(s.read_some(buffers));
291	103x	}
292
293		static constexpr vtable value = {
294		&construct_awaitable_impl,
295	103x	+[](void* p) {
296	103x	return static_cast<Awaitable*>(p)->await_ready();
297		},
298	77x	+[](void* p, std::coroutine_handle<> h, io_env const* env) {
299	77x	return detail::call_await_suspend(
300	77x	static_cast<Awaitable*>(p), h, env);
301		},
302	101x	+[](void* p) {
303	101x	return static_cast<Awaitable*>(p)->await_resume();
304		},
305	115x	+[](void* p) noexcept {
306	26x	static_cast<Awaitable*>(p)->~Awaitable();
307		},
308		sizeof(Awaitable),
309		alignof(Awaitable),
310		&do_destroy_impl
311		};
312		};
313
314		inline
315	123x	any_read_stream::~any_read_stream()
316		{
317	123x	if(storage_)
318		{
319	3x	vt_->destroy(stream_);
320	3x	::operator delete(storage_);
321		}
322	123x	if(cached_awaitable_)
323		{
324	106x	if(awaitable_active_)
325	1x	vt_->destroy_awaitable(cached_awaitable_);
326	106x	::operator delete(cached_awaitable_);
327		}
328	123x	}
329
330		inline any_read_stream&
331	10x	any_read_stream::operator=(any_read_stream&& other) noexcept
332		{
333	10x	if(this != &other)
334		{
335	10x	if(storage_)
336		{
337	1x	vt_->destroy(stream_);
338	1x	::operator delete(storage_);
339		}
340	10x	if(cached_awaitable_)
341		{
342	4x	if(awaitable_active_)
343	1x	vt_->destroy_awaitable(cached_awaitable_);
344	4x	::operator delete(cached_awaitable_);
345		}
346	10x	stream_ = std::exchange(other.stream_, nullptr);
347	10x	vt_ = std::exchange(other.vt_, nullptr);
348	10x	cached_awaitable_ = std::exchange(other.cached_awaitable_, nullptr);
349	10x	storage_ = std::exchange(other.storage_, nullptr);
350	10x	awaitable_active_ = std::exchange(other.awaitable_active_, false);
351		}
352	10x	return *this;
353		}
354
355		template<ReadStream S>
356		requires (!std::same_as<std::decay_t<S>, any_read_stream>)
357	5x	any_read_stream::any_read_stream(S s)
358	5x	: vt_(&vtable_for_impl<S>::value)
359		{
360		struct guard {
361		any_read_stream* self;
362		bool committed = false;
363	1x	~guard() {
364	1x	if(!committed && self->storage_) {
365	✗	if(self->stream_)
366	−	self->vt_->destroy(self->stream_); // LCOV_EXCL_LINE OOM rollback: only when the cached-awaitable allocation throws
367	✗	::operator delete(self->storage_);
368	✗	self->storage_ = nullptr;
369	✗	self->stream_ = nullptr;
370		}
371	1x	}
372	5x	} g{this};
373
374	5x	storage_ = ::operator new(sizeof(S));
375	5x	stream_ = ::new(storage_) S(std::move(s));
376
377		// Preallocate the awaitable storage
378	4x	cached_awaitable_ = ::operator new(vt_->awaitable_size);
379
380	4x	g.committed = true;
381	5x	}
382
383		template<ReadStream S>
384	106x	any_read_stream::any_read_stream(S* s)
385	106x	: stream_(s)
386	106x	, vt_(&vtable_for_impl<S>::value)
387		{
388		// Preallocate the awaitable storage
389	106x	cached_awaitable_ = ::operator new(vt_->awaitable_size);
390	106x	}
391
392		template<MutableBufferSequence MB>
393		auto
394	103x	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	103x	detail::mutable_buffer_array<detail::max_iovec_>(buffers)};
442	103x	}
443
444		} // namespace capy
445		} // namespace boost
446
447		#endif
448