TLA Line data 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 HIT 119 : void set(T v)
51 : {
52 119 : value_ = std::move(v);
53 119 : }
54 :
55 105 : T get() &&
56 : {
57 105 : 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 3 : 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 82 : explicit when_all_core(std::size_t count) noexcept
93 82 : : remaining_count_(count)
94 : {
95 82 : }
96 :
97 : /** Capture an exception (first one wins). */
98 21 : void capture_exception(std::exception_ptr ep)
99 : {
100 21 : bool expected = false;
101 21 : if(has_exception_.compare_exchange_strong(
102 : expected, true, std::memory_order_relaxed))
103 19 : first_exception_ = ep;
104 21 : }
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 66 : when_all_state()
124 66 : : core_(task_count)
125 : {
126 66 : }
127 :
128 : /** Record the first error (subsequent errors are discarded). */
129 46 : void record_error(std::error_code ec)
130 : {
131 46 : bool expected = false;
132 46 : if(has_error_.compare_exchange_strong(
133 : expected, true, std::memory_order_relaxed))
134 32 : first_error_ = ec;
135 46 : }
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 13 : explicit when_all_homogeneous_state(std::size_t count)
156 13 : : core_(count)
157 26 : , results_(count)
158 13 : , runner_handles_(std::make_unique<continuation[]>(count))
159 : {
160 13 : }
161 :
162 21 : void set_result(std::size_t index, T value)
163 : {
164 21 : results_[index].emplace(std::move(value));
165 21 : }
166 :
167 : /** Record the first error (subsequent errors are discarded). */
168 7 : void record_error(std::error_code ec)
169 : {
170 7 : bool expected = false;
171 7 : if(has_error_.compare_exchange_strong(
172 : expected, true, std::memory_order_relaxed))
173 5 : first_error_ = ec;
174 7 : }
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 3 : explicit when_all_homogeneous_state(std::size_t count)
188 3 : : core_(count)
189 3 : , runner_handles_(std::make_unique<continuation[]>(count))
190 : {
191 3 : }
192 :
193 : /** Record the first error (subsequent errors are discarded). */
194 1 : void record_error(std::error_code ec)
195 : {
196 1 : bool expected = false;
197 1 : if(has_error_.compare_exchange_strong(
198 : expected, true, std::memory_order_relaxed))
199 1 : first_error_ = ec;
200 1 : }
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 174 : when_all_runner get_return_object() noexcept
222 : {
223 : return when_all_runner(
224 174 : std::coroutine_handle<promise_type>::from_promise(*this));
225 : }
226 :
227 174 : std::suspend_always initial_suspend() noexcept
228 : {
229 174 : return {};
230 : }
231 :
232 174 : auto final_suspend() noexcept
233 : {
234 : struct awaiter
235 : {
236 : promise_type* p_;
237 174 : bool await_ready() const noexcept { return false; }
238 174 : auto await_suspend(std::coroutine_handle<> h) noexcept
239 : {
240 174 : auto& core = p_->state_->core_;
241 174 : auto* counter = &core.remaining_count_;
242 174 : auto* caller_env = core.caller_env_;
243 174 : auto& cont = core.continuation_;
244 :
245 174 : h.destroy();
246 :
247 174 : auto remaining = counter->fetch_sub(1, std::memory_order_acq_rel);
248 174 : if(remaining == 1)
249 82 : return detail::symmetric_transfer(caller_env->executor.dispatch(cont));
250 92 : return detail::symmetric_transfer(std::noop_coroutine());
251 : }
252 : void await_resume() const noexcept {} // LCOV_EXCL_LINE final_suspend awaiter, never resumed
253 : };
254 174 : return awaiter{this};
255 : }
256 :
257 153 : void return_void() noexcept {}
258 :
259 21 : void unhandled_exception() noexcept
260 : {
261 21 : state_->core_.capture_exception(std::current_exception());
262 21 : state_->core_.stop_source_.request_stop();
263 21 : }
264 :
265 : template<class Awaitable>
266 : struct transform_awaiter
267 : {
268 : std::decay_t<Awaitable> a_;
269 : promise_type* p_;
270 :
271 174 : bool await_ready() { return a_.await_ready(); }
272 174 : decltype(auto) await_resume() { return a_.await_resume(); }
273 :
274 : template<class Promise>
275 174 : 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 174 : 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 174 : 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 348 : std::forward<Awaitable>(a), this};
293 : }
294 : else
295 : {
296 : static_assert(sizeof(A) == 0, "requires IoAwaitable");
297 : }
298 174 : }
299 : };
300 :
301 : std::coroutine_handle<promise_type> h_;
302 :
303 174 : explicit when_all_runner(std::coroutine_handle<promise_type> h) noexcept
304 174 : : h_(h)
305 : {
306 174 : }
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 174 : auto release() noexcept
319 : {
320 174 : 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 37 : 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 74 : }
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 137 : 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 274 : }
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 66 : when_all_io_launcher(
379 : std::tuple<Awaitables...>* awaitables,
380 : state_type* state)
381 66 : : awaitables_(awaitables)
382 66 : , state_(state)
383 : {
384 66 : }
385 :
386 66 : bool await_ready() const noexcept
387 : {
388 66 : return sizeof...(Awaitables) == 0;
389 : }
390 :
391 66 : std::coroutine_handle<> await_suspend(
392 : std::coroutine_handle<> continuation, io_env const* caller_env)
393 : {
394 66 : state_->core_.continuation_.h = continuation;
395 66 : state_->core_.caller_env_ = caller_env;
396 :
397 66 : if(caller_env->stop_token.stop_possible())
398 : {
399 4 : state_->core_.parent_stop_callback_.emplace(
400 2 : caller_env->stop_token,
401 2 : when_all_core::stop_callback_fn{&state_->core_.stop_source_});
402 :
403 2 : if(caller_env->stop_token.stop_requested())
404 1 : state_->core_.stop_source_.request_stop();
405 : }
406 :
407 66 : auto token = state_->core_.stop_source_.get_token();
408 66 : launch_all(std::index_sequence_for<Awaitables...>{},
409 : caller_env->executor, token);
410 :
411 132 : return std::noop_coroutine();
412 66 : }
413 :
414 66 : void await_resume() const noexcept {}
415 :
416 : private:
417 : template<std::size_t... Is>
418 66 : void launch_all(std::index_sequence<Is...>,
419 : executor_ref ex, std::stop_token token)
420 : {
421 66 : (..., launch_one<Is>(ex, token));
422 66 : }
423 :
424 : template<std::size_t I>
425 137 : void launch_one(executor_ref caller_ex, std::stop_token token)
426 : {
427 137 : auto runner = make_when_all_io_runner<I>(
428 137 : std::move(std::get<I>(*awaitables_)), state_);
429 :
430 137 : auto h = runner.release();
431 137 : h.promise().state_ = state_;
432 137 : h.promise().env_ = io_env{caller_ex, token,
433 137 : state_->core_.caller_env_->frame_allocator};
434 :
435 137 : state_->runner_handles_[I].h = std::coroutine_handle<>{h};
436 137 : state_->core_.caller_env_->executor.post(state_->runner_handles_[I]);
437 274 : }
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 105 : auto extract_single_result(when_all_state<Ts...>& state)
446 : {
447 105 : 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 50 : auto extract_results(when_all_state<Ts...>& state)
454 : {
455 82 : 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 50 : extract_single_result<Is>(state)...);
461 100 : }(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 16 : when_all_homogeneous_launcher(
480 : Range* range,
481 : when_all_homogeneous_state<PayloadT>* state)
482 16 : : range_(range)
483 16 : , state_(state)
484 : {
485 16 : }
486 :
487 16 : bool await_ready() const noexcept
488 : {
489 16 : return std::ranges::empty(*range_);
490 : }
491 :
492 16 : std::coroutine_handle<> await_suspend(std::coroutine_handle<> continuation, io_env const* caller_env)
493 : {
494 16 : state_->core_.continuation_.h = continuation;
495 16 : state_->core_.caller_env_ = caller_env;
496 :
497 16 : if(caller_env->stop_token.stop_possible())
498 : {
499 4 : state_->core_.parent_stop_callback_.emplace(
500 2 : caller_env->stop_token,
501 2 : when_all_core::stop_callback_fn{&state_->core_.stop_source_});
502 :
503 2 : if(caller_env->stop_token.stop_requested())
504 1 : state_->core_.stop_source_.request_stop();
505 : }
506 :
507 16 : auto token = state_->core_.stop_source_.get_token();
508 :
509 : // Phase 1: Create all runners without dispatching.
510 16 : std::size_t index = 0;
511 53 : for(auto&& a : *range_)
512 : {
513 37 : auto runner = make_when_all_homogeneous_runner(
514 37 : std::move(a), state_, index);
515 :
516 37 : auto h = runner.release();
517 37 : h.promise().state_ = state_;
518 37 : h.promise().index_ = index;
519 37 : h.promise().env_ = io_env{caller_env->executor, token, caller_env->frame_allocator};
520 :
521 37 : state_->runner_handles_[index].h = std::coroutine_handle<>{h};
522 37 : ++index;
523 : }
524 :
525 : // Phase 2: Post all runners. Any may complete synchronously.
526 : // After last post, state_ and this may be destroyed.
527 16 : auto* handles = state_->runner_handles_.get();
528 16 : std::size_t count = state_->core_.remaining_count_.load(std::memory_order_relaxed);
529 53 : for(std::size_t i = 0; i < count; ++i)
530 37 : caller_env->executor.post(handles[i]);
531 :
532 32 : return std::noop_coroutine();
533 53 : }
534 :
535 16 : void await_resume() const noexcept
536 : {
537 16 : }
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 : @par !example example_1
622 :
623 :
624 : @see IoAwaitableRange, when_all
625 : */
626 : template<IoAwaitableRange R>
627 : requires detail::is_io_result_v<
628 : awaitable_result_t<std::ranges::range_value_t<R>>>
629 : && (!std::is_same_v<
630 : detail::io_result_payload_t<
631 : awaitable_result_t<std::ranges::range_value_t<R>>>,
632 : std::tuple<>>)
633 14 : [[nodiscard]] auto when_all(R&& awaitables)
634 : -> task<io_result<std::vector<
635 : detail::io_result_payload_t<
636 : awaitable_result_t<std::ranges::range_value_t<R>>>>>>
637 : {
638 : using Awaitable = std::ranges::range_value_t<R>;
639 : using PayloadT = detail::io_result_payload_t<
640 : awaitable_result_t<Awaitable>>;
641 : using OwnedRange = std::remove_cvref_t<R>;
642 :
643 : auto count = std::ranges::size(awaitables);
644 : if(count == 0)
645 : throw std::invalid_argument("when_all requires at least one awaitable");
646 :
647 : OwnedRange owned_awaitables = std::forward<R>(awaitables);
648 :
649 : detail::when_all_homogeneous_state<PayloadT> state(count);
650 :
651 : co_await detail::when_all_homogeneous_launcher<OwnedRange>(
652 : &owned_awaitables, &state);
653 :
654 : if(state.core_.first_exception_)
655 : std::rethrow_exception(state.core_.first_exception_);
656 :
657 : if(state.has_error_.load(std::memory_order_relaxed))
658 : co_return io_result<std::vector<PayloadT>>{state.first_error_, {}};
659 :
660 : std::vector<PayloadT> results;
661 : results.reserve(count);
662 : for(auto& opt : state.results_)
663 : results.push_back(std::move(*opt));
664 :
665 : co_return io_result<std::vector<PayloadT>>{std::error_code(), std::move(results)};
666 28 : }
667 :
668 : /** Execute a range of void io_result-returning awaitables concurrently.
669 :
670 : Starts all awaitables simultaneously and waits for all to complete.
671 : Since all awaitables return io_result<>, no payload values are
672 : collected. The first error_code makes a stop request that every
673 : sibling observes, and is propagated. Exceptions always beat error
674 : codes.
675 :
676 : @par Await-effects
677 :
678 : Takes ownership of the range, creates one wrapper coroutine per
679 : element, then posts every wrapper to the caller's executor. All
680 : children therefore run concurrently, each awaited with the caller's
681 : executor and frame allocator and with a stop token owned by this
682 : operation.
683 :
684 : Awaiting an empty range throws `std::invalid_argument` before any
685 : child is started.
686 :
687 : A stop request is made on the operation's own stop token when:
688 :
689 : @li a child await-returns a non-zero `ec`, or
690 : @li a child exits via an exception, or
691 : @li the caller's stop token is triggered.
692 :
693 : Every sibling observes that request through the stop token it was
694 : awaited with. The request does not end the operation: the await
695 : completes only after every child has finished.
696 :
697 : @par Await-returns
698 : An object of type `io_result<>` destructuring as `[ec]`. The children
699 : have no payloads, so nothing else is reported.
700 :
701 : `ec` is the first non-zero `ec` await-returned by a child, in
702 : completion order rather than input order. The `ec` of every other
703 : child is discarded.
704 :
705 : If any child exits via an exception, the first such exception is
706 : rethrown instead of await-returning, even when a child also reported
707 : an `ec`.
708 :
709 : @par Await-postcondition
710 : Every child has finished. `ec` is success only if every child
711 : await-returned success.
712 :
713 : @par Remarks
714 : Supports _IoAwaitable cancellation_.
715 :
716 : @par Thread Safety
717 : The returned task must be awaited from a single execution context.
718 : Child awaitables execute concurrently but complete through the caller's
719 : executor.
720 :
721 : @param awaitables Range of io_result<>-returning awaitables to
722 : execute concurrently (must not be empty).
723 :
724 : @return A task yielding io_result<> whose ec is the first child
725 : error, or default-constructed on success.
726 :
727 : @throws std::invalid_argument if range is empty.
728 :
729 : @par Exception Safety
730 : If a child throws, the first child exception is rethrown after
731 : all children complete (exception beats error_code).
732 :
733 : @par Example
734 : @par !example example_2
735 :
736 :
737 : @see IoAwaitableRange, when_all
738 : */
739 : template<IoAwaitableRange R>
740 : requires detail::is_io_result_v<
741 : awaitable_result_t<std::ranges::range_value_t<R>>>
742 : && std::is_same_v<
743 : detail::io_result_payload_t<
744 : awaitable_result_t<std::ranges::range_value_t<R>>>,
745 : std::tuple<>>
746 4 : [[nodiscard]] auto when_all(R&& awaitables) -> task<io_result<>>
747 : {
748 : using OwnedRange = std::remove_cvref_t<R>;
749 :
750 : auto count = std::ranges::size(awaitables);
751 : if(count == 0)
752 : throw std::invalid_argument("when_all requires at least one awaitable");
753 :
754 : OwnedRange owned_awaitables = std::forward<R>(awaitables);
755 :
756 : detail::when_all_homogeneous_state<std::tuple<>> state(count);
757 :
758 : co_await detail::when_all_homogeneous_launcher<OwnedRange>(
759 : &owned_awaitables, &state);
760 :
761 : if(state.core_.first_exception_)
762 : std::rethrow_exception(state.core_.first_exception_);
763 :
764 : if(state.has_error_.load(std::memory_order_relaxed))
765 : co_return io_result<>{state.first_error_};
766 :
767 : co_return io_result<>{};
768 8 : }
769 :
770 : /** Execute io_result-returning awaitables concurrently, inspecting error codes.
771 :
772 : Overload selected when all children return io_result<Ts...>.
773 : The error_code is lifted out of each child into a single outer
774 : io_result. On success all values are returned; on failure the
775 : first error_code wins.
776 :
777 : @par Await-effects
778 :
779 : Creates and posts one wrapper coroutine per argument to the caller's
780 : executor, in argument order. All children therefore run concurrently,
781 : each awaited with the caller's executor and frame allocator and with
782 : a stop token owned by this operation. The overload requires at least
783 : one awaitable, so there is no empty case.
784 :
785 : A stop request is made on the operation's own stop token when:
786 :
787 : @li a child await-returns a non-zero `ec`, or
788 : @li a child exits via an exception, or
789 : @li the caller's stop token is triggered.
790 :
791 : Every sibling observes that request through the stop token it was
792 : awaited with. The request does not end the operation: the await
793 : completes only after every child has finished.
794 :
795 : @par Await-returns
796 : An object of type `io_result<P1, ..., Pn>` destructuring as
797 : `[ec, v1, ..., vn]`, where `Pi` is the payload of the i-th child's
798 : `io_result`.
799 :
800 : `ec` is the first non-zero `ec` await-returned by a child, in
801 : completion order rather than argument order. The `ec` of every other
802 : child is discarded.
803 :
804 : Each `vi` is the payload the i-th child itself await-returned, even
805 : when that child or a sibling reported an `ec`. A failed child
806 : therefore still contributes whatever payload it produced. This
807 : differs from the range overloads, which discard all payloads once any
808 : child fails.
809 :
810 : If any child exits via an exception, the first such exception is
811 : rethrown instead of await-returning, even when a child also reported
812 : an `ec`.
813 :
814 : @par Await-postcondition
815 : Every child has finished. Each `vi` holds the i-th child's payload,
816 : and `ec` is success only if every child await-returned success.
817 :
818 : @par Remarks
819 : Supports _IoAwaitable cancellation_.
820 :
821 : @par Thread Safety
822 : The returned task must be awaited from a single execution context.
823 : Child awaitables execute concurrently but complete through the caller's
824 : executor.
825 :
826 : @par Exception Safety
827 : If a child throws, the first child exception is rethrown after
828 : all children complete (exception beats error_code).
829 :
830 : @param awaitables One or more awaitables each returning
831 : io_result<Ts...>.
832 :
833 : @return A task yielding io_result<R1, R2, ..., Rn> where each Ri
834 : follows the payload flattening rules.
835 : */
836 : template<IoAwaitable... As>
837 : requires (sizeof...(As) > 0)
838 : && detail::all_io_result_awaitables<As...>
839 66 : [[nodiscard]] auto when_all(As... awaitables)
840 : -> task<io_result<
841 : detail::io_result_payload_t<awaitable_result_t<As>>...>>
842 : {
843 : using result_type = io_result<
844 : detail::io_result_payload_t<awaitable_result_t<As>>...>;
845 :
846 : detail::when_all_state<awaitable_result_t<As>...> state;
847 : std::tuple<As...> awaitable_tuple(std::move(awaitables)...);
848 :
849 : co_await detail::when_all_io_launcher<As...>(&awaitable_tuple, &state);
850 :
851 : // Exception always wins over error_code
852 : if(state.core_.first_exception_)
853 : std::rethrow_exception(state.core_.first_exception_);
854 :
855 : auto r = detail::build_when_all_io_result<result_type>(
856 : detail::extract_results(state));
857 : if(state.has_error_.load(std::memory_order_relaxed))
858 : std::get<0>(r) = state.first_error_;
859 : co_return r;
860 132 : }
861 :
862 : } // namespace capy
863 : } // namespace boost
864 :
865 : #endif
|
