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(std::get<0>(result))
336		{
337		state->record_error(std::get<0>(result));
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 = std::get<0>(result);
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		// Explicit element types: CTAD would collapse a single
457		// io_result child via the tuple copy deduction guide
458		return std::tuple<
459		decltype(extract_single_result<Is>(state))...>(
460	18x	extract_single_result<Is>(state)...);
461	100x	}(std::index_sequence_for<Ts...>{});
462		}
463
464		/** Starts all homogeneous runners concurrently.
465
466		Two-phase approach: create all runners first, then post all.
467		This avoids lifetime issues if a task completes synchronously.
468		*/
469		template<typename Range>
470		class when_all_homogeneous_launcher
471		{
472		using Awaitable = std::ranges::range_value_t<Range>;
473		using PayloadT = io_result_payload_t<awaitable_result_t<Awaitable>>;
474
475		Range* range_;
476		when_all_homogeneous_state<PayloadT>* state_;
477
478		public:
479	16x	when_all_homogeneous_launcher(
480		Range* range,
481		when_all_homogeneous_state<PayloadT>* state)
482	16x	: range_(range)
483	16x	, state_(state)
484		{
485	16x	}
486
487	16x	bool await_ready() const noexcept
488		{
489	16x	return std::ranges::empty(*range_);
490		}
491
492	16x	std::coroutine_handle<> await_suspend(std::coroutine_handle<> continuation, io_env const* caller_env)
493		{
494	16x	state_->core_.continuation_.h = continuation;
495	16x	state_->core_.caller_env_ = caller_env;
496
497	16x	if(caller_env->stop_token.stop_possible())
498		{
499	4x	state_->core_.parent_stop_callback_.emplace(
500	2x	caller_env->stop_token,
501	2x	when_all_core::stop_callback_fn{&state_->core_.stop_source_});
502
503	2x	if(caller_env->stop_token.stop_requested())
504	1x	state_->core_.stop_source_.request_stop();
505		}
506
507	16x	auto token = state_->core_.stop_source_.get_token();
508
509		// Phase 1: Create all runners without dispatching.
510	16x	std::size_t index = 0;
511	53x	for(auto&& a : *range_)
512		{
513	37x	auto runner = make_when_all_homogeneous_runner(
514	37x	std::move(a), state_, index);
515
516	37x	auto h = runner.release();
517	37x	h.promise().state_ = state_;
518	37x	h.promise().index_ = index;
519	37x	h.promise().env_ = io_env{caller_env->executor, token, caller_env->frame_allocator};
520
521	37x	state_->runner_handles_[index].h = std::coroutine_handle<>{h};
522	37x	++index;
523		}
524
525		// Phase 2: Post all runners. Any may complete synchronously.
526		// After last post, state_ and this may be destroyed.
527	16x	auto* handles = state_->runner_handles_.get();
528	16x	std::size_t count = state_->core_.remaining_count_.load(std::memory_order_relaxed);
529	53x	for(std::size_t i = 0; i < count; ++i)
530	37x	caller_env->executor.post(handles[i]);
531
532	32x	return std::noop_coroutine();
533	53x	}
534
535	16x	void await_resume() const noexcept
536		{
537	16x	}
538		};
539
540		} // namespace detail
541
542		/** Execute a range of io_result-returning awaitables concurrently.
543
544		Starts all awaitables simultaneously and waits for all to complete.
545		On success, extracted payloads are collected in a vector preserving
546		input order. The first error_code makes a stop request that every
547		sibling observes, and is propagated in the outer io_result.
548		Exceptions always beat error codes.
549
550		@li All child awaitables run concurrently on the caller's executor.
551		@li Payloads are returned as a vector in input order.
552		@li First error_code wins and makes a stop request that siblings observe.
553		@li Exception always beats error_code.
554		@li Completes only after all children have finished.
555
556		@par Await-effects
557
558		Takes ownership of the range, creates one wrapper coroutine per
559		element, then posts every wrapper to the caller's executor. All
560		children therefore run concurrently, each awaited with the caller's
561		executor and frame allocator and with a stop token owned by this
562		operation.
563
564		Awaiting an empty range throws `std::invalid_argument` before any
565		child is started.
566
567		A stop request is made on the operation's own stop token when:
568
569		@li a child await-returns a non-zero `ec`, or
570		@li a child exits via an exception, or
571		@li the caller's stop token is triggered.
572
573		Every sibling observes that request through the stop token it was
574		awaited with. The request does not end the operation: the await
575		completes only after every child has finished.
576
577		@par Await-returns
578		An object of type `io_result<std::vector<PayloadT>>` destructuring as
579		`[ec, values]`, where `PayloadT` is the payload of one child's
580		`io_result`.
581
582		`ec` is the first non-zero `ec` await-returned by a child, in
583		completion order rather than input order. The `ec` of every other
584		child is discarded.
585
586		On success, `values` holds one payload per element of the input
587		range, in input order. If `ec` is set, `values` is empty: the
588		payloads of the children that did succeed are discarded.
589
590		If any child exits via an exception, the first such exception is
591		rethrown instead of await-returning, even when a child also reported
592		an `ec`.
593
594		@par Await-postcondition
595		Every child has finished. `ec` is success only if every child
596		await-returned success. If `ec` is success, `values` holds one
597		payload per input awaitable; otherwise `values` is empty.
598
599		@par Remarks
600		Supports _IoAwaitable cancellation_.
601
602		@par Thread Safety
603		The returned task must be awaited from a single execution context.
604		Child awaitables execute concurrently but complete through the caller's
605		executor.
606
607		@param awaitables Range of io_result-returning awaitables to execute
608		concurrently (must not be empty).
609
610		@return A task yielding io_result<vector<PayloadT>> where PayloadT
611		is the payload extracted from each child's io_result.
612
613		@throws std::invalid_argument if range is empty (thrown before
614		coroutine suspends).
615
616		@par Exception Safety
617		If a child throws, the first child exception is rethrown after
618		all children complete (exception beats error_code).
619
620		@par Example
621		@code
622		task<void> example()
623		{
624		std::vector<io_task<size_t>> reads;
625		for (auto& buf : buffers)
626		reads.push_back(stream.read_some(buf));
627
628		auto [ec, counts] = co_await when_all(std::move(reads));
629		if (ec) { // handle error
630		}
631		}
632		@endcode
633
634		@see IoAwaitableRange, when_all
635		*/
636		template<IoAwaitableRange R>
637		requires detail::is_io_result_v<
638		awaitable_result_t<std::ranges::range_value_t<R>>>
639		&& (!std::is_same_v<
640		detail::io_result_payload_t<
641		awaitable_result_t<std::ranges::range_value_t<R>>>,
642		std::tuple<>>)
643	14x	[[nodiscard]] auto when_all(R&& awaitables)
644		-> task<io_result<std::vector<
645		detail::io_result_payload_t<
646		awaitable_result_t<std::ranges::range_value_t<R>>>>>>
647		{
648		using Awaitable = std::ranges::range_value_t<R>;
649		using PayloadT = detail::io_result_payload_t<
650		awaitable_result_t<Awaitable>>;
651		using OwnedRange = std::remove_cvref_t<R>;
652
653		auto count = std::ranges::size(awaitables);
654		if(count == 0)
655		throw std::invalid_argument("when_all requires at least one awaitable");
656
657		OwnedRange owned_awaitables = std::forward<R>(awaitables);
658
659		detail::when_all_homogeneous_state<PayloadT> state(count);
660
661		co_await detail::when_all_homogeneous_launcher<OwnedRange>(
662		&owned_awaitables, &state);
663
664		if(state.core_.first_exception_)
665		std::rethrow_exception(state.core_.first_exception_);
666
667		if(state.has_error_.load(std::memory_order_relaxed))
668		co_return io_result<std::vector<PayloadT>>{state.first_error_, {}};
669
670		std::vector<PayloadT> results;
671		results.reserve(count);
672		for(auto& opt : state.results_)
673		results.push_back(std::move(*opt));
674
675		co_return io_result<std::vector<PayloadT>>{std::error_code(), std::move(results)};
676	28x	}
677
678		/** Execute a range of void io_result-returning awaitables concurrently.
679
680		Starts all awaitables simultaneously and waits for all to complete.
681		Since all awaitables return io_result<>, no payload values are
682		collected. The first error_code makes a stop request that every
683		sibling observes, and is propagated. Exceptions always beat error
684		codes.
685
686		@par Await-effects
687
688		Takes ownership of the range, creates one wrapper coroutine per
689		element, then posts every wrapper to the caller's executor. All
690		children therefore run concurrently, each awaited with the caller's
691		executor and frame allocator and with a stop token owned by this
692		operation.
693
694		Awaiting an empty range throws `std::invalid_argument` before any
695		child is started.
696
697		A stop request is made on the operation's own stop token when:
698
699		@li a child await-returns a non-zero `ec`, or
700		@li a child exits via an exception, or
701		@li the caller's stop token is triggered.
702
703		Every sibling observes that request through the stop token it was
704		awaited with. The request does not end the operation: the await
705		completes only after every child has finished.
706
707		@par Await-returns
708		An object of type `io_result<>` destructuring as `[ec]`. The children
709		have no payloads, so nothing else is reported.
710
711		`ec` is the first non-zero `ec` await-returned by a child, in
712		completion order rather than input order. The `ec` of every other
713		child is discarded.
714
715		If any child exits via an exception, the first such exception is
716		rethrown instead of await-returning, even when a child also reported
717		an `ec`.
718
719		@par Await-postcondition
720		Every child has finished. `ec` is success only if every child
721		await-returned success.
722
723		@par Remarks
724		Supports _IoAwaitable cancellation_.
725
726		@par Thread Safety
727		The returned task must be awaited from a single execution context.
728		Child awaitables execute concurrently but complete through the caller's
729		executor.
730
731		@param awaitables Range of io_result<>-returning awaitables to
732		execute concurrently (must not be empty).
733
734		@return A task yielding io_result<> whose ec is the first child
735		error, or default-constructed on success.
736
737		@throws std::invalid_argument if range is empty.
738
739		@par Exception Safety
740		If a child throws, the first child exception is rethrown after
741		all children complete (exception beats error_code).
742
743		@par Example
744		@code
745		task<void> example()
746		{
747		std::vector<io_task<>> jobs;
748		for (int i = 0; i < n; ++i)
749		jobs.push_back(process(i));
750
751		auto [ec] = co_await when_all(std::move(jobs));
752		}
753		@endcode
754
755		@see IoAwaitableRange, when_all
756		*/
757		template<IoAwaitableRange R>
758		requires detail::is_io_result_v<
759		awaitable_result_t<std::ranges::range_value_t<R>>>
760		&& std::is_same_v<
761		detail::io_result_payload_t<
762		awaitable_result_t<std::ranges::range_value_t<R>>>,
763		std::tuple<>>
764	4x	[[nodiscard]] auto when_all(R&& awaitables) -> task<io_result<>>
765		{
766		using OwnedRange = std::remove_cvref_t<R>;
767
768		auto count = std::ranges::size(awaitables);
769		if(count == 0)
770		throw std::invalid_argument("when_all requires at least one awaitable");
771
772		OwnedRange owned_awaitables = std::forward<R>(awaitables);
773
774		detail::when_all_homogeneous_state<std::tuple<>> state(count);
775
776		co_await detail::when_all_homogeneous_launcher<OwnedRange>(
777		&owned_awaitables, &state);
778
779		if(state.core_.first_exception_)
780		std::rethrow_exception(state.core_.first_exception_);
781
782		if(state.has_error_.load(std::memory_order_relaxed))
783		co_return io_result<>{state.first_error_};
784
785		co_return io_result<>{};
786	8x	}
787
788		/** Execute io_result-returning awaitables concurrently, inspecting error codes.
789
790		Overload selected when all children return io_result<Ts...>.
791		The error_code is lifted out of each child into a single outer
792		io_result. On success all values are returned; on failure the
793		first error_code wins.
794
795		@par Await-effects
796
797		Creates and posts one wrapper coroutine per argument to the caller's
798		executor, in argument order. All children therefore run concurrently,
799		each awaited with the caller's executor and frame allocator and with
800		a stop token owned by this operation. The overload requires at least
801		one awaitable, so there is no empty case.
802
803		A stop request is made on the operation's own stop token when:
804
805		@li a child await-returns a non-zero `ec`, or
806		@li a child exits via an exception, or
807		@li the caller's stop token is triggered.
808
809		Every sibling observes that request through the stop token it was
810		awaited with. The request does not end the operation: the await
811		completes only after every child has finished.
812
813		@par Await-returns
814		An object of type `io_result<P1, ..., Pn>` destructuring as
815		`[ec, v1, ..., vn]`, where `Pi` is the payload of the i-th child's
816		`io_result`.
817
818		`ec` is the first non-zero `ec` await-returned by a child, in
819		completion order rather than argument order. The `ec` of every other
820		child is discarded.
821
822		Each `vi` is the payload the i-th child itself await-returned, even
823		when that child or a sibling reported an `ec`. A failed child
824		therefore still contributes whatever payload it produced. This
825		differs from the range overloads, which discard all payloads once any
826		child fails.
827
828		If any child exits via an exception, the first such exception is
829		rethrown instead of await-returning, even when a child also reported
830		an `ec`.
831
832		@par Await-postcondition
833		Every child has finished. Each `vi` holds the i-th child's payload,
834		and `ec` is success only if every child await-returned success.
835
836		@par Remarks
837		Supports _IoAwaitable cancellation_.
838
839		@par Thread Safety
840		The returned task must be awaited from a single execution context.
841		Child awaitables execute concurrently but complete through the caller's
842		executor.
843
844		@par Exception Safety
845		If a child throws, the first child exception is rethrown after
846		all children complete (exception beats error_code).
847
848		@param awaitables One or more awaitables each returning
849		io_result<Ts...>.
850
851		@return A task yielding io_result<R1, R2, ..., Rn> where each Ri
852		follows the payload flattening rules.
853		*/
854		template<IoAwaitable... As>
855		requires (sizeof...(As) > 0)
856		&& detail::all_io_result_awaitables<As...>
857	66x	[[nodiscard]] auto when_all(As... awaitables)
858		-> task<io_result<
859		detail::io_result_payload_t<awaitable_result_t<As>>...>>
860		{
861		using result_type = io_result<
862		detail::io_result_payload_t<awaitable_result_t<As>>...>;
863
864		detail::when_all_state<awaitable_result_t<As>...> state;
865		std::tuple<As...> awaitable_tuple(std::move(awaitables)...);
866
867		co_await detail::when_all_io_launcher<As...>(&awaitable_tuple, &state);
868
869		// Exception always wins over error_code
870		if(state.core_.first_exception_)
871		std::rethrow_exception(state.core_.first_exception_);
872
873		auto r = detail::build_when_all_io_result<result_type>(
874		detail::extract_results(state));
875		if(state.has_error_.load(std::memory_order_relaxed))
876		std::get<0>(r) = state.first_error_;
877		co_return r;
878	132x	}
879
880		} // namespace capy
881		} // namespace boost
882
883		#endif
884