include/boost/capy/when_all.hpp

88.3% Lines (1029/1166) 95.0% List of functions (344/362)

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

f(x) Functions (362)

Function Calls Lines Blocks

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