include/boost/capy/when_all.hpp

100.0% Lines (161/1/162) 95.4% List of functions (513/3/541)

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

when_all.hpp

f(x) Functions (541)

Function Calls Lines Blocks

Line	Hits	Source Code
1		//
2		// Copyright (c) 2026 Steve Gerbino
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_WHEN_ALL_HPP
12		#define BOOST_CAPY_WHEN_ALL_HPP
13
14		#include <boost/capy/detail/config.hpp>
15		#include <boost/capy/detail/io_result_combinators.hpp>
16		#include <boost/capy/continuation.hpp>
17		#include <boost/capy/concept/executor.hpp>
18		#include <boost/capy/concept/io_awaitable.hpp>
19		#include <coroutine>
20		#include <boost/capy/ex/frame_alloc_mixin.hpp>
21		#include <boost/capy/ex/io_env.hpp>
22		#include <boost/capy/ex/frame_allocator.hpp>
23		#include <boost/capy/task.hpp>
24
25		#include <array>
26		#include <atomic>
27		#include <exception>
28		#include <memory>
29		#include <optional>
30		#include <ranges>
31		#include <stdexcept>
32		#include <stop_token>
33		#include <tuple>
34		#include <type_traits>
35		#include <utility>
36		#include <vector>
37
38		namespace boost {
39		namespace capy {
40
41		namespace detail {
42
43		/** Holds the result of a single task within when_all.
44		*/
45		template<typename T>
46		struct result_holder
47		{
48		std::optional<T> value_;
49
50	119x	void set(T v)
51		{
52	119x	value_ = std::move(v);
53	119x	}
54
55	105x	T get() &&
56		{
57	105x	return std::move(*value_);
58		}
59		};
60
61		/** Core shared state for when_all operations.
62
63		Contains all members and methods common to both heterogeneous (variadic)
64		and homogeneous (range) when_all implementations. State classes embed
65		this via composition to avoid CRTP destructor ordering issues.
66
67		@par Thread Safety
68		Atomic operations protect exception capture and completion count.
69		*/
70		struct when_all_core
71		{
72		std::atomic<std::size_t> remaining_count_;
73
74		// Exception storage - first error wins, others discarded
75		std::atomic<bool> has_exception_{false};
76		std::exception_ptr first_exception_;
77
78		std::stop_source stop_source_;
79
80		// Bridges parent's stop token to our stop_source
81		struct stop_callback_fn
82		{
83		std::stop_source* source_;
84	3x	void operator()() const { source_->request_stop(); }
85		};
86		using stop_callback_t = std::stop_callback<stop_callback_fn>;
87		std::optional<stop_callback_t> parent_stop_callback_;
88
89		continuation continuation_;
90		io_env const* caller_env_ = nullptr;
91
92	82x	explicit when_all_core(std::size_t count) noexcept
93	82x	: remaining_count_(count)
94		{
95	82x	}
96
97		/** Capture an exception (first one wins). */
98	21x	void capture_exception(std::exception_ptr ep)
99		{
100	21x	bool expected = false;
101	21x	if(has_exception_.compare_exchange_strong(
102		expected, true, std::memory_order_relaxed))
103	19x	first_exception_ = ep;
104	21x	}
105		};
106
107		/** Shared state for heterogeneous when_all (variadic overload).
108
109		@tparam Ts The result types of the tasks.
110		*/
111		template<typename... Ts>
112		struct when_all_state
113		{
114		static constexpr std::size_t task_count = sizeof...(Ts);
115
116		when_all_core core_;
117		std::tuple<result_holder<Ts>...> results_;
118		std::array<continuation, task_count> runner_handles_{};
119
120		std::atomic<bool> has_error_{false};
121		std::error_code first_error_;
122
123	66x	when_all_state()
124	66x	: core_(task_count)
125		{
126	66x	}
127
128		/** Record the first error (subsequent errors are discarded). */
129	46x	void record_error(std::error_code ec)
130		{
131	46x	bool expected = false;
132	46x	if(has_error_.compare_exchange_strong(
133		expected, true, std::memory_order_relaxed))
134	32x	first_error_ = ec;
135	46x	}
136		};
137
138		/** Shared state for homogeneous when_all (range overload).
139
140		Stores extracted io_result payloads in a vector indexed by task
141		position. Tracks the first error_code for error propagation.
142
143		@tparam T The payload type extracted from io_result.
144		*/
145		template<typename T>
146		struct when_all_homogeneous_state
147		{
148		when_all_core core_;
149		std::vector<std::optional<T>> results_;
150		std::unique_ptr<continuation[]> runner_handles_;
151
152		std::atomic<bool> has_error_{false};
153		std::error_code first_error_;
154
155	13x	explicit when_all_homogeneous_state(std::size_t count)
156	13x	: core_(count)
157	26x	, results_(count)
158	13x	, runner_handles_(std::make_unique<continuation[]>(count))
159		{
160	13x	}
161
162	21x	void set_result(std::size_t index, T value)
163		{
164	21x	results_[index].emplace(std::move(value));
165	21x	}
166
167		/** Record the first error (subsequent errors are discarded). */
168	7x	void record_error(std::error_code ec)
169		{
170	7x	bool expected = false;
171	7x	if(has_error_.compare_exchange_strong(
172		expected, true, std::memory_order_relaxed))
173	5x	first_error_ = ec;
174	7x	}
175		};
176
177		/** Specialization for void io_result children (no payload storage). */
178		template<>
179		struct when_all_homogeneous_state<std::tuple<>>
180		{
181		when_all_core core_;
182		std::unique_ptr<continuation[]> runner_handles_;
183
184		std::atomic<bool> has_error_{false};
185		std::error_code first_error_;
186
187	3x	explicit when_all_homogeneous_state(std::size_t count)
188	3x	: core_(count)
189	3x	, runner_handles_(std::make_unique<continuation[]>(count))
190		{
191	3x	}
192
193		/** Record the first error (subsequent errors are discarded). */
194	1x	void record_error(std::error_code ec)
195		{
196	1x	bool expected = false;
197	1x	if(has_error_.compare_exchange_strong(
198		expected, true, std::memory_order_relaxed))
199	1x	first_error_ = ec;
200	1x	}
201		};
202
203		/** Wrapper coroutine that intercepts task completion for when_all.
204
205		Parameterized on StateType to work with both heterogeneous (variadic)
206		and homogeneous (range) state types. All state types expose their
207		shared members through a `core_` member of type when_all_core.
208
209		@tparam StateType The state type (when_all_state or when_all_homogeneous_state).
210		*/
211		template<typename StateType>
212		struct BOOST_CAPY_CORO_DESTROY_WHEN_COMPLETE when_all_runner
213		{
214		struct promise_type
215		: frame_alloc_mixin
216		{
217		StateType* state_ = nullptr;
218		std::size_t index_ = 0;
219		io_env env_;
220
221	174x	when_all_runner get_return_object() noexcept
222		{
223		return when_all_runner(
224	174x	std::coroutine_handle<promise_type>::from_promise(*this));
225		}
226
227	174x	std::suspend_always initial_suspend() noexcept
228		{
229	174x	return {};
230		}
231
232	174x	auto final_suspend() noexcept
233		{
234		struct awaiter
235		{
236		promise_type* p_;
237	58x	bool await_ready() const noexcept { return false; }
238	58x	auto await_suspend(std::coroutine_handle<> h) noexcept
239		{
240	58x	auto& core = p_->state_->core_;
241	58x	auto* counter = &core.remaining_count_;
242	58x	auto* caller_env = core.caller_env_;
243	58x	auto& cont = core.continuation_;
244
245	58x	h.destroy();
246
247	58x	auto remaining = counter->fetch_sub(1, std::memory_order_acq_rel);
248	58x	if(remaining == 1)
249	28x	return detail::symmetric_transfer(caller_env->executor.dispatch(cont));
250	30x	return detail::symmetric_transfer(std::noop_coroutine());
251		}
252	−	void await_resume() const noexcept {} // LCOV_EXCL_LINE final_suspend awaiter, never resumed
253		};
254	174x	return awaiter{this};
255		}
256
257	153x	void return_void() noexcept {}
258
259	21x	void unhandled_exception() noexcept
260		{
261	21x	state_->core_.capture_exception(std::current_exception());
262	21x	state_->core_.stop_source_.request_stop();
263	21x	}
264
265		template<class Awaitable>
266		struct transform_awaiter
267		{
268		std::decay_t<Awaitable> a_;
269		promise_type* p_;
270
271	174x	bool await_ready() { return a_.await_ready(); }
272	174x	decltype(auto) await_resume() { return a_.await_resume(); }
273
274		template<class Promise>
275	174x	auto await_suspend(std::coroutine_handle<Promise> h)
276		{
277		using R = decltype(a_.await_suspend(h, &p_->env_));
278		if constexpr (std::is_same_v<R, std::coroutine_handle<>>)
279	174x	return detail::symmetric_transfer(a_.await_suspend(h, &p_->env_));
280		else
281		return a_.await_suspend(h, &p_->env_);
282		}
283		};
284
285		template<class Awaitable>
286	174x	auto await_transform(Awaitable&& a)
287		{
288		using A = std::decay_t<Awaitable>;
289		if constexpr (IoAwaitable<A>)
290		{
291		return transform_awaiter<Awaitable>{
292	348x	std::forward<Awaitable>(a), this};
293		}
294		else
295		{
296		static_assert(sizeof(A) == 0, "requires IoAwaitable");
297		}
298	174x	}
299		};
300
301		std::coroutine_handle<promise_type> h_;
302
303	174x	explicit when_all_runner(std::coroutine_handle<promise_type> h) noexcept
304	174x	: h_(h)
305		{
306	174x	}
307
308		// Enable move for all clang versions - some versions need it
309		when_all_runner(when_all_runner&& other) noexcept
310		: h_(std::exchange(other.h_, nullptr))
311		{
312		}
313
314		when_all_runner(when_all_runner const&) = delete;
315		when_all_runner& operator=(when_all_runner const&) = delete;
316		when_all_runner& operator=(when_all_runner&&) = delete;
317
318	174x	auto release() noexcept
319		{
320	174x	return std::exchange(h_, nullptr);
321		}
322		};
323
324		/** Create an io_result-aware runner for a single awaitable (range path).
325
326		Checks the error code, records errors and requests stop on failure,
327		or extracts the payload on success.
328		*/
329		template<IoAwaitable Awaitable, typename StateType>
330		when_all_runner<StateType>
331	37x	make_when_all_homogeneous_runner(Awaitable inner, StateType* state, std::size_t index)
332		{
333		auto result = co_await std::move(inner);
334
335		if(result.ec)
336		{
337		state->record_error(result.ec);
338		state->core_.stop_source_.request_stop();
339		}
340		else
341		{
342		using PayloadT = io_result_payload_t<
343		awaitable_result_t<Awaitable>>;
344		if constexpr (!std::is_same_v<PayloadT, std::tuple<>>)
345		{
346		state->set_result(index,
347		extract_io_payload(std::move(result)));
348		}
349		}
350	74x	}
351
352		/** Create a runner for io_result children that requests stop on ec. */
353		template<std::size_t Index, IoAwaitable Awaitable, typename... Ts>
354		when_all_runner<when_all_state<Ts...>>
355	137x	make_when_all_io_runner(Awaitable inner, when_all_state<Ts...>* state)
356		{
357		auto result = co_await std::move(inner);
358		auto ec = result.ec;
359		std::get<Index>(state->results_).set(std::move(result));
360
361		if(ec)
362		{
363		state->record_error(ec);
364		state->core_.stop_source_.request_stop();
365		}
366	274x	}
367
368		/** Launcher that uses io_result-aware runners. */
369		template<IoAwaitable... Awaitables>
370		class when_all_io_launcher
371		{
372		using state_type = when_all_state<awaitable_result_t<Awaitables>...>;
373
374		std::tuple<Awaitables...>* awaitables_;
375		state_type* state_;
376
377		public:
378	66x	when_all_io_launcher(
379		std::tuple<Awaitables...>* awaitables,
380		state_type* state)
381	66x	: awaitables_(awaitables)
382	66x	, state_(state)
383		{
384	66x	}
385
386	66x	bool await_ready() const noexcept
387		{
388	66x	return sizeof...(Awaitables) == 0;
389		}
390
391	66x	std::coroutine_handle<> await_suspend(
392		std::coroutine_handle<> continuation, io_env const* caller_env)
393		{
394	66x	state_->core_.continuation_.h = continuation;
395	66x	state_->core_.caller_env_ = caller_env;
396
397	66x	if(caller_env->stop_token.stop_possible())
398		{
399	4x	state_->core_.parent_stop_callback_.emplace(
400	2x	caller_env->stop_token,
401	2x	when_all_core::stop_callback_fn{&state_->core_.stop_source_});
402
403	2x	if(caller_env->stop_token.stop_requested())
404	1x	state_->core_.stop_source_.request_stop();
405		}
406
407	66x	auto token = state_->core_.stop_source_.get_token();
408	66x	launch_all(std::index_sequence_for<Awaitables...>{},
409		caller_env->executor, token);
410
411	132x	return std::noop_coroutine();
412	66x	}
413
414	66x	void await_resume() const noexcept {}
415
416		private:
417		template<std::size_t... Is>
418	66x	void launch_all(std::index_sequence<Is...>,
419		executor_ref ex, std::stop_token token)
420		{
421	66x	(..., launch_one<Is>(ex, token));
422	66x	}
423
424		template<std::size_t I>
425	137x	void launch_one(executor_ref caller_ex, std::stop_token token)
426		{
427	137x	auto runner = make_when_all_io_runner<I>(
428	137x	std::move(std::get<I>(*awaitables_)), state_);
429
430	137x	auto h = runner.release();
431	137x	h.promise().state_ = state_;
432	137x	h.promise().env_ = io_env{caller_ex, token,
433	137x	state_->core_.caller_env_->frame_allocator};
434
435	137x	state_->runner_handles_[I].h = std::coroutine_handle<>{h};
436	137x	state_->core_.caller_env_->executor.post(state_->runner_handles_[I]);
437	274x	}
438		};
439
440		/** Helper to extract a single result from state.
441		This is a separate function to work around a GCC-11 ICE that occurs
442		when using nested immediately-invoked lambdas with pack expansion.
443		*/
444		template<std::size_t I, typename... Ts>
445	105x	auto extract_single_result(when_all_state<Ts...>& state)
446		{
447	105x	return std::move(std::get<I>(state.results_)).get();
448		}
449
450		/** Extract all results from state as a tuple.
451		*/
452		template<typename... Ts>
453	50x	auto extract_results(when_all_state<Ts...>& state)
454		{
455	50x	return [&]<std::size_t... Is>(std::index_sequence<Is...>) {
456	18x	return std::tuple(extract_single_result<Is>(state)...);
457	100x	}(std::index_sequence_for<Ts...>{});
458		}
459
460		/** Starts all homogeneous runners concurrently.
461
462		Two-phase approach: create all runners first, then post all.
463		This avoids lifetime issues if a task completes synchronously.
464		*/
465		template<typename Range>
466		class when_all_homogeneous_launcher
467		{
468		using Awaitable = std::ranges::range_value_t<Range>;
469		using PayloadT = io_result_payload_t<awaitable_result_t<Awaitable>>;
470
471		Range* range_;
472		when_all_homogeneous_state<PayloadT>* state_;
473
474		public:
475	16x	when_all_homogeneous_launcher(
476		Range* range,
477		when_all_homogeneous_state<PayloadT>* state)
478	16x	: range_(range)
479	16x	, state_(state)
480		{
481	16x	}
482
483	16x	bool await_ready() const noexcept
484		{
485	16x	return std::ranges::empty(*range_);
486		}
487
488	16x	std::coroutine_handle<> await_suspend(std::coroutine_handle<> continuation, io_env const* caller_env)
489		{
490	16x	state_->core_.continuation_.h = continuation;
491	16x	state_->core_.caller_env_ = caller_env;
492
493	16x	if(caller_env->stop_token.stop_possible())
494		{
495	4x	state_->core_.parent_stop_callback_.emplace(
496	2x	caller_env->stop_token,
497	2x	when_all_core::stop_callback_fn{&state_->core_.stop_source_});
498
499	2x	if(caller_env->stop_token.stop_requested())
500	1x	state_->core_.stop_source_.request_stop();
501		}
502
503	16x	auto token = state_->core_.stop_source_.get_token();
504
505		// Phase 1: Create all runners without dispatching.
506	16x	std::size_t index = 0;
507	53x	for(auto&& a : *range_)
508		{
509	37x	auto runner = make_when_all_homogeneous_runner(
510	37x	std::move(a), state_, index);
511
512	37x	auto h = runner.release();
513	37x	h.promise().state_ = state_;
514	37x	h.promise().index_ = index;
515	37x	h.promise().env_ = io_env{caller_env->executor, token, caller_env->frame_allocator};
516
517	37x	state_->runner_handles_[index].h = std::coroutine_handle<>{h};
518	37x	++index;
519		}
520
521		// Phase 2: Post all runners. Any may complete synchronously.
522		// After last post, state_ and this may be destroyed.
523	16x	auto* handles = state_->runner_handles_.get();
524	16x	std::size_t count = state_->core_.remaining_count_.load(std::memory_order_relaxed);
525	53x	for(std::size_t i = 0; i < count; ++i)
526	37x	caller_env->executor.post(handles[i]);
527
528	32x	return std::noop_coroutine();
529	53x	}
530
531	16x	void await_resume() const noexcept
532		{
533	16x	}
534		};
535
536		} // namespace detail
537
538		/** Execute a range of io_result-returning awaitables concurrently.
539
540		Starts all awaitables simultaneously and waits for all to complete.
541		On success, extracted payloads are collected in a vector preserving
542		input order. The first error_code makes a stop request that every
543		sibling observes, and is propagated in the outer io_result.
544		Exceptions always beat error codes.
545
546		@li All child awaitables run concurrently on the caller's executor.
547		@li Payloads are returned as a vector in input order.
548		@li First error_code wins and makes a stop request that siblings observe.
549		@li Exception always beats error_code.
550		@li Completes only after all children have finished.
551
552		@par Await-effects
553
554		Takes ownership of the range, creates one wrapper coroutine per
555		element, then posts every wrapper to the caller's executor. All
556		children therefore run concurrently, each awaited with the caller's
557		executor and frame allocator and with a stop token owned by this
558		operation.
559
560		Awaiting an empty range throws `std::invalid_argument` before any
561		child is started.
562
563		A stop request is made on the operation's own stop token when:
564
565		@li a child await-returns a non-zero `ec`, or
566		@li a child exits via an exception, or
567		@li the caller's stop token is triggered.
568
569		Every sibling observes that request through the stop token it was
570		awaited with. The request does not end the operation: the await
571		completes only after every child has finished.
572
573		@par Await-returns
574		An object of type `io_result<std::vector<PayloadT>>` destructuring as
575		`[ec, values]`, where `PayloadT` is the payload of one child's
576		`io_result`.
577
578		`ec` is the first non-zero `ec` await-returned by a child, in
579		completion order rather than input order. The `ec` of every other
580		child is discarded.
581
582		On success, `values` holds one payload per element of the input
583		range, in input order. If `ec` is set, `values` is empty: the
584		payloads of the children that did succeed are discarded.
585
586		If any child exits via an exception, the first such exception is
587		rethrown instead of await-returning, even when a child also reported
588		an `ec`.
589
590		@par Await-postcondition
591		Every child has finished. `ec` is success only if every child
592		await-returned success. If `ec` is success, `values` holds one
593		payload per input awaitable; otherwise `values` is empty.
594
595		@par Remarks
596		Supports _IoAwaitable cancellation_.
597
598		@par Thread Safety
599		The returned task must be awaited from a single execution context.
600		Child awaitables execute concurrently but complete through the caller's
601		executor.
602
603		@param awaitables Range of io_result-returning awaitables to execute
604		concurrently (must not be empty).
605
606		@return A task yielding io_result<vector<PayloadT>> where PayloadT
607		is the payload extracted from each child's io_result.
608
609		@throws std::invalid_argument if range is empty (thrown before
610		coroutine suspends).
611
612		@par Exception Safety
613		If a child throws, the first child exception is rethrown after
614		all children complete (exception beats error_code).
615
616		@par Example
617		@code
618		task<void> example()
619		{
620		std::vector<io_task<size_t>> reads;
621		for (auto& buf : buffers)
622		reads.push_back(stream.read_some(buf));
623
624		auto [ec, counts] = co_await when_all(std::move(reads));
625		if (ec) { // handle error
626		}
627		}
628		@endcode
629
630		@see IoAwaitableRange, when_all
631		*/
632		template<IoAwaitableRange R>
633		requires detail::is_io_result_v<
634		awaitable_result_t<std::ranges::range_value_t<R>>>
635		&& (!std::is_same_v<
636		detail::io_result_payload_t<
637		awaitable_result_t<std::ranges::range_value_t<R>>>,
638		std::tuple<>>)
639	14x	[[nodiscard]] auto when_all(R&& awaitables)
640		-> task<io_result<std::vector<
641		detail::io_result_payload_t<
642		awaitable_result_t<std::ranges::range_value_t<R>>>>>>
643		{
644		using Awaitable = std::ranges::range_value_t<R>;
645		using PayloadT = detail::io_result_payload_t<
646		awaitable_result_t<Awaitable>>;
647		using OwnedRange = std::remove_cvref_t<R>;
648
649		auto count = std::ranges::size(awaitables);
650		if(count == 0)
651		throw std::invalid_argument("when_all requires at least one awaitable");
652
653		OwnedRange owned_awaitables = std::forward<R>(awaitables);
654
655		detail::when_all_homogeneous_state<PayloadT> state(count);
656
657		co_await detail::when_all_homogeneous_launcher<OwnedRange>(
658		&owned_awaitables, &state);
659
660		if(state.core_.first_exception_)
661		std::rethrow_exception(state.core_.first_exception_);
662
663		if(state.has_error_.load(std::memory_order_relaxed))
664		co_return io_result<std::vector<PayloadT>>{state.first_error_, {}};
665
666		std::vector<PayloadT> results;
667		results.reserve(count);
668		for(auto& opt : state.results_)
669		results.push_back(std::move(*opt));
670
671		co_return io_result<std::vector<PayloadT>>{{}, std::move(results)};
672	28x	}
673
674		/** Execute a range of void io_result-returning awaitables concurrently.
675
676		Starts all awaitables simultaneously and waits for all to complete.
677		Since all awaitables return io_result<>, no payload values are
678		collected. The first error_code makes a stop request that every
679		sibling observes, and is propagated. Exceptions always beat error
680		codes.
681
682		@par Await-effects
683
684		Takes ownership of the range, creates one wrapper coroutine per
685		element, then posts every wrapper to the caller's executor. All
686		children therefore run concurrently, each awaited with the caller's
687		executor and frame allocator and with a stop token owned by this
688		operation.
689
690		Awaiting an empty range throws `std::invalid_argument` before any
691		child is started.
692
693		A stop request is made on the operation's own stop token when:
694
695		@li a child await-returns a non-zero `ec`, or
696		@li a child exits via an exception, or
697		@li the caller's stop token is triggered.
698
699		Every sibling observes that request through the stop token it was
700		awaited with. The request does not end the operation: the await
701		completes only after every child has finished.
702
703		@par Await-returns
704		An object of type `io_result<>` destructuring as `[ec]`. The children
705		have no payloads, so nothing else is reported.
706
707		`ec` is the first non-zero `ec` await-returned by a child, in
708		completion order rather than input order. The `ec` of every other
709		child is discarded.
710
711		If any child exits via an exception, the first such exception is
712		rethrown instead of await-returning, even when a child also reported
713		an `ec`.
714
715		@par Await-postcondition
716		Every child has finished. `ec` is success only if every child
717		await-returned success.
718
719		@par Remarks
720		Supports _IoAwaitable cancellation_.
721
722		@par Thread Safety
723		The returned task must be awaited from a single execution context.
724		Child awaitables execute concurrently but complete through the caller's
725		executor.
726
727		@param awaitables Range of io_result<>-returning awaitables to
728		execute concurrently (must not be empty).
729
730		@return A task yielding io_result<> whose ec is the first child
731		error, or default-constructed on success.
732
733		@throws std::invalid_argument if range is empty.
734
735		@par Exception Safety
736		If a child throws, the first child exception is rethrown after
737		all children complete (exception beats error_code).
738
739		@par Example
740		@code
741		task<void> example()
742		{
743		std::vector<io_task<>> jobs;
744		for (int i = 0; i < n; ++i)
745		jobs.push_back(process(i));
746
747		auto [ec] = co_await when_all(std::move(jobs));
748		}
749		@endcode
750
751		@see IoAwaitableRange, when_all
752		*/
753		template<IoAwaitableRange R>
754		requires detail::is_io_result_v<
755		awaitable_result_t<std::ranges::range_value_t<R>>>
756		&& std::is_same_v<
757		detail::io_result_payload_t<
758		awaitable_result_t<std::ranges::range_value_t<R>>>,
759		std::tuple<>>
760	4x	[[nodiscard]] auto when_all(R&& awaitables) -> task<io_result<>>
761		{
762		using OwnedRange = std::remove_cvref_t<R>;
763
764		auto count = std::ranges::size(awaitables);
765		if(count == 0)
766		throw std::invalid_argument("when_all requires at least one awaitable");
767
768		OwnedRange owned_awaitables = std::forward<R>(awaitables);
769
770		detail::when_all_homogeneous_state<std::tuple<>> state(count);
771
772		co_await detail::when_all_homogeneous_launcher<OwnedRange>(
773		&owned_awaitables, &state);
774
775		if(state.core_.first_exception_)
776		std::rethrow_exception(state.core_.first_exception_);
777
778		if(state.has_error_.load(std::memory_order_relaxed))
779		co_return io_result<>{state.first_error_};
780
781		co_return io_result<>{};
782	8x	}
783
784		/** Execute io_result-returning awaitables concurrently, inspecting error codes.
785
786		Overload selected when all children return io_result<Ts...>.
787		The error_code is lifted out of each child into a single outer
788		io_result. On success all values are returned; on failure the
789		first error_code wins.
790
791		@par Await-effects
792
793		Creates and posts one wrapper coroutine per argument to the caller's
794		executor, in argument order. All children therefore run concurrently,
795		each awaited with the caller's executor and frame allocator and with
796		a stop token owned by this operation. The overload requires at least
797		one awaitable, so there is no empty case.
798
799		A stop request is made on the operation's own stop token when:
800
801		@li a child await-returns a non-zero `ec`, or
802		@li a child exits via an exception, or
803		@li the caller's stop token is triggered.
804
805		Every sibling observes that request through the stop token it was
806		awaited with. The request does not end the operation: the await
807		completes only after every child has finished.
808
809		@par Await-returns
810		An object of type `io_result<P1, ..., Pn>` destructuring as
811		`[ec, v1, ..., vn]`, where `Pi` is the payload of the i-th child's
812		`io_result`.
813
814		`ec` is the first non-zero `ec` await-returned by a child, in
815		completion order rather than argument order. The `ec` of every other
816		child is discarded.
817
818		Each `vi` is the payload the i-th child itself await-returned, even
819		when that child or a sibling reported an `ec`. A failed child
820		therefore still contributes whatever payload it produced. This
821		differs from the range overloads, which discard all payloads once any
822		child fails.
823
824		If any child exits via an exception, the first such exception is
825		rethrown instead of await-returning, even when a child also reported
826		an `ec`.
827
828		@par Await-postcondition
829		Every child has finished. Each `vi` holds the i-th child's payload,
830		and `ec` is success only if every child await-returned success.
831
832		@par Remarks
833		Supports _IoAwaitable cancellation_.
834
835		@par Thread Safety
836		The returned task must be awaited from a single execution context.
837		Child awaitables execute concurrently but complete through the caller's
838		executor.
839
840		@par Exception Safety
841		If a child throws, the first child exception is rethrown after
842		all children complete (exception beats error_code).
843
844		@param awaitables One or more awaitables each returning
845		io_result<Ts...>.
846
847		@return A task yielding io_result<R1, R2, ..., Rn> where each Ri
848		follows the payload flattening rules.
849		*/
850		template<IoAwaitable... As>
851		requires (sizeof...(As) > 0)
852		&& detail::all_io_result_awaitables<As...>
853	66x	[[nodiscard]] auto when_all(As... awaitables)
854		-> task<io_result<
855		detail::io_result_payload_t<awaitable_result_t<As>>...>>
856		{
857		using result_type = io_result<
858		detail::io_result_payload_t<awaitable_result_t<As>>...>;
859
860		detail::when_all_state<awaitable_result_t<As>...> state;
861		std::tuple<As...> awaitable_tuple(std::move(awaitables)...);
862
863		co_await detail::when_all_io_launcher<As...>(&awaitable_tuple, &state);
864
865		// Exception always wins over error_code
866		if(state.core_.first_exception_)
867		std::rethrow_exception(state.core_.first_exception_);
868
869		auto r = detail::build_when_all_io_result<result_type>(
870		detail::extract_results(state));
871		if(state.has_error_.load(std::memory_order_relaxed))
872		r.ec = state.first_error_;
873		co_return r;
874	132x	}
875
876		} // namespace capy
877		} // namespace boost
878
879		#endif
880