include/boost/capy/io/any_write_stream.hpp

100.0% Lines (94/0/94) 79.3% List of functions (65/0/82)

[ ] prev/next file

any_write_stream.hpp

f(x) Functions (82)

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_WRITE_STREAM_HPP
12		#define BOOST_CAPY_IO_ANY_WRITE_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/write_stream.hpp>
20		#include <coroutine>
21		#include <boost/capy/ex/io_env.hpp>
22		#include <boost/capy/io_result.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		/** Dispatches `write_some` through a type-erased vtable, using preallocated awaitable storage.
38
39		This class provides type erasure for any type satisfying the
40		@ref WriteStream concept, enabling runtime polymorphism for
41		write 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 stream.
47		- Reference: Pass a pointer to wrap without ownership. The
48		pointed-to stream 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 stream'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 stream
68		any_write_stream owning_stream(socket{ioc});
69
70		// Reference - wraps without ownership
71		socket sock(ioc);
72		any_write_stream ref_stream(&sock);
73
74		char data[] = "hello";
75		const_buffer buf(data, sizeof(data));
76		auto [ec, n] = co_await owning_stream.write_some(std::span(&buf, 1));
77		@endcode
78
79		@see any_read_stream, any_stream, WriteStream
80		*/
81		class any_write_stream
82		{
83		struct vtable;
84
85		template<WriteStream S>
86		struct vtable_for_impl;
87
88		// ordered for cache line coherence
89		void* stream_ = nullptr;
90		vtable const* vt_ = nullptr;
91		void* cached_awaitable_ = nullptr;
92		void* storage_ = nullptr;
93		bool awaitable_active_ = false;
94
95		public:
96		/** Destructor.
97
98		Destroys the owned stream (if any) and releases the cached
99		awaitable storage.
100		*/
101		~any_write_stream();
102
103		/** Construct a default instance.
104
105		Constructs an empty wrapper. @ref has_value and `operator bool`
106		report the empty state; calling @ref write_some before the
107		wrapper holds a stream is undefined behavior.
108		*/
109	4x	any_write_stream() = default;
110
111		/** Non-copyable.
112
113		The awaitable cache is per-instance and cannot be shared.
114
115		@param other The wrapper that would be copied.
116		*/
117		any_write_stream(any_write_stream const& other) = delete;
118
119		/** Copy assignment is disabled.
120
121		The awaitable cache is per-instance and cannot be shared.
122
123		@param other The wrapper that would be assigned from.
124
125		@return A reference to `*this`.
126		*/
127		any_write_stream& operator=(any_write_stream const& other) = delete;
128
129		/** Construct by moving.
130
131		Transfers ownership of the wrapped stream (if owned) and
132		cached awaitable storage from `other`. After the move, `other` is
133		in a default-constructed state.
134
135		@param other The wrapper to move from.
136		*/
137	2x	any_write_stream(any_write_stream&& other) noexcept
138	2x	: stream_(std::exchange(other.stream_, nullptr))
139	2x	, vt_(std::exchange(other.vt_, nullptr))
140	2x	, cached_awaitable_(std::exchange(other.cached_awaitable_, nullptr))
141	2x	, storage_(std::exchange(other.storage_, nullptr))
142	2x	, awaitable_active_(std::exchange(other.awaitable_active_, false))
143		{
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_write_stream&
155		operator=(any_write_stream&& other) noexcept;
156
157		/** Construct by taking ownership of a WriteStream.
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<WriteStream S>
165		requires (!std::same_as<std::decay_t<S>, any_write_stream>)
166		any_write_stream(S s);
167
168		/** Construct by wrapping a WriteStream 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<WriteStream S>
176		any_write_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	26x	has_value() const noexcept
185		{
186	26x	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 write operation.
201
202		Writes data from the provided buffer sequence. The operation
203		completes when at least one byte is written, or an error
204		occurs.
205
206		@param buffers The buffer sequence containing data to write.
207		Passed by value to ensure the sequence lives in the
208		coroutine frame 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:
215		@li The buffer sequence is empty, returning `{error_code{}, 0}`.
216		@li The underlying stream's awaitable reports immediate
217		readiness via `await_ready`.
218
219		@note This is a partial operation and may not process the
220		entire buffer sequence. Use the composed @ref write algorithm
221		for guaranteed complete transfer.
222
223		@par Preconditions
224		The wrapper must contain a valid stream (`has_value() == true`).
225
226		@par After an Error
227		A subsequent call is permitted. The wrapper forwards directly
228		to the underlying stream, imposing no stricter rule than
229		@ref WriteStream.
230		*/
231		template<ConstBufferSequence CB>
232		auto
233		write_some(CB buffers);
234
235		protected:
236		/** Rebind to a new stream after move.
237
238		Updates the internal pointer to reference a new stream object.
239		Used by owning wrappers after move assignment when the owned
240		object has moved to a new location.
241
242		@param new_stream The new stream to bind to. Must be the same
243		type as the original stream.
244
245		@note Terminates if called with a stream of different type
246		than the original.
247		*/
248		template<WriteStream S>
249		void
250		rebind(S& new_stream) noexcept
251		{
252		if(vt_ != &vtable_for_impl<S>::value)
253		std::terminate();
254		stream_ = &new_stream;
255		}
256		};
257
258		struct any_write_stream::vtable
259		{
260		// ordered by call frequency for cache line coherence
261		void (*construct_awaitable)(
262		void* stream,
263		void* storage,
264		std::span<const_buffer const> buffers);
265		bool (await_ready)(void);
266		std::coroutine_handle<> (await_suspend)(void, std::coroutine_handle<>, io_env const*);
267		io_result<std::size_t> (await_resume)(void);
268		void (destroy_awaitable)(void) noexcept;
269		std::size_t awaitable_size;
270		std::size_t awaitable_align;
271		void (destroy)(void) noexcept;
272		};
273
274		template<WriteStream S>
275		struct any_write_stream::vtable_for_impl
276		{
277		using Awaitable = decltype(std::declval<S&>().write_some(
278		std::span<const_buffer const>{}));
279
280		static void
281	3x	do_destroy_impl(void* stream) noexcept
282		{
283	3x	static_cast<S*>(stream)->~S();
284	3x	}
285
286		static void
287	89x	construct_awaitable_impl(
288		void* stream,
289		void* storage,
290		std::span<const_buffer const> buffers)
291		{
292	89x	auto& s = static_cast<S>(stream);
293	89x	::new(storage) Awaitable(s.write_some(buffers));
294	89x	}
295
296		static constexpr vtable value = {
297		&construct_awaitable_impl,
298	89x	+[](void* p) {
299	89x	return static_cast<Awaitable*>(p)->await_ready();
300		},
301	79x	+[](void* p, std::coroutine_handle<> h, io_env const* env) {
302	79x	return detail::call_await_suspend(
303	79x	static_cast<Awaitable*>(p), h, env);
304		},
305	87x	+[](void* p) {
306	87x	return static_cast<Awaitable*>(p)->await_resume();
307		},
308	91x	+[](void* p) noexcept {
309	26x	static_cast<Awaitable*>(p)->~Awaitable();
310		},
311		sizeof(Awaitable),
312		alignof(Awaitable),
313		&do_destroy_impl
314		};
315		};
316
317		inline
318	117x	any_write_stream::~any_write_stream()
319		{
320	117x	if(storage_)
321		{
322	2x	vt_->destroy(stream_);
323	2x	::operator delete(storage_);
324		}
325	117x	if(cached_awaitable_)
326		{
327	102x	if(awaitable_active_)
328	1x	vt_->destroy_awaitable(cached_awaitable_);
329	102x	::operator delete(cached_awaitable_);
330		}
331	117x	}
332
333		inline any_write_stream&
334	10x	any_write_stream::operator=(any_write_stream&& other) noexcept
335		{
336	10x	if(this != &other)
337		{
338	10x	if(storage_)
339		{
340	1x	vt_->destroy(stream_);
341	1x	::operator delete(storage_);
342		}
343	10x	if(cached_awaitable_)
344		{
345	4x	if(awaitable_active_)
346	1x	vt_->destroy_awaitable(cached_awaitable_);
347	4x	::operator delete(cached_awaitable_);
348		}
349	10x	stream_ = std::exchange(other.stream_, nullptr);
350	10x	vt_ = std::exchange(other.vt_, nullptr);
351	10x	cached_awaitable_ = std::exchange(other.cached_awaitable_, nullptr);
352	10x	storage_ = std::exchange(other.storage_, nullptr);
353	10x	awaitable_active_ = std::exchange(other.awaitable_active_, false);
354		}
355	10x	return *this;
356		}
357
358		template<WriteStream S>
359		requires (!std::same_as<std::decay_t<S>, any_write_stream>)
360	4x	any_write_stream::any_write_stream(S s)
361	4x	: vt_(&vtable_for_impl<S>::value)
362		{
363		struct guard {
364		any_write_stream* self;
365		bool committed = false;
366		~guard() {
367		if(!committed && self->storage_) {
368		if(self->stream_)
369		self->vt_->destroy(self->stream_); // LCOV_EXCL_LINE OOM rollback: only when the cached-awaitable allocation throws
370		::operator delete(self->storage_);
371		self->storage_ = nullptr;
372		self->stream_ = nullptr;
373		}
374		}
375	4x	} g{this};
376
377	4x	storage_ = ::operator new(sizeof(S));
378	4x	stream_ = ::new(storage_) S(std::move(s));
379
380		// Preallocate the awaitable storage
381	3x	cached_awaitable_ = ::operator new(vt_->awaitable_size);
382
383	3x	g.committed = true;
384	4x	}
385
386		template<WriteStream S>
387	103x	any_write_stream::any_write_stream(S* s)
388	103x	: stream_(s)
389	103x	, vt_(&vtable_for_impl<S>::value)
390		{
391		// Preallocate the awaitable storage
392	103x	cached_awaitable_ = ::operator new(vt_->awaitable_size);
393	103x	}
394
395		template<ConstBufferSequence CB>
396		auto
397	93x	any_write_stream::write_some(CB buffers)
398		{
399		struct awaitable
400		{
401		any_write_stream* self_;
402		detail::const_buffer_array<detail::max_iovec_> ba_;
403
404	93x	awaitable(
405		any_write_stream* self,
406		CB const& buffers) noexcept
407	93x	: self_(self)
408	93x	, ba_(buffers)
409		{
410	93x	}
411
412		bool
413	93x	await_ready() const noexcept
414		{
415	93x	return ba_.to_span().empty();
416		}
417
418		std::coroutine_handle<>
419	89x	await_suspend(std::coroutine_handle<> h, io_env const* env)
420		{
421	89x	self_->vt_->construct_awaitable(
422	89x	self_->stream_,
423	89x	self_->cached_awaitable_,
424	89x	ba_.to_span());
425	89x	self_->awaitable_active_ = true;
426
427	89x	if(self_->vt_->await_ready(self_->cached_awaitable_))
428	10x	return h;
429
430	79x	return self_->vt_->await_suspend(
431	79x	self_->cached_awaitable_, h, env);
432		}
433
434		io_result<std::size_t>
435	91x	await_resume()
436		{
437	91x	if(!self_->awaitable_active_)
438	4x	return {{}, 0};
439		struct guard {
440		any_write_stream* self;
441	87x	~guard() {
442	87x	self->vt_->destroy_awaitable(self->cached_awaitable_);
443	87x	self->awaitable_active_ = false;
444	87x	}
445	87x	} g{self_};
446	87x	return self_->vt_->await_resume(
447	87x	self_->cached_awaitable_);
448	87x	}
449		};
450	93x	return awaitable{this, buffers};
451		}
452
453		} // namespace capy
454		} // namespace boost
455
456		#endif
457