TLA Line data Source code
1 : //
2 : // Copyright (c) 2026 Michael Vandeberg
3 : // Copyright (c) 2026 Steve Gerbino
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_ANY_HPP
12 : #define BOOST_CAPY_WHEN_ANY_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/ex/executor_ref.hpp>
21 : #include <boost/capy/detail/abandonable.hpp>
22 : #include <boost/capy/detail/owned_range.hpp>
23 : #include <boost/capy/ex/abandon.hpp>
24 : #include <boost/capy/ex/frame_alloc_mixin.hpp>
25 : #include <boost/capy/ex/frame_allocator.hpp>
26 : #include <boost/capy/ex/await_context.hpp>
27 : #include <boost/capy/task.hpp>
28 :
29 : #include <array>
30 : #include <atomic>
31 : #include <exception>
32 : #include <memory>
33 : #include <mutex>
34 : #include <optional>
35 : #include <ranges>
36 : #include <stdexcept>
37 : #include <stop_token>
38 : #include <tuple>
39 : #include <type_traits>
40 : #include <utility>
41 : #include <variant>
42 : #include <vector>
43 :
44 : /*
45 : when_any - Race multiple results tasks, select first success
46 : =============================================================
47 :
48 : OVERVIEW:
49 : ---------
50 : when_any launches N results-returning tasks concurrently. A task
51 : wins by returning !ec; errors and exceptions do not win. Once a
52 : winner is found, stop is requested for siblings and the winner's
53 : payload is returned. If no winner exists (all fail), one of the
54 : failures is surfaced (an error_code at variant index 0, or a child's
55 : exception rethrown); which one is unspecified.
56 :
57 : ARCHITECTURE:
58 : -------------
59 : The design mirrors when_all but with inverted completion semantics:
60 :
61 : when_all: complete when remaining_count reaches 0 (all done)
62 : when_any: complete when has_winner becomes true (first done)
63 : BUT still wait for remaining_count to reach 0 for cleanup
64 :
65 : Key components:
66 : - when_any_core: Shared state tracking winner and completion
67 : - when_any_results_runner: Wrapper coroutine for each child task
68 : - when_any_results_launcher/when_any_results_homogeneous_launcher:
69 : Awaitables that start all runners concurrently
70 :
71 : CRITICAL INVARIANTS:
72 : --------------------
73 : 1. Only a task returning !ec can become the winner (via atomic CAS)
74 : 2. All tasks must complete before parent resumes (cleanup safety)
75 : 3. Stop is requested immediately when winner is determined
76 : 4. Exceptions and errors do not claim winner status
77 :
78 : POSITIONAL VARIANT:
79 : -------------------
80 : The variadic overload returns std::variant<error_code, R1, R2, ..., Rn>.
81 : Index 0 is error_code (failure/no-winner). Index 1..N identifies the
82 : winning child and carries its payload.
83 :
84 : RANGE OVERLOAD:
85 : ---------------
86 : The range overload returns variant<error_code, pair<size_t, T>> for
87 : non-void children or variant<error_code, size_t> for void children.
88 :
89 : MEMORY MODEL:
90 : -------------
91 : Synchronization chain from winner's write to parent's read:
92 :
93 : 1. Winner thread writes result_ (non-atomic)
94 : 2. Winner thread calls signal_completion() -> fetch_sub(acq_rel) on remaining_count_
95 : 3. Last task thread (may be winner or non-winner) calls signal_completion()
96 : -> fetch_sub(acq_rel) on remaining_count_, observing count becomes 0
97 : 4. Last task returns caller_ex_.dispatch(continuation_) via symmetric transfer
98 : 5. Parent coroutine resumes and reads result_
99 :
100 : Synchronization analysis:
101 : - All fetch_sub operations on remaining_count_ form a release sequence
102 : - Winner's fetch_sub releases; subsequent fetch_sub operations participate
103 : in the modification order of remaining_count_
104 : - Last task's fetch_sub(acq_rel) synchronizes-with prior releases in the
105 : modification order, establishing happens-before from winner's writes
106 : - Executor dispatch() is expected to provide queue-based synchronization
107 : (release-on-post, acquire-on-execute) completing the chain to parent
108 : - Even inline executors work (same thread = sequenced-before)
109 :
110 : EXCEPTION SEMANTICS:
111 : --------------------
112 : Exceptions do NOT claim winner status. If a child throws, the exception
113 : is recorded but the combinator keeps waiting for a success. Only when
114 : all children complete without a winner is a failure surfaced. There is
115 : no priority between errors and exceptions, and no guarantee about which
116 : child's failure is reported: the result either returns an error_code at
117 : variant index 0 or rethrows a child's exception.
118 : */
119 :
120 : namespace boost {
121 : namespace capy {
122 :
123 : namespace detail {
124 :
125 : /** Core shared state for when_any operations.
126 :
127 : Contains all members and methods common to both heterogeneous (variadic)
128 : and homogeneous (range) when_any implementations. State classes embed
129 : this via composition to avoid CRTP destructor ordering issues.
130 :
131 : @par Thread Safety
132 : Atomic operations protect winner selection and completion count.
133 : */
134 : struct when_any_core
135 : {
136 : std::atomic<std::size_t> remaining_count_;
137 : std::size_t winner_index_{0};
138 : std::exception_ptr winner_exception_;
139 : std::stop_source stop_source_;
140 :
141 : // Bridges parent's stop token to our stop_source
142 : struct stop_callback_fn
143 : {
144 : std::stop_source* source_;
145 HIT 4 : void operator()() const noexcept
146 : {
147 : // Stopping can lose the last branch and destroy the join,
148 : // with its source, before the request returns; the copy
149 : // keeps the stop state alive until then.
150 4 : std::stop_source keep = *source_;
151 4 : keep.request_stop();
152 4 : }
153 : };
154 : using stop_callback_t = std::stop_callback<stop_callback_fn>;
155 : std::optional<stop_callback_t> parent_stop_callback_;
156 :
157 : continuation continuation_;
158 : await_context const* caller_env_ = nullptr;
159 :
160 : // Set when a branch is abandoned before finishing: the parent is
161 : // then abandoned rather than resumed.
162 : std::atomic<bool> lost_{false};
163 :
164 : /// Drop a share held by a runner that never finished.
165 8 : void lose_branch() noexcept
166 : {
167 8 : lost_.store(true, std::memory_order_relaxed);
168 8 : stop_source_.request_stop();
169 16 : if(remaining_count_.fetch_sub(1, std::memory_order_acq_rel) == 1)
170 : {
171 4 : auto parent = continuation_;
172 4 : capy::abandon(parent);
173 : }
174 8 : }
175 :
176 : // Placed last to avoid padding (1-byte atomic followed by 8-byte aligned members)
177 : std::atomic<bool> has_winner_{false};
178 :
179 53 : explicit when_any_core(std::size_t count) noexcept
180 53 : : remaining_count_(count)
181 : {
182 53 : }
183 :
184 : /** Atomically claim winner status; exactly one task succeeds. */
185 68 : bool try_win(std::size_t index) noexcept
186 : {
187 68 : bool expected = false;
188 68 : if(has_winner_.compare_exchange_strong(
189 : expected, true, std::memory_order_acq_rel))
190 : {
191 32 : winner_index_ = index;
192 32 : stop_source_.request_stop();
193 32 : return true;
194 : }
195 36 : return false;
196 : }
197 :
198 : /** @pre try_win() returned true. */
199 1 : void set_winner_exception(std::exception_ptr ep) noexcept
200 : {
201 1 : winner_exception_ = ep;
202 1 : }
203 :
204 : // Runners signal completion directly via final_suspend; no member function needed.
205 : };
206 :
207 : } // namespace detail
208 :
209 : namespace detail {
210 :
211 : // State for `results`-aware when_any: only !ec wins.
212 : template<typename... Ts>
213 : struct when_any_results_state
214 : {
215 : static constexpr std::size_t task_count = sizeof...(Ts);
216 : using variant_type = std::variant<std::error_code, Ts...>;
217 :
218 : when_any_core core_;
219 : std::optional<variant_type> result_;
220 : std::array<continuation, task_count> runner_handles_{};
221 :
222 : // A failure (error or exception) for the all-fail case. record_error
223 : // and record_exception overwrite each other, so which one survives is
224 : // unspecified (no priority between errors and exceptions).
225 : std::mutex failure_mu_;
226 : std::error_code last_error_;
227 : std::exception_ptr last_exception_;
228 :
229 33 : when_any_results_state()
230 33 : : core_(task_count)
231 : {
232 33 : }
233 :
234 16 : void record_error(std::error_code ec)
235 : {
236 16 : std::lock_guard lk(failure_mu_);
237 16 : last_error_ = ec;
238 16 : last_exception_ = nullptr;
239 16 : }
240 :
241 7 : void record_exception(std::exception_ptr ep)
242 : {
243 7 : std::lock_guard lk(failure_mu_);
244 7 : last_exception_ = ep;
245 7 : last_error_ = {};
246 7 : }
247 : };
248 :
249 : // Wrapper coroutine for `results`-aware when_any children.
250 : // unhandled_exception records the exception but does NOT claim winner status.
251 : template<typename StateType>
252 : struct when_any_results_runner
253 : {
254 : struct promise_type
255 : : detail::process_frame_mixin
256 : {
257 : StateType* state_ = nullptr;
258 : std::size_t index_ = 0;
259 : await_context env_;
260 : // Owns the branch's segment.
261 : chain_owner segment_;
262 : bool done_ = false;
263 :
264 126 : ~promise_type()
265 : {
266 126 : if(!done_ && state_)
267 8 : state_->core_.lose_branch();
268 126 : }
269 :
270 126 : when_any_results_runner get_return_object() noexcept
271 : {
272 : return when_any_results_runner(
273 126 : std::coroutine_handle<promise_type>::from_promise(*this));
274 : }
275 :
276 126 : std::suspend_always initial_suspend() noexcept { return {}; }
277 :
278 118 : auto final_suspend() noexcept
279 : {
280 : struct awaiter
281 : {
282 : promise_type* p_;
283 118 : bool await_ready() const noexcept { return false; }
284 118 : auto await_suspend(std::coroutine_handle<> h) noexcept
285 : {
286 118 : auto& core = p_->state_->core_;
287 118 : auto* counter = &core.remaining_count_;
288 118 : auto* lost = &core.lost_;
289 118 : auto* caller_env = core.caller_env_;
290 118 : auto& cont = core.continuation_;
291 :
292 118 : p_->done_ = true;
293 : // Resumed to unwind its abandoned branch: lost, not
294 : // finished. Stop is requested while this branch
295 : // still holds its share.
296 118 : if(p_->segment_.abandoned())
297 : {
298 14 : lost->store(true, std::memory_order_relaxed);
299 14 : core.stop_source_.request_stop();
300 : }
301 118 : h.destroy();
302 :
303 118 : auto remaining = counter->fetch_sub(1, std::memory_order_acq_rel);
304 118 : if(remaining != 1)
305 69 : return detail::symmetric_transfer(std::noop_coroutine());
306 49 : if(lost->load(std::memory_order_relaxed))
307 : {
308 8 : auto parent = cont;
309 8 : capy::abandon(parent);
310 8 : return detail::symmetric_transfer(std::noop_coroutine());
311 : }
312 41 : return detail::symmetric_transfer(caller_env->executor.dispatch(cont));
313 : }
314 : void await_resume() const noexcept {} // LCOV_EXCL_LINE final_suspend awaiter, never resumed
315 : };
316 118 : return awaiter{this};
317 : }
318 :
319 105 : void return_void() noexcept {}
320 :
321 : // Exceptions do NOT win in `results` when_any
322 13 : void unhandled_exception() noexcept
323 : {
324 13 : state_->record_exception(std::current_exception());
325 13 : }
326 :
327 : template<class Awaitable>
328 : struct transform_awaiter
329 : {
330 : std::decay_t<Awaitable> a_;
331 : promise_type* p_;
332 :
333 120 : bool await_ready() { return a_.await_ready(); }
334 118 : decltype(auto) await_resume() { return a_.await_resume(); }
335 :
336 : template<class Promise>
337 119 : auto await_suspend(std::coroutine_handle<Promise> h)
338 : {
339 : using R = decltype(a_.await_suspend(h, &p_->env_));
340 : if constexpr (std::is_same_v<R, std::coroutine_handle<>>)
341 119 : return detail::symmetric_transfer(a_.await_suspend(h, &p_->env_));
342 : else
343 : return a_.await_suspend(h, &p_->env_);
344 : }
345 : };
346 :
347 : template<class Awaitable>
348 120 : auto await_transform(Awaitable&& a)
349 : {
350 : using A = std::decay_t<Awaitable>;
351 : if constexpr (ContextAwaitable<A>)
352 : {
353 : return transform_awaiter<Awaitable>{
354 120 : std::forward<Awaitable>(a), this};
355 : }
356 : else
357 : {
358 : static_assert(sizeof(A) == 0, "requires ContextAwaitable");
359 : }
360 : }
361 : };
362 :
363 : std::coroutine_handle<promise_type> h_;
364 :
365 126 : explicit when_any_results_runner(std::coroutine_handle<promise_type> h) noexcept
366 126 : : h_(h)
367 : {
368 126 : }
369 :
370 : when_any_results_runner(when_any_results_runner&& other) noexcept
371 : : h_(std::exchange(other.h_, nullptr))
372 : {
373 : }
374 :
375 : when_any_results_runner(when_any_results_runner const&) = delete;
376 : when_any_results_runner& operator=(when_any_results_runner const&) = delete;
377 : when_any_results_runner& operator=(when_any_results_runner&&) = delete;
378 :
379 126 : auto release() noexcept
380 : {
381 126 : return std::exchange(h_, nullptr);
382 : }
383 : };
384 :
385 : // Runner coroutine: only tries to win when the child returns !ec.
386 : template<std::size_t I, ContextAwaitable Awaitable, typename StateType>
387 : when_any_results_runner<StateType>
388 61 : make_when_any_results_runner(Awaitable* inner, StateType* state)
389 : {
390 : // Taken from the join's storage only once running, into a local:
391 : // a runner destroyed unstarted owns no branch, and one destroyed
392 : // unfinished destroys it before its promise reports the loss.
393 : Awaitable a(std::move(*inner));
394 : auto r = co_await abandonable<Awaitable>{a};
395 : if(!r)
396 : co_return;
397 : auto& result = *r;
398 :
399 : if(!std::get<0>(result))
400 : {
401 : // Success: try to claim winner
402 : if(state->core_.try_win(I))
403 : {
404 : try
405 : {
406 : state->result_.emplace(
407 : std::in_place_index<I + 1>,
408 : detail::extract_results_payload(std::move(result)));
409 : }
410 : catch(...)
411 : {
412 : state->core_.set_winner_exception(std::current_exception());
413 : }
414 : }
415 : }
416 : else
417 : {
418 : // Error: record but don't win
419 : state->record_error(std::get<0>(result));
420 : }
421 122 : }
422 :
423 : // Launcher for `results`-aware when_any.
424 : template<ContextAwaitable... Awaitables>
425 : class when_any_results_launcher
426 : {
427 : using state_type = when_any_results_state<
428 : results_payload_t<awaitable_result_t<Awaitables>>...>;
429 :
430 : std::tuple<Awaitables...>* tasks_;
431 : state_type* state_;
432 :
433 : public:
434 33 : when_any_results_launcher(
435 : std::tuple<Awaitables...>* tasks,
436 : state_type* state)
437 33 : : tasks_(tasks)
438 33 : , state_(state)
439 : {
440 33 : }
441 :
442 33 : bool await_ready() const noexcept
443 : {
444 33 : return sizeof...(Awaitables) == 0;
445 : }
446 :
447 33 : std::coroutine_handle<> await_suspend(
448 : std::coroutine_handle<> continuation, await_context const* caller_env)
449 : {
450 33 : state_->core_.continuation_.h = continuation;
451 33 : state_->core_.continuation_.owner = caller_env->owner;
452 33 : state_->core_.caller_env_ = caller_env;
453 :
454 33 : if(caller_env->stop_token.stop_possible())
455 : {
456 6 : state_->core_.parent_stop_callback_.emplace(
457 3 : caller_env->stop_token,
458 3 : when_any_core::stop_callback_fn{&state_->core_.stop_source_});
459 :
460 3 : if(caller_env->stop_token.stop_requested())
461 1 : state_->core_.stop_source_.request_stop();
462 : }
463 :
464 33 : auto token = state_->core_.stop_source_.get_token();
465 33 : launch_all(std::index_sequence_for<Awaitables...>{},
466 : caller_env->executor, token);
467 :
468 66 : return std::noop_coroutine();
469 33 : }
470 :
471 23 : void await_resume() const noexcept {}
472 :
473 : private:
474 : template<std::size_t... Is>
475 33 : void launch_all(std::index_sequence<Is...>,
476 : executor_ref ex, std::stop_token token)
477 : {
478 33 : (..., launch_one<Is>(ex, token));
479 33 : }
480 :
481 : template<std::size_t I>
482 61 : void launch_one(executor_ref caller_ex, std::stop_token token)
483 : {
484 61 : auto runner = make_when_any_results_runner<I>(
485 61 : &std::get<I>(*tasks_), state_);
486 :
487 61 : auto h = runner.release();
488 61 : h.promise().state_ = state_;
489 61 : h.promise().index_ = I;
490 61 : h.promise().segment_.bind(h, caller_ex);
491 61 : h.promise().env_ = await_context{caller_ex, token,
492 61 : state_->core_.caller_env_->frame_allocator,
493 122 : &h.promise().segment_};
494 :
495 61 : state_->runner_handles_[I].h = std::coroutine_handle<>{h};
496 61 : caller_ex.post(state_->runner_handles_[I]);
497 122 : }
498 : };
499 :
500 : /** Shared state for homogeneous `results`-aware when_any (range overload).
501 :
502 : @tparam T The payload type extracted from `results`.
503 : */
504 : template<typename T>
505 : struct when_any_results_homogeneous_state
506 : {
507 : when_any_core core_;
508 : std::optional<T> result_;
509 : std::unique_ptr<continuation[]> runner_handles_;
510 :
511 : std::mutex failure_mu_;
512 : std::error_code last_error_;
513 : std::exception_ptr last_exception_;
514 :
515 15 : explicit when_any_results_homogeneous_state(std::size_t count)
516 15 : : core_(count)
517 15 : , runner_handles_(std::make_unique<continuation[]>(count))
518 : {
519 15 : }
520 :
521 6 : void record_error(std::error_code ec)
522 : {
523 6 : std::lock_guard lk(failure_mu_);
524 6 : last_error_ = ec;
525 6 : last_exception_ = nullptr;
526 6 : }
527 :
528 4 : void record_exception(std::exception_ptr ep)
529 : {
530 4 : std::lock_guard lk(failure_mu_);
531 4 : last_exception_ = ep;
532 4 : last_error_ = {};
533 4 : }
534 : };
535 :
536 : /** Specialization for void `results` children (no payload storage). */
537 : template<>
538 : struct when_any_results_homogeneous_state<std::tuple<>>
539 : {
540 : when_any_core core_;
541 : std::unique_ptr<continuation[]> runner_handles_;
542 :
543 : std::mutex failure_mu_;
544 : std::error_code last_error_;
545 : std::exception_ptr last_exception_;
546 :
547 5 : explicit when_any_results_homogeneous_state(std::size_t count)
548 5 : : core_(count)
549 5 : , runner_handles_(std::make_unique<continuation[]>(count))
550 : {
551 5 : }
552 :
553 1 : void record_error(std::error_code ec)
554 : {
555 1 : std::lock_guard lk(failure_mu_);
556 1 : last_error_ = ec;
557 1 : last_exception_ = nullptr;
558 1 : }
559 :
560 2 : void record_exception(std::exception_ptr ep)
561 : {
562 2 : std::lock_guard lk(failure_mu_);
563 2 : last_exception_ = ep;
564 2 : last_error_ = {};
565 2 : }
566 : };
567 :
568 : /** Create a `results`-aware runner for homogeneous when_any (range path).
569 :
570 : Only tries to win when the child returns !ec.
571 : */
572 : template<ContextAwaitable Awaitable, typename StateType>
573 : when_any_results_runner<StateType>
574 65 : make_when_any_results_homogeneous_runner(
575 : Awaitable* inner, StateType* state, std::size_t index)
576 : {
577 : // Taken from the join's storage only once running, into a local:
578 : // a runner destroyed unstarted owns no branch, and one destroyed
579 : // unfinished destroys it before its promise reports the loss.
580 : Awaitable a(std::move(*inner));
581 : auto r = co_await abandonable<Awaitable>{a};
582 : if(!r)
583 : co_return;
584 : auto& result = *r;
585 :
586 : if(!std::get<0>(result))
587 : {
588 : if(state->core_.try_win(index))
589 : {
590 : using PayloadT = results_payload_t<
591 : awaitable_result_t<Awaitable>>;
592 : if constexpr (!std::is_same_v<PayloadT, std::tuple<>>)
593 : {
594 : try
595 : {
596 : state->result_.emplace(
597 : extract_results_payload(std::move(result)));
598 : }
599 : catch(...)
600 : {
601 : state->core_.set_winner_exception(
602 : std::current_exception());
603 : }
604 : }
605 : }
606 : }
607 : else
608 : {
609 : state->record_error(std::get<0>(result));
610 : }
611 130 : }
612 :
613 : /** Starts all `results`-aware homogeneous runners concurrently. */
614 : template<ContextAwaitableRange Range>
615 : class when_any_results_homogeneous_launcher
616 : {
617 : using Awaitable = std::ranges::range_value_t<Range>;
618 : using PayloadT = results_payload_t<awaitable_result_t<Awaitable>>;
619 :
620 : Range* range_;
621 : when_any_results_homogeneous_state<PayloadT>* state_;
622 :
623 : public:
624 20 : when_any_results_homogeneous_launcher(
625 : Range* range,
626 : when_any_results_homogeneous_state<PayloadT>* state)
627 20 : : range_(range)
628 20 : , state_(state)
629 : {
630 20 : }
631 :
632 20 : bool await_ready() const noexcept
633 : {
634 20 : return std::ranges::empty(*range_);
635 : }
636 :
637 20 : std::coroutine_handle<> await_suspend(
638 : std::coroutine_handle<> continuation, await_context const* caller_env)
639 : {
640 20 : state_->core_.continuation_.h = continuation;
641 20 : state_->core_.continuation_.owner = caller_env->owner;
642 20 : state_->core_.caller_env_ = caller_env;
643 :
644 20 : if(caller_env->stop_token.stop_possible())
645 : {
646 4 : state_->core_.parent_stop_callback_.emplace(
647 2 : caller_env->stop_token,
648 2 : when_any_core::stop_callback_fn{&state_->core_.stop_source_});
649 :
650 2 : if(caller_env->stop_token.stop_requested())
651 1 : state_->core_.stop_source_.request_stop();
652 : }
653 :
654 20 : auto token = state_->core_.stop_source_.get_token();
655 :
656 : // Phase 1: Create all runners without dispatching.
657 20 : std::size_t index = 0;
658 85 : for(auto&& a : *range_)
659 : {
660 65 : auto runner = make_when_any_results_homogeneous_runner(
661 : &a, state_, index);
662 :
663 65 : auto h = runner.release();
664 65 : h.promise().state_ = state_;
665 65 : h.promise().index_ = index;
666 65 : h.promise().segment_.bind(h, caller_env->executor);
667 65 : h.promise().env_ = await_context{caller_env->executor, token,
668 130 : caller_env->frame_allocator, &h.promise().segment_};
669 :
670 65 : state_->runner_handles_[index].h = std::coroutine_handle<>{h};
671 65 : ++index;
672 : }
673 :
674 : // Phase 2: Post all runners. Any may complete synchronously.
675 20 : auto* handles = state_->runner_handles_.get();
676 20 : std::size_t count = state_->core_.remaining_count_.load(std::memory_order_relaxed);
677 85 : for(std::size_t i = 0; i < count; ++i)
678 65 : caller_env->executor.post(handles[i]);
679 :
680 40 : return std::noop_coroutine();
681 85 : }
682 :
683 18 : void await_resume() const noexcept {}
684 : };
685 :
686 : } // namespace detail
687 :
688 : /** Race a range of `results`-returning awaitables (non-void payloads).
689 :
690 : Only a child returning !ec can win. Errors and exceptions do not
691 : claim winner status. If all children fail, an unspecified one of
692 : the failures is reported — either an error_code at variant index 0,
693 : or a child's exception rethrown.
694 :
695 : @par Await-effects
696 :
697 : Takes ownership of the range, creates one wrapper coroutine per
698 : element, then posts every wrapper to the caller's executor. All
699 : children therefore run concurrently, each awaited with the caller's
700 : executor and frame allocator and with a stop token owned by this
701 : operation.
702 :
703 : Awaiting an empty range throws `std::invalid_argument` before any
704 : child is started.
705 :
706 : The first child to await-return a zero `ec` claims the win. Claiming
707 : the win requests stop on the operation's own stop token, which every
708 : sibling observes through the stop token it was awaited with. A child
709 : that await-returns a non-zero `ec`, or that exits via an exception,
710 : does not claim the win and does not request stop. The operation keeps
711 : waiting for a success. A stop request on the caller's stop token is
712 : also forwarded to every child.
713 :
714 : The await completes only after every child has finished, regardless of
715 : whether a win was claimed.
716 :
717 : @par Await-returns
718 : An object of type
719 : `std::variant<std::error_code, std::pair<std::size_t, PayloadT>>`,
720 : where `PayloadT` is the payload of one child's `results`.
721 :
722 : @li Index 1 holds the winner's position in the input range paired
723 : with its payload.
724 : @li Index 0 holds a non-zero `error_code` when no child won, that is,
725 : when every child failed. It is the `ec` of one of the failed
726 : children; which one is unspecified.
727 :
728 : A child that succeeds after the win has already been claimed
729 : contributes nothing: its payload is discarded.
730 :
731 : If no child won and the failure selected for reporting is an
732 : exception rather than an `ec`, that exception is rethrown instead of
733 : await-returning. The choice of child is unspecified.
734 :
735 : @par Await-postcondition
736 : Every child has finished. If at least one child await-returned a zero
737 : `ec`, the result holds index 1. If producing the winner's payload
738 : threw, that exception is rethrown. Otherwise the result holds index 0,
739 : or a failed child's exception is rethrown.
740 :
741 : @par Remarks
742 : Supports _ContextAwaitable cancellation_. A canceled child await-returns a
743 : non-zero `ec` and so cannot win; if no child has already succeeded,
744 : the result settles at index 0.
745 :
746 : @par Thread Safety
747 : The returned task must be awaited from a single execution context.
748 : Child awaitables execute concurrently but complete through the caller's
749 : executor.
750 :
751 : @param awaitables Range of `results`-returning awaitables (must
752 : not be empty).
753 :
754 : @return A task yielding variant<error_code, pair<size_t, PayloadT>>
755 : where index 0 is failure and index 1 carries the winner's
756 : index and payload.
757 :
758 : @throws std::invalid_argument if range is empty.
759 :
760 : @par Exception Safety
761 : The winner's exception is rethrown if extracting or
762 : move-constructing the winning payload throws. In that case a winner
763 : was found, but its result could not be produced. If all children
764 : fail and the reported failure is an exception, that child's
765 : exception is rethrown (which child is unspecified).
766 :
767 : @note If a branch's coroutine is abandoned before it finishes, for
768 : example by teardown of its execution context, `when_any`
769 : requests stop on the remaining branches, and the awaiting
770 : parent is abandoned rather than resumed once the last branch
771 : has finished or been abandoned.
772 :
773 : @par Example
774 : @par !example example_1
775 :
776 :
777 : @see ContextAwaitableRange, when_any
778 : */
779 : template<ContextAwaitableRange R>
780 : requires detail::is_results_v<
781 : awaitable_result_t<std::ranges::range_value_t<R>>>
782 : && (!std::is_same_v<
783 : detail::results_payload_t<
784 : awaitable_result_t<std::ranges::range_value_t<R>>>,
785 : std::tuple<>>)
786 16 : [[nodiscard]] auto when_any(R&& awaitables)
787 : -> task<std::variant<std::error_code,
788 : std::pair<std::size_t,
789 : detail::results_payload_t<
790 : awaitable_result_t<std::ranges::range_value_t<R>>>>>>
791 : {
792 : using Awaitable = std::ranges::range_value_t<R>;
793 : using PayloadT = detail::results_payload_t<
794 : awaitable_result_t<Awaitable>>;
795 : using result_type = std::variant<std::error_code,
796 : std::pair<std::size_t, PayloadT>>;
797 : using OwnedRange = detail::owned_range_t<R>;
798 :
799 : auto count = std::ranges::size(awaitables);
800 : if(count == 0)
801 : throw std::invalid_argument("when_any requires at least one awaitable");
802 :
803 : OwnedRange owned_awaitables = detail::own_range(std::forward<R>(awaitables));
804 :
805 : detail::when_any_results_homogeneous_state<PayloadT> state(count);
806 :
807 : co_await detail::when_any_results_homogeneous_launcher<OwnedRange>(
808 : &owned_awaitables, &state);
809 :
810 : // Winner found
811 : if(state.core_.has_winner_.load(std::memory_order_acquire))
812 : {
813 : if(state.core_.winner_exception_)
814 : std::rethrow_exception(state.core_.winner_exception_);
815 : co_return result_type{std::in_place_index<1>,
816 : std::pair{state.core_.winner_index_, std::move(*state.result_)}};
817 : }
818 :
819 : // No winner — report the recorded failure
820 : if(state.last_exception_)
821 : std::rethrow_exception(state.last_exception_);
822 : co_return result_type{std::in_place_index<0>, state.last_error_};
823 32 : }
824 :
825 : /** Race a range of void `results`-returning awaitables.
826 :
827 : Only a child returning !ec can win. Returns the winner's index
828 : at variant index 1, or error_code at index 0 on all-fail.
829 :
830 : @par Await-effects
831 :
832 : Takes ownership of the range, creates one wrapper coroutine per
833 : element, then posts every wrapper to the caller's executor. All
834 : children therefore run concurrently, each awaited with the caller's
835 : executor and frame allocator and with a stop token owned by this
836 : operation.
837 :
838 : Awaiting an empty range throws `std::invalid_argument` before any
839 : child is started.
840 :
841 : The first child to await-return a zero `ec` claims the win. Claiming
842 : the win requests stop on the operation's own stop token, which every
843 : sibling observes through the stop token it was awaited with. A child
844 : that await-returns a non-zero `ec`, or that exits via an exception,
845 : does not claim the win and does not request stop. The operation keeps
846 : waiting for a success. A stop request on the caller's stop token is
847 : also forwarded to every child.
848 :
849 : The await completes only after every child has finished, regardless of
850 : whether a win was claimed.
851 :
852 : @par Await-returns
853 : An object of type `std::variant<std::error_code, std::size_t>`.
854 :
855 : @li Index 1 holds the winner's position in the input range. The
856 : children have no payloads, so nothing else is reported.
857 : @li Index 0 holds a non-zero `error_code` when no child won, that is,
858 : when every child failed. It is the `ec` of one of the failed
859 : children; which one is unspecified.
860 :
861 : If no child won and the failure selected for reporting is an
862 : exception rather than an `ec`, that exception is rethrown instead of
863 : await-returning. The choice of child is unspecified.
864 :
865 : @par Await-postcondition
866 : Every child has finished. The result holds index 1 if at least one
867 : child await-returned a zero `ec`. Otherwise the result holds index 0,
868 : or a failed child's exception is rethrown.
869 :
870 : @par Remarks
871 : Supports _ContextAwaitable cancellation_. A canceled child await-returns a
872 : non-zero `ec` and so cannot win; if no child has already succeeded,
873 : the result settles at index 0.
874 :
875 : @par Thread Safety
876 : The returned task must be awaited from a single execution context.
877 : Child awaitables execute concurrently but complete through the caller's
878 : executor.
879 :
880 : @param awaitables Range of `results<>`-returning awaitables (must
881 : not be empty).
882 :
883 : @return A task yielding variant<error_code, size_t> where index 0
884 : is failure and index 1 carries the winner's index.
885 :
886 : @throws std::invalid_argument if range is empty.
887 :
888 : @par Exception Safety
889 : If all children fail and the reported failure is an exception,
890 : that child's exception is rethrown (which child is unspecified).
891 :
892 : @note If a branch's coroutine is abandoned before it finishes, for
893 : example by teardown of its execution context, `when_any`
894 : requests stop on the remaining branches, and the awaiting
895 : parent is abandoned rather than resumed once the last branch
896 : has finished or been abandoned.
897 :
898 : @par Example
899 : @par !example example_2
900 :
901 :
902 : @see ContextAwaitableRange, when_any
903 : */
904 : template<ContextAwaitableRange R>
905 : requires detail::is_results_v<
906 : awaitable_result_t<std::ranges::range_value_t<R>>>
907 : && std::is_same_v<
908 : detail::results_payload_t<
909 : awaitable_result_t<std::ranges::range_value_t<R>>>,
910 : std::tuple<>>
911 5 : [[nodiscard]] auto when_any(R&& awaitables)
912 : -> task<std::variant<std::error_code, std::size_t>>
913 : {
914 : using OwnedRange = detail::owned_range_t<R>;
915 : using result_type = std::variant<std::error_code, std::size_t>;
916 :
917 : auto count = std::ranges::size(awaitables);
918 : if(count == 0)
919 : throw std::invalid_argument("when_any requires at least one awaitable");
920 :
921 : OwnedRange owned_awaitables = detail::own_range(std::forward<R>(awaitables));
922 :
923 : detail::when_any_results_homogeneous_state<std::tuple<>> state(count);
924 :
925 : co_await detail::when_any_results_homogeneous_launcher<OwnedRange>(
926 : &owned_awaitables, &state);
927 :
928 : // Winner found
929 : if(state.core_.has_winner_.load(std::memory_order_acquire))
930 : {
931 : if(state.core_.winner_exception_)
932 : std::rethrow_exception(state.core_.winner_exception_);
933 : co_return result_type{std::in_place_index<1>,
934 : state.core_.winner_index_};
935 : }
936 :
937 : // No winner — report the recorded failure
938 : if(state.last_exception_)
939 : std::rethrow_exception(state.last_exception_);
940 : co_return result_type{std::in_place_index<0>, state.last_error_};
941 10 : }
942 :
943 : /** Race `results`-returning awaitables, selecting the first success.
944 :
945 : Overload selected when all children return results<Ts...>.
946 : Only a child returning !ec can win. Errors and exceptions do
947 : not claim winner status.
948 :
949 : @par Await-effects
950 :
951 : Creates and posts one wrapper coroutine per argument to the caller's
952 : executor, in argument order. All children therefore run concurrently,
953 : each awaited with the caller's executor and frame allocator and with
954 : a stop token owned by this operation. The overload requires at least
955 : one awaitable, so there is no empty case.
956 :
957 : The first child to await-return a zero `ec` claims the win. Claiming
958 : the win requests stop on the operation's own stop token, which every
959 : sibling observes through the stop token it was awaited with. A child
960 : that await-returns a non-zero `ec`, or that exits via an exception,
961 : does not claim the win and does not request stop. The operation keeps
962 : waiting for a success. A stop request on the caller's stop token is
963 : also forwarded to every child.
964 :
965 : The await completes only after every child has finished, regardless of
966 : whether a win was claimed.
967 :
968 : @par Await-returns
969 : An object of type `std::variant<std::error_code, P1, ..., Pn>`, where
970 : `Pi` is the payload of the i-th child's `results`.
971 :
972 : @li Index i+1 identifies the i-th argument as the winner and holds
973 : its payload.
974 : @li Index 0 holds a non-zero `error_code` when no child won, that is,
975 : when every child failed. It is the `ec` of one of the failed
976 : children; which one is unspecified.
977 :
978 : A child that succeeds after the win has already been claimed
979 : contributes nothing: its payload is discarded.
980 :
981 : If no child won and the failure selected for reporting is an
982 : exception rather than an `ec`, that exception is rethrown instead of
983 : await-returning. The choice of child is unspecified.
984 :
985 : @par Await-postcondition
986 : Every child has finished. If at least one child await-returned a zero
987 : `ec`, the result holds the index of the winning child. If producing
988 : the winner's payload threw, that exception is rethrown. Otherwise the
989 : result holds index 0, or a failed child's exception is rethrown.
990 :
991 : @par Remarks
992 : Supports _ContextAwaitable cancellation_. A canceled child await-returns a
993 : non-zero `ec` and so cannot win; if no child has already succeeded,
994 : the result settles at index 0.
995 :
996 : @par Thread Safety
997 : The returned task must be awaited from a single execution context.
998 : Child awaitables execute concurrently but complete through the caller's
999 : executor.
1000 :
1001 : @param as The awaitables to race. Each must satisfy @ref
1002 : ContextAwaitable and is consumed (moved-from) when `when_any`
1003 : is awaited.
1004 :
1005 : @return A task yielding variant<error_code, R1, ..., Rn> where
1006 : index 0 is the failure/no-winner case and index i+1
1007 : identifies the winning child. On all-fail, index 0 holds
1008 : an error_code from one of the failed children (unspecified
1009 : which; no priority between errors and exceptions).
1010 :
1011 : @par Exception Safety
1012 : The winner's exception is rethrown if extracting or constructing
1013 : the winning payload throws. In that case a winner was found, but
1014 : its result could not be produced. If all children fail and the
1015 : reported failure is an exception, that child's exception is
1016 : rethrown (which child is unspecified).
1017 :
1018 : @note A failing child does not cancel its siblings; `when_any`
1019 : waits for a success or for every child to finish. To make a
1020 : benign error (e.g. @c cond::canceled) count as a win, wrap
1021 : the child to translate the error into success. See the
1022 : Concurrent Composition tutorial.
1023 :
1024 : @note If a branch's coroutine is abandoned before it finishes, for
1025 : example by teardown of its execution context, `when_any`
1026 : requests stop on the remaining branches, and the awaiting
1027 : parent is abandoned rather than resumed once the last branch
1028 : has finished or been abandoned.
1029 : */
1030 : template<ContextAwaitable... As>
1031 : requires (sizeof...(As) > 0)
1032 : && detail::all_results_awaitables<As...>
1033 33 : [[nodiscard]] auto when_any(As... as)
1034 : -> task<std::variant<
1035 : std::error_code,
1036 : detail::results_payload_t<awaitable_result_t<As>>...>>
1037 : {
1038 : using result_type = std::variant<
1039 : std::error_code,
1040 : detail::results_payload_t<awaitable_result_t<As>>...>;
1041 :
1042 : detail::when_any_results_state<
1043 : detail::results_payload_t<awaitable_result_t<As>>...> state;
1044 : std::tuple<As...> awaitable_tuple(std::move(as)...);
1045 :
1046 : co_await detail::when_any_results_launcher<As...>(
1047 : &awaitable_tuple, &state);
1048 :
1049 : // Winner found: return their result
1050 : if(state.result_.has_value())
1051 : co_return std::move(*state.result_);
1052 :
1053 : // Winner claimed but payload construction failed
1054 : if(state.core_.winner_exception_)
1055 : std::rethrow_exception(state.core_.winner_exception_);
1056 :
1057 : // No winner — report the recorded failure
1058 : if(state.last_exception_)
1059 : std::rethrow_exception(state.last_exception_);
1060 : co_return result_type{std::in_place_index<0>, state.last_error_};
1061 66 : }
1062 :
1063 : } // namespace capy
1064 : } // namespace boost
1065 :
1066 : #endif
|
