include/boost/capy/io/any_buffer_source.hpp

100.0% Lines (138/138) 82.6% List of functions (38/46)

[ ] prev/next file

any_buffer_source.hpp

f(x) Functions (46)

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_BUFFER_SOURCE_HPP
11		#define BOOST_CAPY_IO_ANY_BUFFER_SOURCE_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_copy.hpp>
17		#include <boost/capy/buffers/buffer_param.hpp>
18		#include <boost/capy/concept/buffer_source.hpp>
19		#include <boost/capy/concept/io_awaitable.hpp>
20		#include <boost/capy/concept/read_source.hpp>
21		#include <boost/capy/error.hpp>
22		#include <boost/capy/ex/io_env.hpp>
23		#include <boost/capy/io_result.hpp>
24		#include <boost/capy/io_task.hpp>
25
26		#include <concepts>
27		#include <coroutine>
28		#include <cstddef>
29		#include <exception>
30		#include <new>
31		#include <span>
32		#include <stop_token>
33		#include <system_error>
34		#include <utility>
35
36		namespace boost {
37		namespace capy {
38
39		/** Type-erased wrapper for any BufferSource.
40
41		This class provides type erasure for any type satisfying the
42		@ref BufferSource concept, enabling runtime polymorphism for
43		buffer pull operations. It uses cached awaitable storage to achieve
44		zero steady-state allocation after construction.
45
46		The wrapper also satisfies @ref ReadSource. When the wrapped type
47		satisfies only @ref BufferSource, the read operations are
48		synthesized using @ref pull and @ref consume with an extra
49		buffer copy. When the wrapped type satisfies both @ref BufferSource
50		and @ref ReadSource, the native read operations are forwarded
51		directly across the virtual boundary, avoiding the copy.
52
53		The wrapper supports two construction modes:
54		- Owning: Pass by value to transfer ownership. The wrapper
55		allocates storage and owns the source.
56		- Reference: Pass a pointer to wrap without ownership. The
57		pointed-to source must outlive this wrapper.
58
59		Within each mode, the vtable is populated at compile time based
60		on whether the wrapped type also satisfies @ref ReadSource:
61		- BufferSource only: @ref read_some and @ref read are
62		synthesized from @ref pull and @ref consume, incurring one
63		buffer copy per operation.
64		- BufferSource + ReadSource: All read operations are
65		forwarded natively through the type-erased boundary with
66		no extra copy.
67
68		@par Awaitable Preallocation
69		The constructor preallocates storage for the type-erased awaitable.
70		This reserves all virtual address space at server startup
71		so memory usage can be measured up front, rather than
72		allocating piecemeal as traffic arrives.
73
74		@par Thread Safety
75		Not thread-safe. Concurrent operations on the same wrapper
76		are undefined behavior.
77
78		@par Example
79		@code
80		// Owning - takes ownership of the source
81		any_buffer_source abs(some_buffer_source{args...});
82
83		// Reference - wraps without ownership
84		some_buffer_source src;
85		any_buffer_source abs(&src);
86
87		const_buffer arr[16];
88		auto [ec, bufs] = co_await abs.pull(arr);
89
90		// ReadSource interface also available
91		char buf[64];
92		auto [ec2, n] = co_await abs.read_some(mutable_buffer(buf, 64));
93		@endcode
94
95		@see any_buffer_sink, BufferSource, ReadSource
96		*/
97		class any_buffer_source
98		{
99		struct vtable;
100		struct awaitable_ops;
101		struct read_awaitable_ops;
102
103		template<BufferSource S>
104		struct vtable_for_impl;
105
106		// hot-path members first for cache locality
107		void* source_ = nullptr;
108		vtable const* vt_ = nullptr;
109		void* cached_awaitable_ = nullptr;
110		awaitable_ops const* active_ops_ = nullptr;
111		read_awaitable_ops const* active_read_ops_ = nullptr;
112		void* storage_ = nullptr;
113
114		public:
115		/** Destructor.
116
117		Destroys the owned source (if any) and releases the cached
118		awaitable storage.
119		*/
120		~any_buffer_source();
121
122		/** Construct a default instance.
123
124		Constructs an empty wrapper. Operations on a default-constructed
125		wrapper result in undefined behavior.
126		*/
127		any_buffer_source() = default;
128
129		/** Non-copyable.
130
131		The awaitable cache is per-instance and cannot be shared.
132		*/
133		any_buffer_source(any_buffer_source const&) = delete;
134		any_buffer_source& operator=(any_buffer_source const&) = delete;
135
136		/** Construct by moving.
137
138		Transfers ownership of the wrapped source (if owned) and
139		cached awaitable storage from `other`. After the move, `other` is
140		in a default-constructed state.
141
142		@param other The wrapper to move from.
143		*/
144	2x	any_buffer_source(any_buffer_source&& other) noexcept
145	2x	: source_(std::exchange(other.source_, nullptr))
146	2x	, vt_(std::exchange(other.vt_, nullptr))
147	2x	, cached_awaitable_(std::exchange(other.cached_awaitable_, nullptr))
148	2x	, active_ops_(std::exchange(other.active_ops_, nullptr))
149	2x	, active_read_ops_(std::exchange(other.active_read_ops_, nullptr))
150	2x	, storage_(std::exchange(other.storage_, nullptr))
151		{
152	2x	}
153
154		/** Assign by moving.
155
156		Destroys any owned source and releases existing resources,
157		then transfers ownership from `other`.
158
159		@param other The wrapper to move from.
160		@return Reference to this wrapper.
161		*/
162		any_buffer_source&
163		operator=(any_buffer_source&& other) noexcept;
164
165		/** Construct by taking ownership of a BufferSource.
166
167		Allocates storage and moves the source into this wrapper.
168		The wrapper owns the source and will destroy it. If `S` also
169		satisfies @ref ReadSource, native read operations are
170		forwarded through the virtual boundary.
171
172		@param s The source to take ownership of.
173		*/
174		template<BufferSource S>
175		requires (!std::same_as<std::decay_t<S>, any_buffer_source>)
176		any_buffer_source(S s);
177
178		/** Construct by wrapping a BufferSource without ownership.
179
180		Wraps the given source by pointer. The source must remain
181		valid for the lifetime of this wrapper. If `S` also
182		satisfies @ref ReadSource, native read operations are
183		forwarded through the virtual boundary.
184
185		@param s Pointer to the source to wrap.
186		*/
187		template<BufferSource S>
188		any_buffer_source(S* s);
189
190		/** Check if the wrapper contains a valid source.
191
192		@return `true` if wrapping a source, `false` if default-constructed
193		or moved-from.
194		*/
195		bool
196	19x	has_value() const noexcept
197		{
198	19x	return source_ != nullptr;
199		}
200
201		/** Check if the wrapper contains a valid source.
202
203		@return `true` if wrapping a source, `false` if default-constructed
204		or moved-from.
205		*/
206		explicit
207	2x	operator bool() const noexcept
208		{
209	2x	return has_value();
210		}
211
212		/** Consume bytes from the source.
213
214		Advances the internal read position of the underlying source
215		by the specified number of bytes. The next call to @ref pull
216		returns data starting after the consumed bytes.
217
218		@param n The number of bytes to consume. Must not exceed the
219		total size of buffers returned by the previous @ref pull.
220
221		@par Preconditions
222		The wrapper must contain a valid source (`has_value() == true`).
223		*/
224		void
225		consume(std::size_t n) noexcept;
226
227		/** Pull buffer data from the source.
228
229		Fills the provided span with buffer descriptors from the
230		underlying source. The operation completes when data is
231		available, the source is exhausted, or an error occurs.
232
233		@param dest Span of const_buffer to fill.
234
235		@return An awaitable that await-returns `(error_code,std::span<const_buffer>)`.
236		On success with data, a non-empty span of filled buffers.
237		On EOF, `ec == cond::eof` and span is empty.
238
239		@par Preconditions
240		The wrapper must contain a valid source (`has_value() == true`).
241		The caller must not call this function again after a prior
242		call returned an error.
243		*/
244		auto
245		pull(std::span<const_buffer> dest);
246
247		/** Read some data into a mutable buffer sequence.
248
249		Attempt to read up to `buffer_size( buffers )` bytes into
250		the caller's buffers. May fill less than the full sequence.
251
252		When the wrapped type provides native @ref ReadSource support,
253		the operation forwards directly. Otherwise it is synthesized
254		from @ref pull, @ref buffer_copy, and @ref consume.
255
256		@param buffers The buffer sequence to fill.
257
258		@return An awaitable that await-returns `(error_code,std::size_t)`.
259
260		@par Preconditions
261		The wrapper must contain a valid source (`has_value() == true`).
262		The caller must not call this function again after a prior
263		call returned an error (including EOF).
264
265		@see pull, consume
266		*/
267		template<MutableBufferSequence MB>
268		io_task<std::size_t>
269		read_some(MB buffers);
270
271		/** Read data into a mutable buffer sequence.
272
273		Fills the provided buffer sequence completely. When the
274		wrapped type provides native @ref ReadSource support, each
275		window is forwarded directly. Otherwise the data is
276		synthesized from @ref pull, @ref buffer_copy, and @ref consume.
277
278		@param buffers The buffer sequence to fill.
279
280		@return An awaitable that await-returns `(error_code,std::size_t)`.
281		On success, `n == buffer_size(buffers)`.
282		On EOF, `ec == error::eof` and `n` is bytes transferred.
283
284		@par Preconditions
285		The wrapper must contain a valid source (`has_value() == true`).
286		The caller must not call this function again after a prior
287		call returned an error (including EOF).
288
289		@see pull, consume
290		*/
291		template<MutableBufferSequence MB>
292		io_task<std::size_t>
293		read(MB buffers);
294
295		protected:
296		/** Rebind to a new source after move.
297
298		Updates the internal pointer to reference a new source object.
299		Used by owning wrappers after move assignment when the owned
300		object has moved to a new location.
301
302		@param new_source The new source to bind to. Must be the same
303		type as the original source.
304
305		@note Terminates if called with a source of different type
306		than the original.
307		*/
308		template<BufferSource S>
309		void
310		rebind(S& new_source) noexcept
311		{
312		if(vt_ != &vtable_for_impl<S>::value)
313		std::terminate();
314		source_ = &new_source;
315		}
316
317		private:
318		/** Forward a partial read through the vtable.
319
320		Constructs the underlying `read_some` awaitable in
321		cached storage and returns a type-erased awaitable.
322		*/
323		auto
324		read_some_(std::span<mutable_buffer const> buffers);
325
326		/** Forward a complete read through the vtable.
327
328		Constructs the underlying `read` awaitable in
329		cached storage and returns a type-erased awaitable.
330		*/
331		auto
332		read_(std::span<mutable_buffer const> buffers);
333		};
334
335		/** Type-erased ops for awaitables that await-return `io_result<std::span<const_buffer>>`. */
336		struct any_buffer_source::awaitable_ops
337		{
338		bool (await_ready)(void);
339		std::coroutine_handle<> (await_suspend)(void, std::coroutine_handle<>, io_env const*);
340		io_result<std::span<const_buffer>> (await_resume)(void);
341		void (destroy)(void) noexcept;
342		};
343
344		/** Type-erased ops for awaitables that await-return `io_result<std::size_t>`. */
345		struct any_buffer_source::read_awaitable_ops
346		{
347		bool (await_ready)(void);
348		std::coroutine_handle<> (await_suspend)(void, std::coroutine_handle<>, io_env const*);
349		io_result<std::size_t> (await_resume)(void);
350		void (destroy)(void) noexcept;
351		};
352
353		struct any_buffer_source::vtable
354		{
355		// BufferSource ops (always populated)
356		void (destroy)(void) noexcept;
357		void (do_consume)(void source, std::size_t n) noexcept;
358		std::size_t awaitable_size;
359		std::size_t awaitable_align;
360		awaitable_ops const* (*construct_awaitable)(
361		void* source,
362		void* storage,
363		std::span<const_buffer> dest);
364
365		// ReadSource forwarding (null when wrapped type is BufferSource-only)
366		read_awaitable_ops const* (*construct_read_some_awaitable)(
367		void* source,
368		void* storage,
369		std::span<mutable_buffer const> buffers);
370		read_awaitable_ops const* (*construct_read_awaitable)(
371		void* source,
372		void* storage,
373		std::span<mutable_buffer const> buffers);
374		};
375
376		template<BufferSource S>
377		struct any_buffer_source::vtable_for_impl
378		{
379		using PullAwaitable = decltype(std::declval<S&>().pull(
380		std::declval<std::span<const_buffer>>()));
381
382		static void
383	9x	do_destroy_impl(void* source) noexcept
384		{
385	9x	static_cast<S*>(source)->~S();
386	9x	}
387
388		static void
389	45x	do_consume_impl(void* source, std::size_t n) noexcept
390		{
391	45x	static_cast<S*>(source)->consume(n);
392	45x	}
393
394		static awaitable_ops const*
395	111x	construct_awaitable_impl(
396		void* source,
397		void* storage,
398		std::span<const_buffer> dest)
399		{
400	111x	auto& s = static_cast<S>(source);
401	111x	::new(storage) PullAwaitable(s.pull(dest));
402
403		static constexpr awaitable_ops ops = {
404		+[](void* p) {
405		return static_cast<PullAwaitable*>(p)->await_ready();
406		},
407		+[](void* p, std::coroutine_handle<> h, io_env const* env) {
408		return detail::call_await_suspend(
409		static_cast<PullAwaitable*>(p), h, env);
410		},
411		+[](void* p) {
412		return static_cast<PullAwaitable*>(p)->await_resume();
413		},
414		+[](void* p) noexcept {
415		static_cast<PullAwaitable*>(p)->~PullAwaitable();
416		}
417		};
418	111x	return &ops;
419		}
420
421		static read_awaitable_ops const*
422	49x	construct_read_some_awaitable_impl(
423		void* source,
424		void* storage,
425		std::span<mutable_buffer const> buffers)
426		requires ReadSource<S>
427		{
428		using Aw = decltype(std::declval<S&>().read_some(
429		std::span<mutable_buffer const>{}));
430	49x	auto& s = static_cast<S>(source);
431	49x	::new(storage) Aw(s.read_some(buffers));
432
433		static constexpr read_awaitable_ops ops = {
434		+[](void* p) {
435		return static_cast<Aw*>(p)->await_ready();
436		},
437		+[](void* p, std::coroutine_handle<> h, io_env const* env) {
438		return detail::call_await_suspend(
439		static_cast<Aw*>(p), h, env);
440		},
441		+[](void* p) {
442		return static_cast<Aw*>(p)->await_resume();
443		},
444		+[](void* p) noexcept {
445		static_cast<Aw*>(p)->~Aw();
446		}
447		};
448	49x	return &ops;
449		}
450
451		static read_awaitable_ops const*
452	19x	construct_read_awaitable_impl(
453		void* source,
454		void* storage,
455		std::span<mutable_buffer const> buffers)
456		requires ReadSource<S>
457		{
458		using Aw = decltype(std::declval<S&>().read(
459		std::span<mutable_buffer const>{}));
460	19x	auto& s = static_cast<S>(source);
461	19x	::new(storage) Aw(s.read(buffers));
462
463		static constexpr read_awaitable_ops ops = {
464		+[](void* p) {
465		return static_cast<Aw*>(p)->await_ready();
466		},
467		+[](void* p, std::coroutine_handle<> h, io_env const* env) {
468		return detail::call_await_suspend(
469		static_cast<Aw*>(p), h, env);
470		},
471		+[](void* p) {
472		return static_cast<Aw*>(p)->await_resume();
473		},
474		+[](void* p) noexcept {
475		static_cast<Aw*>(p)->~Aw();
476		}
477		};
478	19x	return &ops;
479		}
480
481		static consteval std::size_t
482		compute_max_size() noexcept
483		{
484		std::size_t s = sizeof(PullAwaitable);
485		if constexpr (ReadSource<S>)
486		{
487		using RS = decltype(std::declval<S&>().read_some(
488		std::span<mutable_buffer const>{}));
489		using R = decltype(std::declval<S&>().read(
490		std::span<mutable_buffer const>{}));
491
492		if(sizeof(RS) > s) s = sizeof(RS);
493		if(sizeof(R) > s) s = sizeof(R);
494		}
495		return s;
496		}
497
498		static consteval std::size_t
499		compute_max_align() noexcept
500		{
501		std::size_t a = alignof(PullAwaitable);
502		if constexpr (ReadSource<S>)
503		{
504		using RS = decltype(std::declval<S&>().read_some(
505		std::span<mutable_buffer const>{}));
506		using R = decltype(std::declval<S&>().read(
507		std::span<mutable_buffer const>{}));
508
509		if(alignof(RS) > a) a = alignof(RS);
510		if(alignof(R) > a) a = alignof(R);
511		}
512		return a;
513		}
514
515		static consteval vtable
516		make_vtable() noexcept
517		{
518		vtable v{};
519		v.destroy = &do_destroy_impl;
520		v.do_consume = &do_consume_impl;
521		v.awaitable_size = compute_max_size();
522		v.awaitable_align = compute_max_align();
523		v.construct_awaitable = &construct_awaitable_impl;
524		v.construct_read_some_awaitable = nullptr;
525		v.construct_read_awaitable = nullptr;
526
527		if constexpr (ReadSource<S>)
528		{
529		v.construct_read_some_awaitable =
530		&construct_read_some_awaitable_impl;
531		v.construct_read_awaitable =
532		&construct_read_awaitable_impl;
533		}
534		return v;
535		}
536
537		static constexpr vtable value = make_vtable();
538		};
539
540		inline
541	127x	any_buffer_source::~any_buffer_source()
542		{
543	127x	if(storage_)
544		{
545	8x	vt_->destroy(source_);
546	8x	::operator delete(storage_);
547		}
548	127x	if(cached_awaitable_)
549	121x	::operator delete(cached_awaitable_);
550	127x	}
551
552		inline any_buffer_source&
553	3x	any_buffer_source::operator=(any_buffer_source&& other) noexcept
554		{
555	3x	if(this != &other)
556		{
557	3x	if(storage_)
558		{
559	1x	vt_->destroy(source_);
560	1x	::operator delete(storage_);
561		}
562	3x	if(cached_awaitable_)
563	1x	::operator delete(cached_awaitable_);
564	3x	source_ = std::exchange(other.source_, nullptr);
565	3x	vt_ = std::exchange(other.vt_, nullptr);
566	3x	cached_awaitable_ = std::exchange(other.cached_awaitable_, nullptr);
567	3x	storage_ = std::exchange(other.storage_, nullptr);
568	3x	active_ops_ = std::exchange(other.active_ops_, nullptr);
569	3x	active_read_ops_ = std::exchange(other.active_read_ops_, nullptr);
570		}
571	3x	return *this;
572		}
573
574		template<BufferSource S>
575		requires (!std::same_as<std::decay_t<S>, any_buffer_source>)
576	10x	any_buffer_source::any_buffer_source(S s)
577	10x	: vt_(&vtable_for_impl<S>::value)
578		{
579		struct guard {
580		any_buffer_source* self;
581		bool committed = false;
582		~guard() {
583		if(!committed && self->storage_) {
584		if(self->source_)
585		self->vt_->destroy(self->source_); // LCOV_EXCL_LINE OOM rollback: only when the cached-awaitable allocation throws
586		::operator delete(self->storage_);
587		self->storage_ = nullptr;
588		self->source_ = nullptr;
589		}
590		}
591	10x	} g{this};
592
593	10x	storage_ = ::operator new(sizeof(S));
594	10x	source_ = ::new(storage_) S(std::move(s));
595
596	9x	cached_awaitable_ = ::operator new(vt_->awaitable_size);
597
598	9x	g.committed = true;
599	10x	}
600
601		template<BufferSource S>
602	113x	any_buffer_source::any_buffer_source(S* s)
603	113x	: source_(s)
604	113x	, vt_(&vtable_for_impl<S>::value)
605		{
606	113x	cached_awaitable_ = ::operator new(vt_->awaitable_size);
607	113x	}
608
609		inline void
610	45x	any_buffer_source::consume(std::size_t n) noexcept
611		{
612	45x	vt_->do_consume(source_, n);
613	45x	}
614
615		inline auto
616	111x	any_buffer_source::pull(std::span<const_buffer> dest)
617		{
618		struct awaitable
619		{
620		any_buffer_source* self_;
621		std::span<const_buffer> dest_;
622
623		bool
624	111x	await_ready()
625		{
626	222x	self_->active_ops_ = self_->vt_->construct_awaitable(
627	111x	self_->source_,
628	111x	self_->cached_awaitable_,
629		dest_);
630	111x	return self_->active_ops_->await_ready(self_->cached_awaitable_);
631		}
632
633		std::coroutine_handle<>
634	1x	await_suspend(std::coroutine_handle<> h, io_env const* env)
635		{
636	1x	return self_->active_ops_->await_suspend(
637	1x	self_->cached_awaitable_, h, env);
638		}
639
640		io_result<std::span<const_buffer>>
641	111x	await_resume()
642		{
643		struct guard {
644		any_buffer_source* self;
645	111x	~guard() {
646	111x	self->active_ops_->destroy(self->cached_awaitable_);
647	111x	self->active_ops_ = nullptr;
648	111x	}
649	111x	} g{self_};
650	111x	return self_->active_ops_->await_resume(
651	197x	self_->cached_awaitable_);
652	111x	}
653		};
654	111x	return awaitable{this, dest};
655		}
656
657		inline auto
658	49x	any_buffer_source::read_some_(
659		std::span<mutable_buffer const> buffers)
660		{
661		struct awaitable
662		{
663		any_buffer_source* self_;
664		std::span<mutable_buffer const> buffers_;
665
666		bool
667	49x	await_ready() const noexcept
668		{
669	49x	return false;
670		}
671
672		std::coroutine_handle<>
673	49x	await_suspend(std::coroutine_handle<> h, io_env const* env)
674		{
675	98x	self_->active_read_ops_ =
676	98x	self_->vt_->construct_read_some_awaitable(
677	49x	self_->source_,
678	49x	self_->cached_awaitable_,
679		buffers_);
680
681	49x	if(self_->active_read_ops_->await_ready(
682	49x	self_->cached_awaitable_))
683	48x	return h;
684
685	1x	return self_->active_read_ops_->await_suspend(
686	1x	self_->cached_awaitable_, h, env);
687		}
688
689		io_result<std::size_t>
690	49x	await_resume()
691		{
692		struct guard {
693		any_buffer_source* self;
694	49x	~guard() {
695	49x	self->active_read_ops_->destroy(
696	49x	self->cached_awaitable_);
697	49x	self->active_read_ops_ = nullptr;
698	49x	}
699	49x	} g{self_};
700	49x	return self_->active_read_ops_->await_resume(
701	90x	self_->cached_awaitable_);
702	49x	}
703		};
704	49x	return awaitable{this, buffers};
705		}
706
707		inline auto
708	19x	any_buffer_source::read_(
709		std::span<mutable_buffer const> buffers)
710		{
711		struct awaitable
712		{
713		any_buffer_source* self_;
714		std::span<mutable_buffer const> buffers_;
715
716		bool
717	19x	await_ready() const noexcept
718		{
719	19x	return false;
720		}
721
722		std::coroutine_handle<>
723	19x	await_suspend(std::coroutine_handle<> h, io_env const* env)
724		{
725	38x	self_->active_read_ops_ =
726	38x	self_->vt_->construct_read_awaitable(
727	19x	self_->source_,
728	19x	self_->cached_awaitable_,
729		buffers_);
730
731	19x	if(self_->active_read_ops_->await_ready(
732	19x	self_->cached_awaitable_))
733	18x	return h;
734
735	1x	return self_->active_read_ops_->await_suspend(
736	1x	self_->cached_awaitable_, h, env);
737		}
738
739		io_result<std::size_t>
740	19x	await_resume()
741		{
742		struct guard {
743		any_buffer_source* self;
744	19x	~guard() {
745	19x	self->active_read_ops_->destroy(
746	19x	self->cached_awaitable_);
747	19x	self->active_read_ops_ = nullptr;
748	19x	}
749	19x	} g{self_};
750	19x	return self_->active_read_ops_->await_resume(
751	32x	self_->cached_awaitable_);
752	19x	}
753		};
754	19x	return awaitable{this, buffers};
755		}
756
757		template<MutableBufferSequence MB>
758		io_task<std::size_t>
759	59x	any_buffer_source::read_some(MB buffers)
760		{
761		buffer_param<MB> bp(buffers);
762		auto dest = bp.data();
763		if(dest.empty())
764		co_return {{}, 0};
765
766		// Native ReadSource path
767		if(vt_->construct_read_some_awaitable)
768		co_return co_await read_some_(dest);
769
770		// Synthesized path: pull + buffer_copy + consume
771		const_buffer arr[detail::max_iovec_];
772		auto [ec, bufs] = co_await pull(arr);
773		if(ec)
774		co_return {ec, 0};
775
776		auto n = buffer_copy(dest, bufs);
777		consume(n);
778		co_return {{}, n};
779	118x	}
780
781		template<MutableBufferSequence MB>
782		io_task<std::size_t>
783	25x	any_buffer_source::read(MB buffers)
784		{
785		buffer_param<MB> bp(buffers);
786		std::size_t total = 0;
787
788		// Native ReadSource path
789		if(vt_->construct_read_awaitable)
790		{
791		for(;;)
792		{
793		auto dest = bp.data();
794		if(dest.empty())
795		break;
796
797		auto [ec, n] = co_await read_(dest);
798		total += n;
799		if(ec)
800		co_return {ec, total};
801		bp.consume(n);
802		}
803		co_return {{}, total};
804		}
805
806		// Synthesized path: pull + buffer_copy + consume
807		for(;;)
808		{
809		auto dest = bp.data();
810		if(dest.empty())
811		break;
812
813		const_buffer arr[detail::max_iovec_];
814		auto [ec, bufs] = co_await pull(arr);
815
816		if(ec)
817		co_return {ec, total};
818
819		auto n = buffer_copy(dest, bufs);
820		consume(n);
821		total += n;
822		bp.consume(n);
823		}
824
825		co_return {{}, total};
826	50x	}
827
828		static_assert(BufferSource<any_buffer_source>);
829		static_assert(ReadSource<any_buffer_source>);
830
831		} // namespace capy
832		} // namespace boost
833
834		#endif
835