include/boost/capy/io/any_write_sink.hpp

100.0% Lines (169/169) 83.9% List of functions (47/56)

[ ] prev/next file

any_write_sink.hpp

f(x) Functions (56)

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_SINK_HPP
11		#define BOOST_CAPY_IO_ANY_WRITE_SINK_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/detail/buffer_array.hpp>
17		#include <boost/capy/buffers/buffer_param.hpp>
18		#include <boost/capy/concept/io_awaitable.hpp>
19		#include <boost/capy/concept/write_sink.hpp>
20		#include <coroutine>
21		#include <boost/capy/ex/io_env.hpp>
22		#include <boost/capy/io_result.hpp>
23		#include <boost/capy/io_task.hpp>
24
25		#include <concepts>
26		#include <coroutine>
27		#include <cstddef>
28		#include <exception>
29		#include <new>
30		#include <span>
31		#include <stop_token>
32		#include <system_error>
33		#include <utility>
34
35		namespace boost {
36		namespace capy {
37
38		/** Type-erased wrapper for any WriteSink.
39
40		This class provides type erasure for any type satisfying the
41		@ref WriteSink concept, enabling runtime polymorphism for
42		sink write operations. It uses cached awaitable storage to achieve
43		zero steady-state allocation after construction.
44
45		The wrapper supports two construction modes:
46		- Owning: Pass by value to transfer ownership. The wrapper
47		allocates storage and owns the sink.
48		- Reference: Pass a pointer to wrap without ownership. The
49		pointed-to sink must outlive this wrapper.
50
51		@par Awaitable Preallocation
52		The constructor preallocates storage for the type-erased awaitable.
53		This reserves all virtual address space at server startup
54		so memory usage can be measured up front, rather than
55		allocating piecemeal as traffic arrives.
56
57		@par Immediate Completion
58		Operations complete immediately without suspending when the
59		buffer sequence is empty, or when the underlying sink's
60		awaitable reports readiness via `await_ready`.
61
62		@par Thread Safety
63		Not thread-safe. Concurrent operations on the same wrapper
64		are undefined behavior.
65
66		@par Example
67		@code
68		// Owning - takes ownership of the sink
69		any_write_sink ws(some_sink{args...});
70
71		// Reference - wraps without ownership
72		some_sink sink;
73		any_write_sink ws(&sink);
74
75		const_buffer buf(data, size);
76		auto [ec, n] = co_await ws.write(std::span(&buf, 1));
77		auto [ec2] = co_await ws.write_eof();
78		@endcode
79
80		@see any_write_stream, WriteSink
81		*/
82		class any_write_sink
83		{
84		struct vtable;
85		struct write_awaitable_ops;
86		struct eof_awaitable_ops;
87
88		template<WriteSink S>
89		struct vtable_for_impl;
90
91		void* sink_ = nullptr;
92		vtable const* vt_ = nullptr;
93		void* cached_awaitable_ = nullptr;
94		void* storage_ = nullptr;
95		write_awaitable_ops const* active_write_ops_ = nullptr;
96		eof_awaitable_ops const* active_eof_ops_ = nullptr;
97
98		public:
99		/** Destructor.
100
101		Destroys the owned sink (if any) and releases the cached
102		awaitable storage.
103		*/
104		~any_write_sink();
105
106		/** Construct a default instance.
107
108		Constructs an empty wrapper. Operations on a default-constructed
109		wrapper result in undefined behavior.
110		*/
111		any_write_sink() = default;
112
113		/** Non-copyable.
114
115		The awaitable cache is per-instance and cannot be shared.
116		*/
117		any_write_sink(any_write_sink const&) = delete;
118		any_write_sink& operator=(any_write_sink const&) = delete;
119
120		/** Construct by moving.
121
122		Transfers ownership of the wrapped sink (if owned) and
123		cached awaitable storage from `other`. After the move, `other` is
124		in a default-constructed state.
125
126		@param other The wrapper to move from.
127		*/
128	1x	any_write_sink(any_write_sink&& other) noexcept
129	1x	: sink_(std::exchange(other.sink_, nullptr))
130	1x	, vt_(std::exchange(other.vt_, nullptr))
131	1x	, cached_awaitable_(std::exchange(other.cached_awaitable_, nullptr))
132	1x	, storage_(std::exchange(other.storage_, nullptr))
133	1x	, active_write_ops_(std::exchange(other.active_write_ops_, nullptr))
134	1x	, active_eof_ops_(std::exchange(other.active_eof_ops_, nullptr))
135		{
136	1x	}
137
138		/** Assign by moving.
139
140		Destroys any owned sink and releases existing resources,
141		then transfers ownership from `other`.
142
143		@param other The wrapper to move from.
144		@return Reference to this wrapper.
145		*/
146		any_write_sink&
147		operator=(any_write_sink&& other) noexcept;
148
149		/** Construct by taking ownership of a WriteSink.
150
151		Allocates storage and moves the sink into this wrapper.
152		The wrapper owns the sink and will destroy it.
153
154		@param s The sink to take ownership of.
155		*/
156		template<WriteSink S>
157		requires (!std::same_as<std::decay_t<S>, any_write_sink>)
158		any_write_sink(S s);
159
160		/** Construct by wrapping a WriteSink without ownership.
161
162		Wraps the given sink by pointer. The sink must remain
163		valid for the lifetime of this wrapper.
164
165		@param s Pointer to the sink to wrap.
166		*/
167		template<WriteSink S>
168		any_write_sink(S* s);
169
170		/** Check if the wrapper contains a valid sink.
171
172		@return `true` if wrapping a sink, `false` if default-constructed
173		or moved-from.
174		*/
175		bool
176	18x	has_value() const noexcept
177		{
178	18x	return sink_ != nullptr;
179		}
180
181		/** Check if the wrapper contains a valid sink.
182
183		@return `true` if wrapping a sink, `false` if default-constructed
184		or moved-from.
185		*/
186		explicit
187	2x	operator bool() const noexcept
188		{
189	2x	return has_value();
190		}
191
192		/** Initiate a partial write operation.
193
194		Attempt to write up to `buffer_size( buffers )` bytes from
195		the provided buffer sequence. May consume less than the
196		full sequence.
197
198		@param buffers The buffer sequence containing data to write.
199
200		@return An awaitable that await-returns `(error_code,std::size_t)`.
201
202		@par Immediate Completion
203		The operation completes immediately without suspending
204		the calling coroutine when:
205		@li The buffer sequence is empty, returning `{error_code{}, 0}`.
206		@li The underlying sink's awaitable reports immediate
207		readiness via `await_ready`.
208
209		@note This is a partial operation and may not process the
210		entire buffer sequence. Use @ref write for guaranteed
211		complete transfer.
212
213		@par Preconditions
214		The wrapper must contain a valid sink (`has_value() == true`).
215		*/
216		template<ConstBufferSequence CB>
217		auto
218		write_some(CB buffers);
219
220		/** Initiate a complete write operation.
221
222		Writes data from the provided buffer sequence. The operation
223		completes when all bytes have been consumed, or an error
224		occurs. Forwards to the underlying sink's `write` operation,
225		windowed through @ref buffer_param when the sequence exceeds
226		the per-call buffer limit.
227
228		@param buffers The buffer sequence containing data to write.
229
230		@return An awaitable that await-returns `(error_code,std::size_t)`.
231
232		@par Immediate Completion
233		The operation completes immediately without suspending
234		the calling coroutine when:
235		@li The buffer sequence is empty, returning `{error_code{}, 0}`.
236		@li Every underlying `write` call completes
237		immediately (the wrapped sink reports readiness
238		via `await_ready` on each iteration).
239
240		@par Preconditions
241		The wrapper must contain a valid sink (`has_value() == true`).
242		*/
243		template<ConstBufferSequence CB>
244		io_task<std::size_t>
245		write(CB buffers);
246
247		/** Atomically write data and signal end-of-stream.
248
249		Writes all data from the buffer sequence and then signals
250		end-of-stream. The implementation decides how to partition
251		the data across calls to the underlying sink's @ref write
252		and `write_eof`. When the caller's buffer sequence is
253		non-empty, the final call to the underlying sink is always
254		`write_eof` with a non-empty buffer sequence. When the
255		caller's buffer sequence is empty, only `write_eof()` with
256		no data is called.
257
258		@param buffers The buffer sequence containing data to write.
259
260		@return An awaitable that await-returns `(error_code,std::size_t)`.
261
262		@par Immediate Completion
263		The operation completes immediately without suspending
264		the calling coroutine when:
265		@li The buffer sequence is empty. Only the @ref write_eof()
266		call is performed.
267		@li All underlying operations complete immediately (the
268		wrapped sink reports readiness via `await_ready`).
269
270		@par Preconditions
271		The wrapper must contain a valid sink (`has_value() == true`).
272		*/
273		template<ConstBufferSequence CB>
274		io_task<std::size_t>
275		write_eof(CB buffers);
276
277		/** Signal end of data.
278
279		Indicates that no more data will be written to the sink.
280		The operation completes when the sink is finalized, or
281		an error occurs.
282
283		@return An awaitable that await-returns `(error_code)`.
284
285		@par Immediate Completion
286		The operation completes immediately without suspending
287		the calling coroutine when the underlying sink's awaitable
288		reports immediate readiness via `await_ready`.
289
290		@par Preconditions
291		The wrapper must contain a valid sink (`has_value() == true`).
292		*/
293		auto
294		write_eof();
295
296		protected:
297		/** Rebind to a new sink after move.
298
299		Updates the internal pointer to reference a new sink object.
300		Used by owning wrappers after move assignment when the owned
301		object has moved to a new location.
302
303		@param new_sink The new sink to bind to. Must be the same
304		type as the original sink.
305
306		@note Terminates if called with a sink of different type
307		than the original.
308		*/
309		template<WriteSink S>
310		void
311		rebind(S& new_sink) noexcept
312		{
313		if(vt_ != &vtable_for_impl<S>::value)
314		std::terminate();
315		sink_ = &new_sink;
316		}
317
318		private:
319		auto
320		write_some_(std::span<const_buffer const> buffers);
321
322		auto
323		write_(std::span<const_buffer const> buffers);
324
325		auto
326		write_eof_buffers_(std::span<const_buffer const> buffers);
327		};
328
329		struct any_write_sink::write_awaitable_ops
330		{
331		bool (await_ready)(void);
332		std::coroutine_handle<> (await_suspend)(void, std::coroutine_handle<>, io_env const*);
333		io_result<std::size_t> (await_resume)(void);
334		void (destroy)(void) noexcept;
335		};
336
337		struct any_write_sink::eof_awaitable_ops
338		{
339		bool (await_ready)(void);
340		std::coroutine_handle<> (await_suspend)(void, std::coroutine_handle<>, io_env const*);
341		io_result<> (await_resume)(void);
342		void (destroy)(void) noexcept;
343		};
344
345		struct any_write_sink::vtable
346		{
347		write_awaitable_ops const* (*construct_write_some_awaitable)(
348		void* sink,
349		void* storage,
350		std::span<const_buffer const> buffers);
351		write_awaitable_ops const* (*construct_write_awaitable)(
352		void* sink,
353		void* storage,
354		std::span<const_buffer const> buffers);
355		write_awaitable_ops const* (*construct_write_eof_buffers_awaitable)(
356		void* sink,
357		void* storage,
358		std::span<const_buffer const> buffers);
359		eof_awaitable_ops const* (*construct_eof_awaitable)(
360		void* sink,
361		void* storage);
362		std::size_t awaitable_size;
363		std::size_t awaitable_align;
364		void (destroy)(void) noexcept;
365		};
366
367		template<WriteSink S>
368		struct any_write_sink::vtable_for_impl
369		{
370		using WriteSomeAwaitable = decltype(std::declval<S&>().write_some(
371		std::span<const_buffer const>{}));
372		using WriteAwaitable = decltype(std::declval<S&>().write(
373		std::span<const_buffer const>{}));
374		using WriteEofBuffersAwaitable = decltype(std::declval<S&>().write_eof(
375		std::span<const_buffer const>{}));
376		using EofAwaitable = decltype(std::declval<S&>().write_eof());
377
378		static void
379	8x	do_destroy_impl(void* sink) noexcept
380		{
381	8x	static_cast<S*>(sink)->~S();
382	8x	}
383
384		static write_awaitable_ops const*
385	41x	construct_write_some_awaitable_impl(
386		void* sink,
387		void* storage,
388		std::span<const_buffer const> buffers)
389		{
390	41x	auto& s = static_cast<S>(sink);
391	41x	::new(storage) WriteSomeAwaitable(s.write_some(buffers));
392
393		static constexpr write_awaitable_ops ops = {
394		+[](void* p) {
395		return static_cast<WriteSomeAwaitable*>(p)->await_ready();
396		},
397		+[](void* p, std::coroutine_handle<> h, io_env const* env) {
398		return detail::call_await_suspend(
399		static_cast<WriteSomeAwaitable*>(p), h, env);
400		},
401		+[](void* p) {
402		return static_cast<WriteSomeAwaitable*>(p)->await_resume();
403		},
404		+[](void* p) noexcept {
405		static_cast<WriteSomeAwaitable*>(p)->~WriteSomeAwaitable();
406		}
407		};
408	41x	return &ops;
409		}
410
411		static write_awaitable_ops const*
412	79x	construct_write_awaitable_impl(
413		void* sink,
414		void* storage,
415		std::span<const_buffer const> buffers)
416		{
417	79x	auto& s = static_cast<S>(sink);
418	79x	::new(storage) WriteAwaitable(s.write(buffers));
419
420		static constexpr write_awaitable_ops ops = {
421		+[](void* p) {
422		return static_cast<WriteAwaitable*>(p)->await_ready();
423		},
424		+[](void* p, std::coroutine_handle<> h, io_env const* env) {
425		return detail::call_await_suspend(
426		static_cast<WriteAwaitable*>(p), h, env);
427		},
428		+[](void* p) {
429		return static_cast<WriteAwaitable*>(p)->await_resume();
430		},
431		+[](void* p) noexcept {
432		static_cast<WriteAwaitable*>(p)->~WriteAwaitable(); // LCOV_EXCL_LINE runs via destroy tests; gcov miscounts fn-ptr thunk
433		}
434		};
435	79x	return &ops;
436		}
437
438		static write_awaitable_ops const*
439	17x	construct_write_eof_buffers_awaitable_impl(
440		void* sink,
441		void* storage,
442		std::span<const_buffer const> buffers)
443		{
444	17x	auto& s = static_cast<S>(sink);
445	17x	::new(storage) WriteEofBuffersAwaitable(s.write_eof(buffers));
446
447		static constexpr write_awaitable_ops ops = {
448		+[](void* p) {
449		return static_cast<WriteEofBuffersAwaitable*>(p)->await_ready();
450		},
451		+[](void* p, std::coroutine_handle<> h, io_env const* env) {
452		return detail::call_await_suspend(
453		static_cast<WriteEofBuffersAwaitable*>(p), h, env);
454		},
455		+[](void* p) {
456		return static_cast<WriteEofBuffersAwaitable*>(p)->await_resume();
457		},
458		+[](void* p) noexcept {
459		static_cast<WriteEofBuffersAwaitable*>(p)->~WriteEofBuffersAwaitable(); // LCOV_EXCL_LINE runs via destroy tests; gcov miscounts fn-ptr thunk
460		}
461		};
462	17x	return &ops;
463		}
464
465		static eof_awaitable_ops const*
466	19x	construct_eof_awaitable_impl(
467		void* sink,
468		void* storage)
469		{
470	19x	auto& s = static_cast<S>(sink);
471	19x	::new(storage) EofAwaitable(s.write_eof());
472
473		static constexpr eof_awaitable_ops ops = {
474		+[](void* p) {
475		return static_cast<EofAwaitable*>(p)->await_ready();
476		},
477		+[](void* p, std::coroutine_handle<> h, io_env const* env) {
478		return detail::call_await_suspend(
479		static_cast<EofAwaitable*>(p), h, env);
480		},
481		+[](void* p) {
482		return static_cast<EofAwaitable*>(p)->await_resume();
483		},
484		+[](void* p) noexcept {
485		static_cast<EofAwaitable*>(p)->~EofAwaitable();
486		}
487		};
488	19x	return &ops;
489		}
490
491		static constexpr std::size_t max4(
492		std::size_t a, std::size_t b,
493		std::size_t c, std::size_t d) noexcept
494		{
495		std::size_t ab = a > b ? a : b;
496		std::size_t cd = c > d ? c : d;
497		return ab > cd ? ab : cd;
498		}
499
500		static constexpr std::size_t max_awaitable_size =
501		max4(sizeof(WriteSomeAwaitable),
502		sizeof(WriteAwaitable),
503		sizeof(WriteEofBuffersAwaitable),
504		sizeof(EofAwaitable));
505
506		static constexpr std::size_t max_awaitable_align =
507		max4(alignof(WriteSomeAwaitable),
508		alignof(WriteAwaitable),
509		alignof(WriteEofBuffersAwaitable),
510		alignof(EofAwaitable));
511
512		static constexpr vtable value = {
513		&construct_write_some_awaitable_impl,
514		&construct_write_awaitable_impl,
515		&construct_write_eof_buffers_awaitable_impl,
516		&construct_eof_awaitable_impl,
517		max_awaitable_size,
518		max_awaitable_align,
519		&do_destroy_impl
520		};
521		};
522
523		inline
524	134x	any_write_sink::~any_write_sink()
525		{
526	134x	if(storage_)
527		{
528	7x	vt_->destroy(sink_);
529	7x	::operator delete(storage_);
530		}
531	134x	if(cached_awaitable_)
532		{
533	126x	if(active_write_ops_)
534	1x	active_write_ops_->destroy(cached_awaitable_);
535	125x	else if(active_eof_ops_)
536	1x	active_eof_ops_->destroy(cached_awaitable_);
537	126x	::operator delete(cached_awaitable_);
538		}
539	134x	}
540
541		inline any_write_sink&
542	4x	any_write_sink::operator=(any_write_sink&& other) noexcept
543		{
544	4x	if(this != &other)
545		{
546	4x	if(storage_)
547		{
548	1x	vt_->destroy(sink_);
549	1x	::operator delete(storage_);
550		}
551	4x	if(cached_awaitable_)
552		{
553	3x	if(active_write_ops_)
554	1x	active_write_ops_->destroy(cached_awaitable_);
555	2x	else if(active_eof_ops_)
556	1x	active_eof_ops_->destroy(cached_awaitable_);
557	3x	::operator delete(cached_awaitable_);
558		}
559	4x	sink_ = std::exchange(other.sink_, nullptr);
560	4x	vt_ = std::exchange(other.vt_, nullptr);
561	4x	cached_awaitable_ = std::exchange(other.cached_awaitable_, nullptr);
562	4x	storage_ = std::exchange(other.storage_, nullptr);
563	4x	active_write_ops_ = std::exchange(other.active_write_ops_, nullptr);
564	4x	active_eof_ops_ = std::exchange(other.active_eof_ops_, nullptr);
565		}
566	4x	return *this;
567		}
568
569		template<WriteSink S>
570		requires (!std::same_as<std::decay_t<S>, any_write_sink>)
571	9x	any_write_sink::any_write_sink(S s)
572	9x	: vt_(&vtable_for_impl<S>::value)
573		{
574		struct guard {
575		any_write_sink* self;
576		bool committed = false;
577		~guard() {
578		if(!committed && self->storage_) {
579		if(self->sink_)
580		self->vt_->destroy(self->sink_); // LCOV_EXCL_LINE OOM rollback: only when the cached-awaitable allocation throws
581		::operator delete(self->storage_);
582		self->storage_ = nullptr;
583		self->sink_ = nullptr;
584		}
585		}
586	9x	} g{this};
587
588	9x	storage_ = ::operator new(sizeof(S));
589	9x	sink_ = ::new(storage_) S(std::move(s));
590
591		// Preallocate the awaitable storage (sized for max of write/eof)
592	8x	cached_awaitable_ = ::operator new(vt_->awaitable_size);
593
594	8x	g.committed = true;
595	9x	}
596
597		template<WriteSink S>
598	121x	any_write_sink::any_write_sink(S* s)
599	121x	: sink_(s)
600	121x	, vt_(&vtable_for_impl<S>::value)
601		{
602		// Preallocate the awaitable storage (sized for max of write/eof)
603	121x	cached_awaitable_ = ::operator new(vt_->awaitable_size);
604	121x	}
605
606		inline auto
607		any_write_sink::write_some_(
608		std::span<const_buffer const> buffers)
609		{
610		struct awaitable
611		{
612		any_write_sink* self_;
613		std::span<const_buffer const> buffers_;
614
615		bool
616		await_ready() const noexcept
617		{
618		return false;
619		}
620
621		std::coroutine_handle<>
622		await_suspend(std::coroutine_handle<> h, io_env const* env)
623		{
624		self_->active_write_ops_ = self_->vt_->construct_write_some_awaitable(
625		self_->sink_,
626		self_->cached_awaitable_,
627		buffers_);
628
629		if(self_->active_write_ops_->await_ready(self_->cached_awaitable_))
630		return h;
631
632		return self_->active_write_ops_->await_suspend(
633		self_->cached_awaitable_, h, env);
634		}
635
636		io_result<std::size_t>
637		await_resume()
638		{
639		struct guard {
640		any_write_sink* self;
641		~guard() {
642		self->active_write_ops_->destroy(self->cached_awaitable_);
643		self->active_write_ops_ = nullptr;
644		}
645		} g{self_};
646		return self_->active_write_ops_->await_resume(
647		self_->cached_awaitable_);
648		}
649		};
650		return awaitable{this, buffers};
651		}
652
653		inline auto
654	79x	any_write_sink::write_(
655		std::span<const_buffer const> buffers)
656		{
657		struct awaitable
658		{
659		any_write_sink* self_;
660		std::span<const_buffer const> buffers_;
661
662		bool
663	79x	await_ready() const noexcept
664		{
665	79x	return false;
666		}
667
668		std::coroutine_handle<>
669	79x	await_suspend(std::coroutine_handle<> h, io_env const* env)
670		{
671	158x	self_->active_write_ops_ = self_->vt_->construct_write_awaitable(
672	79x	self_->sink_,
673	79x	self_->cached_awaitable_,
674		buffers_);
675
676	79x	if(self_->active_write_ops_->await_ready(self_->cached_awaitable_))
677	78x	return h;
678
679	1x	return self_->active_write_ops_->await_suspend(
680	1x	self_->cached_awaitable_, h, env);
681		}
682
683		io_result<std::size_t>
684	79x	await_resume()
685		{
686		struct guard {
687		any_write_sink* self;
688	79x	~guard() {
689	79x	self->active_write_ops_->destroy(self->cached_awaitable_);
690	79x	self->active_write_ops_ = nullptr;
691	79x	}
692	79x	} g{self_};
693	79x	return self_->active_write_ops_->await_resume(
694	137x	self_->cached_awaitable_);
695	79x	}
696		};
697	79x	return awaitable{this, buffers};
698		}
699
700		inline auto
701	19x	any_write_sink::write_eof()
702		{
703		struct awaitable
704		{
705		any_write_sink* self_;
706
707		bool
708	19x	await_ready() const noexcept
709		{
710	19x	return false;
711		}
712
713		std::coroutine_handle<>
714	19x	await_suspend(std::coroutine_handle<> h, io_env const* env)
715		{
716		// Construct the underlying awaitable into cached storage
717	38x	self_->active_eof_ops_ = self_->vt_->construct_eof_awaitable(
718	19x	self_->sink_,
719	19x	self_->cached_awaitable_);
720
721		// Check if underlying is immediately ready
722	19x	if(self_->active_eof_ops_->await_ready(self_->cached_awaitable_))
723	16x	return h;
724
725		// Forward to underlying awaitable
726	3x	return self_->active_eof_ops_->await_suspend(
727	3x	self_->cached_awaitable_, h, env);
728		}
729
730		io_result<>
731	17x	await_resume()
732		{
733		struct guard {
734		any_write_sink* self;
735	17x	~guard() {
736	17x	self->active_eof_ops_->destroy(self->cached_awaitable_);
737	17x	self->active_eof_ops_ = nullptr;
738	17x	}
739	17x	} g{self_};
740	17x	return self_->active_eof_ops_->await_resume(
741	29x	self_->cached_awaitable_);
742	17x	}
743		};
744	19x	return awaitable{this};
745		}
746
747		inline auto
748	17x	any_write_sink::write_eof_buffers_(
749		std::span<const_buffer const> buffers)
750		{
751		struct awaitable
752		{
753		any_write_sink* self_;
754		std::span<const_buffer const> buffers_;
755
756		bool
757	17x	await_ready() const noexcept
758		{
759	17x	return false;
760		}
761
762		std::coroutine_handle<>
763	17x	await_suspend(std::coroutine_handle<> h, io_env const* env)
764		{
765	34x	self_->active_write_ops_ =
766	34x	self_->vt_->construct_write_eof_buffers_awaitable(
767	17x	self_->sink_,
768	17x	self_->cached_awaitable_,
769		buffers_);
770
771	17x	if(self_->active_write_ops_->await_ready(self_->cached_awaitable_))
772	16x	return h;
773
774	1x	return self_->active_write_ops_->await_suspend(
775	1x	self_->cached_awaitable_, h, env);
776		}
777
778		io_result<std::size_t>
779	17x	await_resume()
780		{
781		struct guard {
782		any_write_sink* self;
783	17x	~guard() {
784	17x	self->active_write_ops_->destroy(self->cached_awaitable_);
785	17x	self->active_write_ops_ = nullptr;
786	17x	}
787	17x	} g{self_};
788	17x	return self_->active_write_ops_->await_resume(
789	29x	self_->cached_awaitable_);
790	17x	}
791		};
792	17x	return awaitable{this, buffers};
793		}
794
795		template<ConstBufferSequence CB>
796		auto
797	43x	any_write_sink::write_some(CB buffers)
798		{
799		struct awaitable
800		{
801		any_write_sink* self_;
802		detail::const_buffer_array<detail::max_iovec_> ba_;
803
804	43x	awaitable(
805		any_write_sink* self,
806		CB const& buffers)
807	43x	: self_(self)
808	43x	, ba_(buffers)
809		{
810	43x	}
811
812		bool
813	43x	await_ready() const noexcept
814		{
815	43x	return ba_.to_span().empty();
816		}
817
818		std::coroutine_handle<>
819	41x	await_suspend(std::coroutine_handle<> h, io_env const* env)
820		{
821	41x	self_->active_write_ops_ = self_->vt_->construct_write_some_awaitable(
822	41x	self_->sink_,
823	41x	self_->cached_awaitable_,
824	41x	ba_.to_span());
825
826	41x	if(self_->active_write_ops_->await_ready(self_->cached_awaitable_))
827	38x	return h;
828
829	3x	return self_->active_write_ops_->await_suspend(
830	3x	self_->cached_awaitable_, h, env);
831		}
832
833		io_result<std::size_t>
834	41x	await_resume()
835		{
836	41x	if(ba_.to_span().empty())
837	2x	return {{}, 0};
838
839		struct guard {
840		any_write_sink* self;
841	39x	~guard() {
842	39x	self->active_write_ops_->destroy(self->cached_awaitable_);
843	39x	self->active_write_ops_ = nullptr;
844	39x	}
845	39x	} g{self_};
846	39x	return self_->active_write_ops_->await_resume(
847	39x	self_->cached_awaitable_);
848	39x	}
849		};
850	43x	return awaitable{this, buffers};
851		}
852
853		template<ConstBufferSequence CB>
854		io_task<std::size_t>
855	69x	any_write_sink::write(CB buffers)
856		{
857		buffer_param<CB> bp(buffers);
858		std::size_t total = 0;
859
860		for(;;)
861		{
862		auto bufs = bp.data();
863		if(bufs.empty())
864		break;
865
866		auto [ec, n] = co_await write_(bufs);
867		total += n;
868		if(ec)
869		co_return {ec, total};
870		bp.consume(n);
871		}
872
873		co_return {{}, total};
874	138x	}
875
876		template<ConstBufferSequence CB>
877		io_task<std::size_t>
878	27x	any_write_sink::write_eof(CB buffers)
879		{
880		const_buffer_param<CB> bp(buffers);
881		std::size_t total = 0;
882
883		for(;;)
884		{
885		auto bufs = bp.data();
886		if(bufs.empty())
887		{
888		auto [ec] = co_await write_eof();
889		co_return {ec, total};
890		}
891
892		if(! bp.more())
893		{
894		// Last window — send atomically with EOF
895		auto [ec, n] = co_await write_eof_buffers_(bufs);
896		total += n;
897		co_return {ec, total};
898		}
899
900		auto [ec, n] = co_await write_(bufs);
901		total += n;
902		if(ec)
903		co_return {ec, total};
904		bp.consume(n);
905		}
906	54x	}
907
908		} // namespace capy
909		} // namespace boost
910
911		#endif
912