include/boost/capy/io/any_write_stream.hpp

100.0% Lines (123 / 123) 63.4% Functions (64 / 101)

[ ] prev/next file

any_write_stream.hpp

f(x) Functions (101)

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/context_awaitable.hpp>
19		#include <boost/capy/concept/write_stream.hpp>
20		#include <coroutine>
21		#include <boost/capy/ex/await_context.hpp>
22		#include <boost/capy/results.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		@par !example example
67
68
69		@see any_read_stream, any_stream, WriteStream
70		*/
71		class any_write_stream
72		{
73		struct vtable;
74
75		template<WriteStream S>
76		struct vtable_for_impl;
77
78		// ordered for cache line coherence
79		void* stream_ = nullptr;
80		vtable const* vt_ = nullptr;
81		void* cached_awaitable_ = nullptr;
82		void* storage_ = nullptr;
83		bool awaitable_active_ = false;
84		// The flag of the awaitable whose operation is active, cleared
85		// when the stream ends that operation itself.
86		bool* owner_ = nullptr;
87
88		void
89	208x	disown() noexcept
90		{
91	208x	if(owner_)
92	6x	*std::exchange(owner_, nullptr) = false;
93	208x	}
94
95		public:
96		/** Destructor.
97
98		Ends any operation in flight, then destroys the owned stream
99		(if any) and releases the cached 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	10x	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	other.disown();
145	2x	}
146
147		/** Assign by moving.
148
149		Destroys any owned stream and releases existing resources,
150		then transfers ownership from `other`.
151
152		@param other The wrapper to move from.
153		@return Reference to this wrapper.
154		*/
155		any_write_stream&
156		operator=(any_write_stream&& other) noexcept;
157
158		/** Construct by taking ownership of a WriteStream.
159
160		Allocates storage and moves the stream into this wrapper.
161		The wrapper owns the stream and destroys it.
162
163		@param s The stream to take ownership of.
164		*/
165		template<WriteStream S>
166		requires (!std::same_as<std::decay_t<S>, any_write_stream>)
167		any_write_stream(S s);
168
169		/** Construct by wrapping a WriteStream without ownership.
170
171		Wraps the given stream by pointer. The stream must remain
172		valid for the lifetime of this wrapper.
173
174		@param s Pointer to the stream to wrap.
175		*/
176		template<WriteStream S>
177		any_write_stream(S* s);
178
179		/** Check if the wrapper contains a valid stream.
180
181		@return `true` if wrapping a stream, `false` if default-constructed
182		or moved-from.
183		*/
184		bool
185	26x	has_value() const noexcept
186		{
187	26x	return stream_ != nullptr;
188		}
189
190		/** Check if the wrapper contains a valid stream.
191
192		@return `true` if wrapping a stream, `false` if default-constructed
193		or moved-from.
194		*/
195		explicit
196	3x	operator bool() const noexcept
197		{
198	3x	return has_value();
199		}
200
201		/** Initiate an asynchronous write operation.
202
203		Writes data from the provided buffer sequence. The operation
204		completes when at least one byte is written, or an error
205		occurs.
206
207		@param buffers The buffer sequence containing data to write.
208		Passed by value to ensure the sequence lives in the
209		coroutine frame across suspension points.
210
211		@return An awaitable that await-returns `(error_code,std::size_t)`.
212
213		@par Immediate Completion
214		The operation completes immediately without suspending
215		the calling coroutine when:
216		@li The buffer sequence is empty, returning `{error_code{}, 0}`.
217		@li The underlying stream's awaitable reports immediate
218		readiness via `await_ready`.
219
220		@note This is a partial operation and may not process the
221		entire buffer sequence. Use the composed @ref write algorithm
222		for guaranteed complete transfer.
223
224		@par Preconditions
225		The wrapper must contain a valid stream (`has_value() == true`).
226
227		@par After an Error
228		A subsequent call is permitted. The wrapper forwards directly
229		to the underlying stream, imposing no stricter rule than
230		@ref WriteStream.
231		*/
232		template<ConstBufferSequence CB>
233		auto
234		write_some(CB buffers);
235
236		protected:
237		/** End the operation in flight, if any.
238
239		Destroys the wrapped stream's active operation, which may refer
240		to the stream, so an owner must call this before destroying the
241		stream. The awaitable that started the operation no longer ends
242		it.
243		*/
244		void
245		end_operation() noexcept;
246
247		/** Rebind to a new stream after move.
248
249		Updates the internal pointer to reference a new stream object.
250		Used by owning wrappers after move assignment when the owned
251		object has moved to a new location.
252
253		@param new_stream The new stream to bind to. Must be the same
254		type as the original stream.
255
256		@note Terminates if called with a stream of different type
257		than the original.
258		*/
259		template<WriteStream S>
260		void
261		rebind(S& new_stream) noexcept
262		{
263		if(vt_ != &vtable_for_impl<S>::value)
264		std::terminate();
265		stream_ = &new_stream;
266		}
267		};
268
269		struct any_write_stream::vtable
270		{
271		// ordered by call frequency for cache line coherence
272		void (*construct_awaitable)(
273		void* stream,
274		void* storage,
275		std::span<const_buffer const> buffers);
276		bool (await_ready)(void);
277		std::coroutine_handle<> (await_suspend)(void, std::coroutine_handle<>, await_context const*);
278		results<std::size_t> (await_resume)(void);
279		void (destroy_awaitable)(void) noexcept;
280		std::size_t awaitable_size;
281		std::size_t awaitable_align;
282		void (destroy)(void) noexcept;
283		};
284
285		template<WriteStream S>
286		struct any_write_stream::vtable_for_impl
287		{
288		using Awaitable = decltype(std::declval<S&>().write_some(
289		std::span<const_buffer const>{}));
290
291		static void
292	3x	do_destroy_impl(void* stream) noexcept
293		{
294	3x	static_cast<S*>(stream)->~S();
295	3x	}
296
297		static void
298	86x	construct_awaitable_impl(
299		void* stream,
300		void* storage,
301		std::span<const_buffer const> buffers)
302		{
303	86x	auto& s = static_cast<S>(stream);
304	86x	::new(storage) Awaitable(s.write_some(buffers));
305	86x	}
306
307		static constexpr vtable value = {
308		&construct_awaitable_impl,
309	86x	+[](void* p) {
310	86x	return static_cast<Awaitable*>(p)->await_ready();
311		},
312	74x	+[](void* p, std::coroutine_handle<> h, await_context const* env) {
313	74x	return detail::call_await_suspend(
314	74x	static_cast<Awaitable*>(p), h, env);
315		},
316	77x	+[](void* p) {
317	77x	return static_cast<Awaitable*>(p)->await_resume();
318		},
319	95x	+[](void* p) noexcept {
320	12x	static_cast<Awaitable*>(p)->~Awaitable();
321		},
322		sizeof(Awaitable),
323		alignof(Awaitable),
324		&do_destroy_impl
325		};
326		};
327
328		inline void
329	187x	any_write_stream::end_operation() noexcept
330		{
331	187x	disown();
332	187x	if(awaitable_active_)
333		{
334	6x	vt_->destroy_awaitable(cached_awaitable_);
335	6x	awaitable_active_ = false;
336		}
337	187x	}
338
339		inline
340	125x	any_write_stream::~any_write_stream()
341		{
342	125x	end_operation();
343	125x	if(cached_awaitable_)
344	101x	::operator delete(cached_awaitable_);
345	125x	if(storage_)
346		{
347	2x	vt_->destroy(stream_);
348	2x	::operator delete(storage_);
349		}
350	125x	}
351
352		inline any_write_stream&
353	19x	any_write_stream::operator=(any_write_stream&& other) noexcept
354		{
355	19x	if(this != &other)
356		{
357	19x	end_operation();
358	19x	other.disown();
359	19x	if(cached_awaitable_)
360	7x	::operator delete(cached_awaitable_);
361	19x	if(storage_)
362		{
363	1x	vt_->destroy(stream_);
364	1x	::operator delete(storage_);
365		}
366	19x	stream_ = std::exchange(other.stream_, nullptr);
367	19x	vt_ = std::exchange(other.vt_, nullptr);
368	19x	cached_awaitable_ = std::exchange(other.cached_awaitable_, nullptr);
369	19x	storage_ = std::exchange(other.storage_, nullptr);
370	19x	awaitable_active_ = std::exchange(other.awaitable_active_, false);
371		}
372	19x	return *this;
373		}
374
375		template<WriteStream S>
376		requires (!std::same_as<std::decay_t<S>, any_write_stream>)
377	4x	any_write_stream::any_write_stream(S s)
378	4x	: vt_(&vtable_for_impl<S>::value)
379		{
380		struct guard {
381		any_write_stream* self;
382		bool committed = false;
383		~guard() {
384		if(!committed && self->storage_) {
385		if(self->stream_)
386		self->vt_->destroy(self->stream_); // LCOV_EXCL_LINE OOM rollback: only when the cached-awaitable allocation throws
387		::operator delete(self->storage_);
388		self->storage_ = nullptr;
389		self->stream_ = nullptr;
390		}
391		}
392	4x	} g{this};
393
394	4x	storage_ = ::operator new(sizeof(S));
395	4x	stream_ = ::new(storage_) S(std::move(s));
396
397		// Preallocate the awaitable storage
398	3x	cached_awaitable_ = ::operator new(vt_->awaitable_size);
399
400	3x	g.committed = true;
401	4x	}
402
403		template<WriteStream S>
404	105x	any_write_stream::any_write_stream(S* s)
405	105x	: stream_(s)
406	105x	, vt_(&vtable_for_impl<S>::value)
407		{
408		// Preallocate the awaitable storage
409	105x	cached_awaitable_ = ::operator new(vt_->awaitable_size);
410	105x	}
411
412		template<ConstBufferSequence CB>
413		auto
414	91x	any_write_stream::write_some(CB buffers)
415		{
416		struct awaitable
417		{
418		any_write_stream* self_;
419		detail::const_buffer_array<detail::max_iovec_> ba_;
420		// Set while this awaitable, not another on the same stream,
421		// holds the stream's inner awaitable.
422		bool owns_ = false;
423
424		// A chain unwound while suspended here never reaches
425		// await_resume, so the inner awaitable is destroyed here.
426	178x	~awaitable()
427		{
428	178x	if(owns_)
429		{
430	3x	self_->vt_->destroy_awaitable(self_->cached_awaitable_);
431	3x	self_->awaitable_active_ = false;
432	3x	self_->owner_ = nullptr;
433		}
434	178x	}
435
436	91x	awaitable(
437		any_write_stream* self,
438		CB const& buffers) noexcept
439	91x	: self_(self)
440	91x	, ba_(buffers)
441		{
442	91x	}
443
444	87x	awaitable(awaitable&& other) noexcept
445	87x	: self_(std::exchange(other.self_, nullptr))
446	87x	, ba_(other.ba_)
447	87x	, owns_(std::exchange(other.owns_, false))
448		{
449	87x	if(owns_)
450	1x	self_->owner_ = &owns_;
451	87x	}
452
453		bool
454	90x	await_ready()
455		{
456		// An empty write never touches the underlying stream
457	90x	if(ba_.to_span().empty())
458	4x	return true;
459
460	86x	self_->vt_->construct_awaitable(
461	86x	self_->stream_,
462	86x	self_->cached_awaitable_,
463	86x	ba_.to_span());
464	86x	self_->awaitable_active_ = true;
465	86x	owns_ = true;
466	86x	self_->owner_ = &owns_;
467
468	172x	return self_->vt_->await_ready(
469	86x	self_->cached_awaitable_);
470		}
471
472		std::coroutine_handle<>
473	74x	await_suspend(std::coroutine_handle<> h, await_context const* env)
474		{
475	74x	return self_->vt_->await_suspend(
476	74x	self_->cached_awaitable_, h, env);
477		}
478
479		[[nodiscard]] results<std::size_t>
480	81x	await_resume()
481		{
482	81x	if(!owns_)
483	4x	return {std::error_code(), 0};
484		struct guard {
485		awaitable* a;
486	77x	~guard() {
487	77x	a->self_->vt_->destroy_awaitable(
488	77x	a->self_->cached_awaitable_);
489	77x	a->self_->awaitable_active_ = false;
490	77x	a->self_->owner_ = nullptr;
491	77x	a->owns_ = false;
492	77x	}
493	77x	} g{this};
494	77x	return self_->vt_->await_resume(
495	77x	self_->cached_awaitable_);
496	77x	}
497		};
498	91x	return awaitable{this, buffers};
499		}
500
501		} // namespace capy
502		} // namespace boost
503
504		#endif
505