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