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