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