include/boost/capy/io/any_write_stream.hpp

94.1% Lines (95/101) 89.8% List of functions (44/49)

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

any_write_stream.hpp

f(x) Functions (49)

Function Calls Lines Blocks

Line	Hits	Source 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_WRITE_STREAM_HPP
11		#define BOOST_CAPY_IO_ANY_WRITE_STREAM_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/buffers/buffer_array.hpp>
17		#include <boost/capy/concept/io_awaitable.hpp>
18		#include <boost/capy/concept/write_stream.hpp>
19		#include <coroutine>
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		/** Type-erased wrapper for any WriteStream.
37
38		This class provides type erasure for any type satisfying the
39		@ref WriteStream concept, enabling runtime polymorphism for
40		write 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		Operations complete immediately without suspending when the
57		buffer sequence is empty, or when the underlying stream's
58		awaitable reports readiness via `await_ready`.
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_write_stream stream(socket{ioc});
68
69		// Reference - wraps without ownership
70		socket sock(ioc);
71		any_write_stream stream(&sock);
72
73		const_buffer buf(data, size);
74		auto [ec, n] = co_await stream.write_some(std::span(&buf, 1));
75		@endcode
76
77		@see any_read_stream, any_stream, WriteStream
78		*/
79		class any_write_stream
80		{
81		struct vtable;
82
83		template<WriteStream S>
84		struct vtable_for_impl;
85
86		// ordered for cache line coherence
87		void* stream_ = nullptr;
88		vtable const* vt_ = nullptr;
89		void* cached_awaitable_ = nullptr;
90		void* storage_ = nullptr;
91		bool awaitable_active_ = false;
92
93		public:
94		/** Destructor.
95
96		Destroys the owned stream (if any) and releases the cached
97		awaitable storage.
98		*/
99		~any_write_stream();
100
101		/** Construct a default instance.
102
103		Constructs an empty wrapper. Operations on a default-constructed
104		wrapper result in undefined behavior.
105		*/
106	1x	any_write_stream() = default;
107
108		/** Non-copyable.
109
110		The awaitable cache is per-instance and cannot be shared.
111		*/
112		any_write_stream(any_write_stream const&) = delete;
113		any_write_stream& operator=(any_write_stream const&) = delete;
114
115		/** Construct by moving.
116
117		Transfers ownership of the wrapped stream (if owned) and
118		cached awaitable storage from `other`. After the move, `other` is
119		in a default-constructed state.
120
121		@param other The wrapper to move from.
122		*/
123	2x	any_write_stream(any_write_stream&& other) noexcept
124	2x	: stream_(std::exchange(other.stream_, nullptr))
125	2x	, vt_(std::exchange(other.vt_, nullptr))
126	2x	, cached_awaitable_(std::exchange(other.cached_awaitable_, nullptr))
127	2x	, storage_(std::exchange(other.storage_, nullptr))
128	2x	, awaitable_active_(std::exchange(other.awaitable_active_, false))
129		{
130	2x	}
131
132		/** Assign by moving.
133
134		Destroys any owned stream and releases existing resources,
135		then transfers ownership from `other`.
136
137		@param other The wrapper to move from.
138		@return Reference to this wrapper.
139		*/
140		any_write_stream&
141		operator=(any_write_stream&& other) noexcept;
142
143		/** Construct by taking ownership of a WriteStream.
144
145		Allocates storage and moves the stream into this wrapper.
146		The wrapper owns the stream and will destroy it.
147
148		@param s The stream to take ownership of.
149		*/
150		template<WriteStream S>
151		requires (!std::same_as<std::decay_t<S>, any_write_stream>)
152		any_write_stream(S s);
153
154		/** Construct by wrapping a WriteStream without ownership.
155
156		Wraps the given stream by pointer. The stream must remain
157		valid for the lifetime of this wrapper.
158
159		@param s Pointer to the stream to wrap.
160		*/
161		template<WriteStream S>
162		any_write_stream(S* s);
163
164		/** Check if the wrapper contains a valid stream.
165
166		@return `true` if wrapping a stream, `false` if default-constructed
167		or moved-from.
168		*/
169		bool
170	21x	has_value() const noexcept
171		{
172	21x	return stream_ != nullptr;
173		}
174
175		/** Check if the wrapper contains a valid stream.
176
177		@return `true` if wrapping a stream, `false` if default-constructed
178		or moved-from.
179		*/
180		explicit
181	3x	operator bool() const noexcept
182		{
183	3x	return has_value();
184		}
185
186		/** Initiate an asynchronous write operation.
187
188		Writes data from the provided buffer sequence. The operation
189		completes when at least one byte has been written, or an error
190		occurs.
191
192		@param buffers The buffer sequence containing data to write.
193		Passed by value to ensure the sequence lives in the
194		coroutine frame across suspension points.
195
196		@return An awaitable that await-returns `(error_code,std::size_t)`.
197
198		@par Immediate Completion
199		The operation completes immediately without suspending
200		the calling coroutine when:
201		@li The buffer sequence is empty, returning `{error_code{}, 0}`.
202		@li The underlying stream's awaitable reports immediate
203		readiness via `await_ready`.
204
205		@note This is a partial operation and may not process the
206		entire buffer sequence. Use the composed @ref write algorithm
207		for guaranteed complete transfer.
208
209		@par Preconditions
210		The wrapper must contain a valid stream (`has_value() == true`).
211		*/
212		template<ConstBufferSequence CB>
213		auto
214		write_some(CB buffers);
215
216		protected:
217		/** Rebind to a new stream after move.
218
219		Updates the internal pointer to reference a new stream object.
220		Used by owning wrappers after move assignment when the owned
221		object has moved to a new location.
222
223		@param new_stream The new stream to bind to. Must be the same
224		type as the original stream.
225
226		@note Terminates if called with a stream of different type
227		than the original.
228		*/
229		template<WriteStream S>
230		void
231		rebind(S& new_stream) noexcept
232		{
233		if(vt_ != &vtable_for_impl<S>::value)
234		std::terminate();
235		stream_ = &new_stream;
236		}
237		};
238
239		struct any_write_stream::vtable
240		{
241		// ordered by call frequency for cache line coherence
242		void (*construct_awaitable)(
243		void* stream,
244		void* storage,
245		std::span<const_buffer const> buffers);
246		bool (await_ready)(void);
247		std::coroutine_handle<> (await_suspend)(void, std::coroutine_handle<>, io_env const*);
248		io_result<std::size_t> (await_resume)(void);
249		void (destroy_awaitable)(void) noexcept;
250		std::size_t awaitable_size;
251		std::size_t awaitable_align;
252		void (destroy)(void) noexcept;
253		};
254
255		template<WriteStream S>
256		struct any_write_stream::vtable_for_impl
257		{
258		using Awaitable = decltype(std::declval<S&>().write_some(
259		std::span<const_buffer const>{}));
260
261		static void
262	2x	do_destroy_impl(void* stream) noexcept
263		{
264	2x	static_cast<S*>(stream)->~S();
265	2x	}
266
267		static void
268	76x	construct_awaitable_impl(
269		void* stream,
270		void* storage,
271		std::span<const_buffer const> buffers)
272		{
273	76x	auto& s = static_cast<S>(stream);
274	76x	::new(storage) Awaitable(s.write_some(buffers));
275	76x	}
276
277		static constexpr vtable value = {
278		&construct_awaitable_impl,
279	76x	+[](void* p) {
280	76x	return static_cast<Awaitable*>(p)->await_ready();
281		},
282	3x	+[](void* p, std::coroutine_handle<> h, io_env const* env) {
283	3x	return detail::call_await_suspend(
284	3x	static_cast<Awaitable*>(p), h, env);
285		},
286	74x	+[](void* p) {
287	74x	return static_cast<Awaitable*>(p)->await_resume();
288		},
289	79x	+[](void* p) noexcept {
290	13x	static_cast<Awaitable*>(p)->~Awaitable();
291		},
292		sizeof(Awaitable),
293		alignof(Awaitable),
294		&do_destroy_impl
295		};
296		};
297
298		inline
299	96x	any_write_stream::~any_write_stream()
300		{
301	96x	if(storage_)
302		{
303	2x	vt_->destroy(stream_);
304	2x	::operator delete(storage_);
305		}
306	96x	if(cached_awaitable_)
307		{
308	86x	if(awaitable_active_)
309	1x	vt_->destroy_awaitable(cached_awaitable_);
310	86x	::operator delete(cached_awaitable_);
311		}
312	96x	}
313
314		inline any_write_stream&
315	5x	any_write_stream::operator=(any_write_stream&& other) noexcept
316		{
317	5x	if(this != &other)
318		{
319	5x	if(storage_)
320		{
321	✗	vt_->destroy(stream_);
322	✗	::operator delete(storage_);
323		}
324	5x	if(cached_awaitable_)
325		{
326	2x	if(awaitable_active_)
327	1x	vt_->destroy_awaitable(cached_awaitable_);
328	2x	::operator delete(cached_awaitable_);
329		}
330	5x	stream_ = std::exchange(other.stream_, nullptr);
331	5x	vt_ = std::exchange(other.vt_, nullptr);
332	5x	cached_awaitable_ = std::exchange(other.cached_awaitable_, nullptr);
333	5x	storage_ = std::exchange(other.storage_, nullptr);
334	5x	awaitable_active_ = std::exchange(other.awaitable_active_, false);
335		}
336	5x	return *this;
337		}
338
339		template<WriteStream S>
340		requires (!std::same_as<std::decay_t<S>, any_write_stream>)
341	2x	any_write_stream::any_write_stream(S s)
342	2x	: vt_(&vtable_for_impl<S>::value)
343		{
344		struct guard {
345		any_write_stream* self;
346		bool committed = false;
347	2x	~guard() {
348	2x	if(!committed && self->storage_) {
349	✗	self->vt_->destroy(self->stream_);
350	✗	::operator delete(self->storage_);
351	✗	self->storage_ = nullptr;
352	✗	self->stream_ = nullptr;
353		}
354	2x	}
355	2x	} g{this};
356
357	2x	storage_ = ::operator new(sizeof(S));
358	2x	stream_ = ::new(storage_) S(std::move(s));
359
360		// Preallocate the awaitable storage
361	2x	cached_awaitable_ = ::operator new(vt_->awaitable_size);
362
363	2x	g.committed = true;
364	2x	}
365
366		template<WriteStream S>
367	86x	any_write_stream::any_write_stream(S* s)
368	86x	: stream_(s)
369	86x	, vt_(&vtable_for_impl<S>::value)
370		{
371		// Preallocate the awaitable storage
372	86x	cached_awaitable_ = ::operator new(vt_->awaitable_size);
373	86x	}
374
375		template<ConstBufferSequence CB>
376		auto
377	80x	any_write_stream::write_some(CB buffers)
378		{
379		struct awaitable
380		{
381		any_write_stream* self_;
382		const_buffer_array<detail::max_iovec_> ba_;
383
384	80x	awaitable(
385		any_write_stream* self,
386		CB const& buffers) noexcept
387	80x	: self_(self)
388	80x	, ba_(buffers)
389		{
390	80x	}
391
392		bool
393	80x	await_ready() const noexcept
394		{
395	80x	return ba_.to_span().empty();
396		}
397
398		std::coroutine_handle<>
399	76x	await_suspend(std::coroutine_handle<> h, io_env const* env)
400		{
401	76x	self_->vt_->construct_awaitable(
402	76x	self_->stream_,
403	76x	self_->cached_awaitable_,
404	76x	ba_.to_span());
405	76x	self_->awaitable_active_ = true;
406
407	76x	if(self_->vt_->await_ready(self_->cached_awaitable_))
408	73x	return h;
409
410	3x	return self_->vt_->await_suspend(
411	3x	self_->cached_awaitable_, h, env);
412		}
413
414		io_result<std::size_t>
415	78x	await_resume()
416		{
417	78x	if(!self_->awaitable_active_)
418	4x	return {{}, 0};
419		struct guard {
420		any_write_stream* self;
421	74x	~guard() {
422	74x	self->vt_->destroy_awaitable(self->cached_awaitable_);
423	74x	self->awaitable_active_ = false;
424	74x	}
425	74x	} g{self_};
426	74x	return self_->vt_->await_resume(
427	74x	self_->cached_awaitable_);
428	74x	}
429		};
430	80x	return awaitable{this, buffers};
431		}
432
433		} // namespace capy
434		} // namespace boost
435
436		#endif
437