1   
//
1   
//
2   
// Copyright (c) 2026 Steve Gerbino
2   
// Copyright (c) 2026 Steve Gerbino
3   
//
3   
//
4   
// Distributed under the Boost Software License, Version 1.0. (See accompanying
4   
// Distributed under the Boost Software License, Version 1.0. (See accompanying
5   
// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
5   
// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
6   
//
6   
//
7   
// Official repository: https://github.com/cppalliance/capy
7   
// Official repository: https://github.com/cppalliance/capy
8   
//
8   
//
9  
9  
10   
#ifndef BOOST_CAPY_WHEN_ALL_HPP
10   
#ifndef BOOST_CAPY_WHEN_ALL_HPP
11   
#define BOOST_CAPY_WHEN_ALL_HPP
11   
#define BOOST_CAPY_WHEN_ALL_HPP
12  
12  
13 - 
#include <boost/capy/detail/void_to_monostate.hpp>
 
14   
#include <boost/capy/detail/config.hpp>
13   
#include <boost/capy/detail/config.hpp>
15   
#include <boost/capy/concept/executor.hpp>
14   
#include <boost/capy/concept/executor.hpp>
16   
#include <boost/capy/concept/io_awaitable.hpp>
15   
#include <boost/capy/concept/io_awaitable.hpp>
17   
#include <coroutine>
16   
#include <coroutine>
18   
#include <boost/capy/ex/io_env.hpp>
17   
#include <boost/capy/ex/io_env.hpp>
19   
#include <boost/capy/ex/frame_allocator.hpp>
18   
#include <boost/capy/ex/frame_allocator.hpp>
20   
#include <boost/capy/task.hpp>
19   
#include <boost/capy/task.hpp>
21  
20  
22   
#include <array>
21   
#include <array>
23   
#include <atomic>
22   
#include <atomic>
24   
#include <exception>
23   
#include <exception>
25 - 
#include <ranges>
 
26 - 
#include <stdexcept>
 
27   
#include <optional>
24   
#include <optional>
28   
#include <stop_token>
25   
#include <stop_token>
29   
#include <tuple>
26   
#include <tuple>
30   
#include <type_traits>
27   
#include <type_traits>
31 - 
#include <vector>
 
32   
#include <utility>
28   
#include <utility>
33  
29  
34   
namespace boost {
30   
namespace boost {
35   
namespace capy {
31   
namespace capy {
36  
32  
37   
namespace detail {
33   
namespace detail {
38  
34  
 
35 + 
/** Type trait to filter void types from a tuple.
 
36 +
 
37 + 
    Void-returning tasks do not contribute a value to the result tuple.
 
38 + 
    This trait computes the filtered result type.
 
39 +
 
40 + 
    Example: filter_void_tuple_t<int, void, string> = tuple<int, string>
 
41 + 
*/
 
42 + 
template<typename T>
 
43 + 
using wrap_non_void_t = std::conditional_t<std::is_void_v<T>, std::tuple<>, std::tuple<T>>;
 
44 +
 
45 + 
template<typename... Ts>
 
46 + 
using filter_void_tuple_t = decltype(std::tuple_cat(std::declval<wrap_non_void_t<Ts>>()...));
 
47 +
39   
/** Holds the result of a single task within when_all.
48   
/** Holds the result of a single task within when_all.
40   
*/
49   
*/
41   
template<typename T>
50   
template<typename T>
42   
struct result_holder
51   
struct result_holder
43   
{
52   
{
44   
    std::optional<T> value_;
53   
    std::optional<T> value_;
45  
54  
46   
    void set(T v)
55   
    void set(T v)
47   
    {
56   
    {
48   
        value_ = std::move(v);
57   
        value_ = std::move(v);
49   
    }
58   
    }
50  
59  
51   
    T get() &&
60   
    T get() &&
52   
    {
61   
    {
53   
        return std::move(*value_);
62   
        return std::move(*value_);
54   
    }
63   
    }
55   
};
64   
};
56  
65  
57 - 
/** Specialization for void tasks - returns monostate to preserve index mapping.
66 + 
/** Specialization for void tasks - no value storage needed.
58   
*/
67   
*/
59   
template<>
68   
template<>
60   
struct result_holder<void>
69   
struct result_holder<void>
61 - 
    std::monostate get() && { return {}; }
 
62   
{
70   
{
63   
};
71   
};
64  
72  
65 - 
/** Core shared state for when_all operations.
73 + 
/** Shared state for when_all operation.
66 -
 
67 - 
    Contains all members and methods common to both heterogeneous (variadic)
 
68 - 
    and homogeneous (range) when_all implementations. State classes embed
 
69 - 
    this via composition to avoid CRTP destructor ordering issues.
 
70  
74  
71 - 
    @par Thread Safety
75 + 
    @tparam Ts The result types of the tasks.
72 - 
    Atomic operations protect exception capture and completion count.
 
73   
*/
76   
*/
74 - 
struct when_all_core
77 + 
template<typename... Ts>
 
78 + 
struct when_all_state
75   
{
79   
{
 
80 + 
    static constexpr std::size_t task_count = sizeof...(Ts);
 
81 +
 
82 + 
    // Completion tracking - when_all waits for all children
76   
    std::atomic<std::size_t> remaining_count_;
83   
    std::atomic<std::size_t> remaining_count_;
77  
84  
 
85 + 
    // Result storage in input order
 
86 + 
    std::tuple<result_holder<Ts>...> results_;
 
87 +
 
88 + 
    // Runner handles - destroyed in await_resume while allocator is valid
 
89 + 
    std::array<std::coroutine_handle<>, task_count> runner_handles_{};
 
90 +
78   
    // Exception storage - first error wins, others discarded
91   
    // Exception storage - first error wins, others discarded
79   
    std::atomic<bool> has_exception_{false};
92   
    std::atomic<bool> has_exception_{false};
80   
    std::exception_ptr first_exception_;
93   
    std::exception_ptr first_exception_;
81  
94  
 
95 + 
    // Stop propagation - on error, request stop for siblings
82   
    std::stop_source stop_source_;
96   
    std::stop_source stop_source_;
83  
97  
84 - 
    // Bridges parent's stop token to our stop_source
98 + 
    // Connects parent's stop_token to our stop_source
85   
    struct stop_callback_fn
99   
    struct stop_callback_fn
86   
    {
100   
    {
87   
        std::stop_source* source_;
101   
        std::stop_source* source_;
88   
        void operator()() const { source_->request_stop(); }
102   
        void operator()() const { source_->request_stop(); }
89   
    };
103   
    };
90   
    using stop_callback_t = std::stop_callback<stop_callback_fn>;
104   
    using stop_callback_t = std::stop_callback<stop_callback_fn>;
91   
    std::optional<stop_callback_t> parent_stop_callback_;
105   
    std::optional<stop_callback_t> parent_stop_callback_;
92  
106  
 
107 + 
    // Parent resumption
93   
    std::coroutine_handle<> continuation_;
108   
    std::coroutine_handle<> continuation_;
94   
    io_env const* caller_env_ = nullptr;
109   
    io_env const* caller_env_ = nullptr;
95  
110  
96 - 
    explicit when_all_core(std::size_t count) noexcept
111 + 
    when_all_state()
97 - 
        : remaining_count_(count)
112 + 
        : remaining_count_(task_count)
98   
    {
113   
    {
99   
    }
114   
    }
100  
115  
101 - 
    /** Capture an exception (first one wins). */
116 + 
    // Runners self-destruct in final_suspend. No destruction needed here.
 
117 +
 
118 + 
    /** Capture an exception (first one wins).
 
119 + 
    */
102   
    void capture_exception(std::exception_ptr ep)
120   
    void capture_exception(std::exception_ptr ep)
103   
    {
121   
    {
104   
        bool expected = false;
122   
        bool expected = false;
105   
        if(has_exception_.compare_exchange_strong(
123   
        if(has_exception_.compare_exchange_strong(
106   
            expected, true, std::memory_order_relaxed))
124   
            expected, true, std::memory_order_relaxed))
107   
            first_exception_ = ep;
125   
            first_exception_ = ep;
108 - 
};
 
109 -
 
110 - 
/** Shared state for heterogeneous when_all (variadic overload).
 
111 -
 
112 - 
    @tparam Ts The result types of the tasks.
 
113 - 
*/
 
114 - 
template<typename... Ts>
 
115 - 
struct when_all_state
 
116 - 
{
 
117 - 
    static constexpr std::size_t task_count = sizeof...(Ts);
 
118 -
 
119 - 
    when_all_core core_;
 
120 - 
    std::tuple<result_holder<Ts>...> results_;
 
121 - 
    std::array<std::coroutine_handle<>, task_count> runner_handles_{};
 
122 -
 
123 - 
    when_all_state()
 
124 - 
        : core_(task_count)
 
125 - 
    {
 
126 - 
    }
 
127 - 
};
 
128 -
 
129 - 
/** Shared state for homogeneous when_all (range overload).
 
130 -
 
131 - 
    Stores all results in a vector indexed by task position.
 
132 -
 
133 - 
    @tparam T The common result type of all tasks.
 
134 - 
*/
 
135 - 
template<typename T>
 
136 - 
struct when_all_homogeneous_state
 
137 - 
{
 
138 - 
    when_all_core core_;
 
139 - 
    std::vector<std::optional<T>> results_;
 
140 - 
    std::vector<std::coroutine_handle<>> runner_handles_;
 
141 -
 
142 - 
    explicit when_all_homogeneous_state(std::size_t count)
 
143 - 
        : core_(count)
 
144 - 
        , results_(count)
 
145 - 
        , runner_handles_(count)
 
146 - 
    {
 
147 - 
    }
 
148 -
 
149 - 
    void set_result(std::size_t index, T value)
 
150 - 
    {
 
151 - 
        results_[index].emplace(std::move(value));
 
152 - 
    }
 
153 - 
};
 
154 -
 
155 - 
/** Specialization for void tasks (no result storage). */
 
156 - 
template<>
 
157 - 
struct when_all_homogeneous_state<void>
 
158 - 
{
 
159 - 
    when_all_core core_;
 
160 - 
    std::vector<std::coroutine_handle<>> runner_handles_;
 
161   
    }
126   
    }
162 - 
    explicit when_all_homogeneous_state(std::size_t count)
 
163 - 
        : core_(count)
 
164 - 
        , runner_handles_(count)
 
165 - 
    {
 
166 - 
    }
 
167  
127  
168   
};
128   
};
169  
129  
170 - 
/** Wrapper coroutine that intercepts task completion for when_all.
130 + 
/** Wrapper coroutine that intercepts task completion.
171 -
 
172 - 
    Parameterized on StateType to work with both heterogeneous (variadic)
 
173 - 
    and homogeneous (range) state types. All state types expose their
 
174 - 
    shared members through a `core_` member of type when_all_core.
 
175  
131  
176 - 
    @tparam StateType The state type (when_all_state or when_all_homogeneous_state).
132 + 
    This runner awaits its assigned task and stores the result in
 
133 + 
    the shared state, or captures the exception and requests stop.
177   
*/
134   
*/
178 - 
template<typename StateType>
135 + 
template<typename T, typename... Ts>
179   
struct when_all_runner
136   
struct when_all_runner
180   
{
137   
{
181 - 
    struct promise_type
138 + 
    struct promise_type // : frame_allocating_base  // DISABLED FOR TESTING
182   
    {
139   
    {
183 - 
        StateType* state_ = nullptr;
140 + 
        when_all_state<Ts...>* state_ = nullptr;
184 - 
        std::size_t index_ = 0;
 
185   
        io_env env_;
141   
        io_env env_;
186  
142  
187 - 
        when_all_runner get_return_object() noexcept
143 + 
        when_all_runner get_return_object()
188   
        {
144   
        {
189 - 
            return when_all_runner(
145 + 
            return when_all_runner(std::coroutine_handle<promise_type>::from_promise(*this));
190 - 
                std::coroutine_handle<promise_type>::from_promise(*this));
 
191   
        }
146   
        }
192  
147  
193   
        std::suspend_always initial_suspend() noexcept
148   
        std::suspend_always initial_suspend() noexcept
194   
        {
149   
        {
195   
            return {};
150   
            return {};
196   
        }
151   
        }
197  
152  
198   
        auto final_suspend() noexcept
153   
        auto final_suspend() noexcept
199   
        {
154   
        {
200   
            struct awaiter
155   
            struct awaiter
201   
            {
156   
            {
202   
                promise_type* p_;
157   
                promise_type* p_;
203 - 
                bool await_ready() const noexcept { return false; }
158 +
 
159 + 
                bool await_ready() const noexcept
 
160 + 
                {
 
161 + 
                    return false;
 
162 + 
                }
 
163 +
204   
                auto await_suspend(std::coroutine_handle<> h) noexcept
164   
                auto await_suspend(std::coroutine_handle<> h) noexcept
205   
                {
165   
                {
206 - 
                    auto& core = p_->state_->core_;
166 + 
                    // Extract everything needed before self-destruction.
207 - 
                    auto* counter = &core.remaining_count_;
167 + 
                    auto* state = p_->state_;
208 - 
                    auto* caller_env = core.caller_env_;
168 + 
                    auto* counter = &state->remaining_count_;
209 - 
                    auto cont = core.continuation_;
169 + 
                    auto* caller_env = state->caller_env_;
 
170 + 
                    auto cont = state->continuation_;
210  
171  
211   
                    h.destroy();
172   
                    h.destroy();
212  
173  
 
174 + 
                    // If last runner, dispatch parent for symmetric transfer.
213   
                    auto remaining = counter->fetch_sub(1, std::memory_order_acq_rel);
175   
                    auto remaining = counter->fetch_sub(1, std::memory_order_acq_rel);
214   
                    if(remaining == 1)
176   
                    if(remaining == 1)
215   
                        return detail::symmetric_transfer(caller_env->executor.dispatch(cont));
177   
                        return detail::symmetric_transfer(caller_env->executor.dispatch(cont));
216   
                    return detail::symmetric_transfer(std::noop_coroutine());
178   
                    return detail::symmetric_transfer(std::noop_coroutine());
217   
                }
179   
                }
218 - 
                void await_resume() const noexcept {}
180 +
 
181 + 
                void await_resume() const noexcept
 
182 + 
                {
 
183 + 
                }
219   
            };
184   
            };
220   
            return awaiter{this};
185   
            return awaiter{this};
221   
        }
186   
        }
222  
187  
223 - 
        void return_void() noexcept {}
188 + 
        void return_void()
 
189 + 
        {
 
190 + 
        }
224  
191  
225   
        void unhandled_exception()
192   
        void unhandled_exception()
226   
        {
193   
        {
227 - 
            state_->core_.capture_exception(std::current_exception());
194 + 
            state_->capture_exception(std::current_exception());
228 - 
            state_->core_.stop_source_.request_stop();
195 + 
            // Request stop for sibling tasks
 
196 + 
            state_->stop_source_.request_stop();
229   
        }
197   
        }
230  
198  
231   
        template<class Awaitable>
199   
        template<class Awaitable>
232   
        struct transform_awaiter
200   
        struct transform_awaiter
233   
        {
201   
        {
234   
            std::decay_t<Awaitable> a_;
202   
            std::decay_t<Awaitable> a_;
235   
            promise_type* p_;
203   
            promise_type* p_;
236  
204  
237 - 
            bool await_ready() { return a_.await_ready(); }
205 + 
            bool await_ready()
238 - 
            decltype(auto) await_resume() { return a_.await_resume(); }
206 + 
            {
 
207 + 
                return a_.await_ready();
 
208 + 
            }
 
209 +
 
210 + 
            decltype(auto) await_resume()
 
211 + 
            {
 
212 + 
                return a_.await_resume();
 
213 + 
            }
239  
214  
240   
            template<class Promise>
215   
            template<class Promise>
241   
            auto await_suspend(std::coroutine_handle<Promise> h)
216   
            auto await_suspend(std::coroutine_handle<Promise> h)
242   
            {
217   
            {
243   
                using R = decltype(a_.await_suspend(h, &p_->env_));
218   
                using R = decltype(a_.await_suspend(h, &p_->env_));
244   
                if constexpr (std::is_same_v<R, std::coroutine_handle<>>)
219   
                if constexpr (std::is_same_v<R, std::coroutine_handle<>>)
245   
                    return detail::symmetric_transfer(a_.await_suspend(h, &p_->env_));
220   
                    return detail::symmetric_transfer(a_.await_suspend(h, &p_->env_));
246   
                else
221   
                else
247   
                    return a_.await_suspend(h, &p_->env_);
222   
                    return a_.await_suspend(h, &p_->env_);
248   
            }
223   
            }
249   
        };
224   
        };
250  
225  
251   
        template<class Awaitable>
226   
        template<class Awaitable>
252   
        auto await_transform(Awaitable&& a)
227   
        auto await_transform(Awaitable&& a)
253   
        {
228   
        {
254   
            using A = std::decay_t<Awaitable>;
229   
            using A = std::decay_t<Awaitable>;
255   
            if constexpr (IoAwaitable<A>)
230   
            if constexpr (IoAwaitable<A>)
256   
            {
231   
            {
257   
                return transform_awaiter<Awaitable>{
232   
                return transform_awaiter<Awaitable>{
258   
                    std::forward<Awaitable>(a), this};
233   
                    std::forward<Awaitable>(a), this};
259   
            }
234   
            }
260   
            else
235   
            else
261   
            {
236   
            {
262   
                static_assert(sizeof(A) == 0, "requires IoAwaitable");
237   
                static_assert(sizeof(A) == 0, "requires IoAwaitable");
263   
            }
238   
            }
264   
        }
239   
        }
265   
    };
240   
    };
266  
241  
267   
    std::coroutine_handle<promise_type> h_;
242   
    std::coroutine_handle<promise_type> h_;
268  
243  
269 - 
    explicit when_all_runner(std::coroutine_handle<promise_type> h) noexcept
244 + 
    explicit when_all_runner(std::coroutine_handle<promise_type> h)
270   
        : h_(h)
245   
        : h_(h)
271   
    {
246   
    {
272   
    }
247   
    }
273  
248  
274   
    // Enable move for all clang versions - some versions need it
249   
    // Enable move for all clang versions - some versions need it
275 - 
    when_all_runner(when_all_runner&& other) noexcept
250 + 
    when_all_runner(when_all_runner&& other) noexcept : h_(std::exchange(other.h_, nullptr)) {}
276 - 
        : h_(std::exchange(other.h_, nullptr))
 
277 - 
    {
 
278 - 
    }
 
279  
251  
 
252 + 
    // Non-copyable
280   
    when_all_runner(when_all_runner const&) = delete;
253   
    when_all_runner(when_all_runner const&) = delete;
281   
    when_all_runner& operator=(when_all_runner const&) = delete;
254   
    when_all_runner& operator=(when_all_runner const&) = delete;
282   
    when_all_runner& operator=(when_all_runner&&) = delete;
255   
    when_all_runner& operator=(when_all_runner&&) = delete;
283  
256  
284   
    auto release() noexcept
257   
    auto release() noexcept
285   
    {
258   
    {
286   
        return std::exchange(h_, nullptr);
259   
        return std::exchange(h_, nullptr);
287   
    }
260   
    }
288   
};
261   
};
289  
262  
290 - 
/** Create a runner coroutine for a single awaitable (variadic path).
263 + 
/** Create a runner coroutine for a single awaitable.
291  
264  
292 - 
    Uses compile-time index for tuple-based result storage.
265 + 
    Awaitable is passed directly to ensure proper coroutine frame storage.
293   
*/
266   
*/
294   
template<std::size_t Index, IoAwaitable Awaitable, typename... Ts>
267   
template<std::size_t Index, IoAwaitable Awaitable, typename... Ts>
295 - 
when_all_runner<when_all_state<Ts...>>
268 + 
when_all_runner<awaitable_result_t<Awaitable>, Ts...>
296   
make_when_all_runner(Awaitable inner, when_all_state<Ts...>* state)
269   
make_when_all_runner(Awaitable inner, when_all_state<Ts...>* state)
297   
{
270   
{
298   
    using T = awaitable_result_t<Awaitable>;
271   
    using T = awaitable_result_t<Awaitable>;
299   
    if constexpr (std::is_void_v<T>)
272   
    if constexpr (std::is_void_v<T>)
300   
    {
273   
    {
301   
        co_await std::move(inner);
274   
        co_await std::move(inner);
302   
    }
275   
    }
303   
    else
276   
    else
304   
    {
277   
    {
305   
        std::get<Index>(state->results_).set(co_await std::move(inner));
278   
        std::get<Index>(state->results_).set(co_await std::move(inner));
306   
    }
279   
    }
307   
}
280   
}
308  
281  
309 - 
/** Create a runner coroutine for a single awaitable (range path).
282 + 
/** Internal awaitable that launches all runner coroutines and waits.
310 -
 
311 - 
    Uses runtime index for vector-based result storage.
 
312 - 
*/
 
313 - 
template<IoAwaitable Awaitable, typename StateType>
 
314 - 
when_all_runner<StateType>
 
315 - 
make_when_all_homogeneous_runner(Awaitable inner, StateType* state, std::size_t index)
 
316 - 
{
 
317 - 
    using T = awaitable_result_t<Awaitable>;
 
318 - 
    if constexpr (std::is_void_v<T>)
 
319 - 
    {
 
320 - 
        co_await std::move(inner);
 
321 - 
    }
 
322 - 
    else
 
323 - 
    {
 
324 - 
        state->set_result(index, co_await std::move(inner));
 
325 - 
    }
 
326 - 
}
 
327 -
 
328 - 
/** Internal awaitable that launches all variadic runner coroutines.
 
329  
283  
330 - 
    CRITICAL: If the last task finishes synchronously then the parent
284 + 
    This awaitable is used inside the when_all coroutine to handle
331 - 
    coroutine resumes, destroying its frame, and destroying this object
285 + 
    the concurrent execution of child awaitables.
332 - 
    prior to the completion of await_suspend. Therefore, await_suspend
 
333 - 
    must ensure `this` cannot be referenced after calling `launch_one`
 
334 - 
    for the last time.
 
335   
*/
286   
*/
336   
template<IoAwaitable... Awaitables>
287   
template<IoAwaitable... Awaitables>
337   
class when_all_launcher
288   
class when_all_launcher
338   
{
289   
{
339   
    using state_type = when_all_state<awaitable_result_t<Awaitables>...>;
290   
    using state_type = when_all_state<awaitable_result_t<Awaitables>...>;
340  
291  
341   
    std::tuple<Awaitables...>* awaitables_;
292   
    std::tuple<Awaitables...>* awaitables_;
342   
    state_type* state_;
293   
    state_type* state_;
343  
294  
344   
public:
295   
public:
345   
    when_all_launcher(
296   
    when_all_launcher(
346   
        std::tuple<Awaitables...>* awaitables,
297   
        std::tuple<Awaitables...>* awaitables,
347   
        state_type* state)
298   
        state_type* state)
348   
        : awaitables_(awaitables)
299   
        : awaitables_(awaitables)
349   
        , state_(state)
300   
        , state_(state)
350   
    {
301   
    {
351   
    }
302   
    }
352  
303  
353   
    bool await_ready() const noexcept
304   
    bool await_ready() const noexcept
354   
    {
305   
    {
355   
        return sizeof...(Awaitables) == 0;
306   
        return sizeof...(Awaitables) == 0;
356   
    }
307   
    }
357  
308  
358   
    std::coroutine_handle<> await_suspend(std::coroutine_handle<> continuation, io_env const* caller_env)
309   
    std::coroutine_handle<> await_suspend(std::coroutine_handle<> continuation, io_env const* caller_env)
359   
    {
310   
    {
360 - 
        state_->core_.continuation_ = continuation;
311 + 
        state_->continuation_ = continuation;
361 - 
        state_->core_.caller_env_ = caller_env;
312 + 
        state_->caller_env_ = caller_env;
362  
313  
 
314 + 
        // Forward parent's stop requests to children
363   
        if(caller_env->stop_token.stop_possible())
315   
        if(caller_env->stop_token.stop_possible())
364   
        {
316   
        {
365 - 
            state_->core_.parent_stop_callback_.emplace(
317 + 
            state_->parent_stop_callback_.emplace(
366   
                caller_env->stop_token,
318   
                caller_env->stop_token,
367 - 
                when_all_core::stop_callback_fn{&state_->core_.stop_source_});
319 + 
                typename state_type::stop_callback_fn{&state_->stop_source_});
368  
320  
369   
            if(caller_env->stop_token.stop_requested())
321   
            if(caller_env->stop_token.stop_requested())
370 - 
                state_->core_.stop_source_.request_stop();
322 + 
                state_->stop_source_.request_stop();
371   
        }
323   
        }
372  
324  
373 - 
        auto token = state_->core_.stop_source_.get_token();
325 + 
        // CRITICAL: If the last task finishes synchronously then the parent
 
326 + 
        // coroutine resumes, destroying its frame, and destroying this object
 
327 + 
        // prior to the completion of await_suspend. Therefore, await_suspend
 
328 + 
        // must ensure `this` cannot be referenced after calling `launch_one`
 
329 + 
        // for the last time.
 
330 + 
        auto token = state_->stop_source_.get_token();
374   
        [&]<std::size_t... Is>(std::index_sequence<Is...>) {
331   
        [&]<std::size_t... Is>(std::index_sequence<Is...>) {
375   
            (..., launch_one<Is>(caller_env->executor, token));
332   
            (..., launch_one<Is>(caller_env->executor, token));
376   
        }(std::index_sequence_for<Awaitables...>{});
333   
        }(std::index_sequence_for<Awaitables...>{});
377  
334  
 
335 + 
        // Let signal_completion() handle resumption
378   
        return std::noop_coroutine();
336   
        return std::noop_coroutine();
379   
    }
337   
    }
380  
338  
381   
    void await_resume() const noexcept
339   
    void await_resume() const noexcept
382   
    {
340   
    {
 
341 + 
        // Results are extracted by the when_all coroutine from state
383   
    }
342   
    }
384  
343  
385   
private:
344   
private:
386   
    template<std::size_t I>
345   
    template<std::size_t I>
387   
    void launch_one(executor_ref caller_ex, std::stop_token token)
346   
    void launch_one(executor_ref caller_ex, std::stop_token token)
388   
    {
347   
    {
389   
        auto runner = make_when_all_runner<I>(
348   
        auto runner = make_when_all_runner<I>(
390   
            std::move(std::get<I>(*awaitables_)), state_);
349   
            std::move(std::get<I>(*awaitables_)), state_);
391  
350  
392   
        auto h = runner.release();
351   
        auto h = runner.release();
393   
        h.promise().state_ = state_;
352   
        h.promise().state_ = state_;
394 - 
        h.promise().env_ = io_env{caller_ex, token, state_->core_.caller_env_->frame_allocator};
353 + 
        h.promise().env_ = io_env{caller_ex, token, state_->caller_env_->frame_allocator};
395  
354  
396   
        std::coroutine_handle<> ch{h};
355   
        std::coroutine_handle<> ch{h};
397   
        state_->runner_handles_[I] = ch;
356   
        state_->runner_handles_[I] = ch;
398 - 
        state_->core_.caller_env_->executor.post(ch);
357 + 
        state_->caller_env_->executor.post(ch);
399   
    }
358   
    }
400   
};
359   
};
401  
360  
402 - 
/** Helper to extract a single result from state.
361 + 
/** Helper to extract a single result, returning empty tuple for void.
403   
    This is a separate function to work around a GCC-11 ICE that occurs
362   
    This is a separate function to work around a GCC-11 ICE that occurs
404   
    when using nested immediately-invoked lambdas with pack expansion.
363   
    when using nested immediately-invoked lambdas with pack expansion.
405   
*/
364   
*/
406   
template<std::size_t I, typename... Ts>
365   
template<std::size_t I, typename... Ts>
407   
auto extract_single_result(when_all_state<Ts...>& state)
366   
auto extract_single_result(when_all_state<Ts...>& state)
408   
{
367   
{
409 - 
    return std::move(std::get<I>(state.results_)).get();
368 + 
    using T = std::tuple_element_t<I, std::tuple<Ts...>>;
 
369 + 
    if constexpr (std::is_void_v<T>)
 
370 + 
        return std::tuple<>();
 
371 + 
    else
 
372 + 
        return std::make_tuple(std::move(std::get<I>(state.results_)).get());
410   
}
373   
}
411  
374  
412 - 
/** Extract all results from state as a tuple.
375 + 
/** Extract results from state, filtering void types.
413   
*/
376   
*/
414   
template<typename... Ts>
377   
template<typename... Ts>
415   
auto extract_results(when_all_state<Ts...>& state)
378   
auto extract_results(when_all_state<Ts...>& state)
416   
{
379   
{
417   
    return [&]<std::size_t... Is>(std::index_sequence<Is...>) {
380   
    return [&]<std::size_t... Is>(std::index_sequence<Is...>) {
418 - 
        return std::tuple(extract_single_result<Is>(state)...);
381 + 
        return std::tuple_cat(extract_single_result<Is>(state)...);
419   
    }(std::index_sequence_for<Ts...>{});
382   
    }(std::index_sequence_for<Ts...>{});
420   
}
383   
}
421 - 
/** Launches all homogeneous runners concurrently.
 
422 -
 
423 - 
    Two-phase approach: create all runners first, then post all.
 
424 - 
    This avoids lifetime issues if a task completes synchronously.
 
425 - 
*/
 
426 - 
template<typename Range>
 
427 - 
class when_all_homogeneous_launcher
 
428 - 
{
 
429 - 
    using Awaitable = std::ranges::range_value_t<Range>;
 
430 - 
    using T = awaitable_result_t<Awaitable>;
 
431 -
 
432 - 
    Range* range_;
 
433 - 
    when_all_homogeneous_state<T>* state_;
 
434 -
 
435 - 
public:
 
436 - 
    when_all_homogeneous_launcher(
 
437 - 
        Range* range,
 
438 - 
        when_all_homogeneous_state<T>* state)
 
439 - 
        : range_(range)
 
440 - 
        , state_(state)
 
441 - 
    {
 
442 - 
    }
 
443 -
 
444 - 
    bool await_ready() const noexcept
 
445 - 
    {
 
446 - 
        return std::ranges::empty(*range_);
 
447 - 
    }
 
448 -
 
449 - 
    std::coroutine_handle<> await_suspend(std::coroutine_handle<> continuation, io_env const* caller_env)
 
450 - 
    {
 
451 - 
        state_->core_.continuation_ = continuation;
 
452 - 
        state_->core_.caller_env_ = caller_env;
 
453 -
 
454 - 
        if(caller_env->stop_token.stop_possible())
 
455 - 
        {
 
456 - 
            state_->core_.parent_stop_callback_.emplace(
 
457 - 
                caller_env->stop_token,
 
458 - 
                when_all_core::stop_callback_fn{&state_->core_.stop_source_});
 
459 -
 
460 - 
            if(caller_env->stop_token.stop_requested())
 
461 - 
                state_->core_.stop_source_.request_stop();
 
462 - 
        }
 
463 -
 
464 - 
        auto token = state_->core_.stop_source_.get_token();
 
465 -
 
466 - 
        // Phase 1: Create all runners without dispatching.
 
467 - 
        std::size_t index = 0;
 
468 - 
        for(auto&& a : *range_)
 
469 - 
        {
 
470 - 
            auto runner = make_when_all_homogeneous_runner(
 
471 - 
                std::move(a), state_, index);
 
472 -
 
473 - 
            auto h = runner.release();
 
474 - 
            h.promise().state_ = state_;
 
475 - 
            h.promise().index_ = index;
 
476 - 
            h.promise().env_ = io_env{caller_env->executor, token, caller_env->frame_allocator};
 
477 -
 
478 - 
            state_->runner_handles_[index] = std::coroutine_handle<>{h};
 
479 - 
            ++index;
 
480 - 
        }
 
481 -
 
482 - 
        // Phase 2: Post all runners. Any may complete synchronously.
 
483 - 
        // After last post, state_ and this may be destroyed.
 
484 - 
        std::coroutine_handle<>* handles = state_->runner_handles_.data();
 
485 - 
        std::size_t count = state_->runner_handles_.size();
 
486 - 
        for(std::size_t i = 0; i < count; ++i)
 
487 - 
            caller_env->executor.post(handles[i]);
 
488 -
 
489 - 
        return std::noop_coroutine();
 
490 - 
    }
 
491 -
 
492 - 
    void await_resume() const noexcept
 
493 - 
    {
 
494 - 
    }
 
495 - 
};
 
496 -
 
497  
384  
498   
} // namespace detail
385   
} // namespace detail
499  
386  
500 - 
/** Compute the when_all result tuple type.
387 + 
/** Compute a tuple type with void types filtered out.
501  
388  
502 - 
    Void-returning tasks contribute std::monostate to preserve the
389 + 
    Returns void when all types are void (P2300 aligned),
503 - 
    task-index-to-result-index mapping, matching when_any's approach.
390 + 
    otherwise returns a std::tuple with void types removed.
504  
391  
505 - 
    Example: when_all_result_t<int, void, string> = std::tuple<int, std::monostate, string>
392 + 
    Example: non_void_tuple_t<int, void, string> = std::tuple<int, string>
506 - 
    Example: when_all_result_t<void, void> = std::tuple<std::monostate, std::monostate>
393 + 
    Example: non_void_tuple_t<void, void> = void
507   
*/
394   
*/
508   
template<typename... Ts>
395   
template<typename... Ts>
509 - 
using when_all_result_t = std::tuple<void_to_monostate_t<Ts>...>;
396 + 
using non_void_tuple_t = std::conditional_t<
 
397 + 
    std::is_same_v<detail::filter_void_tuple_t<Ts...>, std::tuple<>>,
 
398 + 
    void,
 
399 + 
    detail::filter_void_tuple_t<Ts...>>;
510  
400  
511   
/** Execute multiple awaitables concurrently and collect their results.
401   
/** Execute multiple awaitables concurrently and collect their results.
512  
402  
513   
    Launches all awaitables simultaneously and waits for all to complete
403   
    Launches all awaitables simultaneously and waits for all to complete
514   
    before returning. Results are collected in input order. If any
404   
    before returning. Results are collected in input order. If any
515   
    awaitable throws, cancellation is requested for siblings and the first
405   
    awaitable throws, cancellation is requested for siblings and the first
516   
    exception is rethrown after all awaitables complete.
406   
    exception is rethrown after all awaitables complete.
517  
407  
518   
    @li All child awaitables run concurrently on the caller's executor
408   
    @li All child awaitables run concurrently on the caller's executor
519   
    @li Results are returned as a tuple in input order
409   
    @li Results are returned as a tuple in input order
520 - 
    @li Void-returning awaitables contribute std::monostate to the
410 + 
    @li Void-returning awaitables do not contribute to the result tuple
521 - 
        result tuple, preserving the task-index-to-result-index mapping
411 + 
    @li If all awaitables return void, `when_all` returns `task<void>`
522   
    @li First exception wins; subsequent exceptions are discarded
412   
    @li First exception wins; subsequent exceptions are discarded
523   
    @li Stop is requested for siblings on first error
413   
    @li Stop is requested for siblings on first error
524   
    @li Completes only after all children have finished
414   
    @li Completes only after all children have finished
525  
415  
526   
    @par Thread Safety
416   
    @par Thread Safety
527   
    The returned task must be awaited from a single execution context.
417   
    The returned task must be awaited from a single execution context.
528   
    Child awaitables execute concurrently but complete through the caller's
418   
    Child awaitables execute concurrently but complete through the caller's
529   
    executor.
419   
    executor.
530  
420  
531   
    @param awaitables The awaitables to execute concurrently. Each must
421   
    @param awaitables The awaitables to execute concurrently. Each must
532   
        satisfy @ref IoAwaitable and is consumed (moved-from) when
422   
        satisfy @ref IoAwaitable and is consumed (moved-from) when
533   
        `when_all` is awaited.
423   
        `when_all` is awaited.
534  
424  
535 - 
    @return A task yielding a tuple of results in input order. Void tasks
425 + 
    @return A task yielding a tuple of non-void results. Returns
536 - 
        contribute std::monostate to preserve index correspondence.
426 + 
        `task<void>` when all input awaitables return void.
537  
427  
538   
    @par Example
428   
    @par Example
539  
429  
540   
    @code
430   
    @code
541   
    task<> example()
431   
    task<> example()
542   
    {
432   
    {
543   
        // Concurrent fetch, results collected in order
433   
        // Concurrent fetch, results collected in order
544   
        auto [user, posts] = co_await when_all(
434   
        auto [user, posts] = co_await when_all(
545   
            fetch_user( id ),      // task<User>
435   
            fetch_user( id ),      // task<User>
546   
            fetch_posts( id )      // task<std::vector<Post>>
436   
            fetch_posts( id )      // task<std::vector<Post>>
547   
        );
437   
        );
548  
438  
549 - 
        // Void awaitables contribute monostate
439 + 
        // Void awaitables don't contribute to result
550 - 
        auto [a, _, b] = co_await when_all(
440 + 
        co_await when_all(
551 - 
            fetch_int(),           // task<int>
441 + 
            log_event( "start" ),  // task<void>
552 - 
            log_event( "start" ),  // task<void>  → monostate
442 + 
            notify_user( id )      // task<void>
553 - 
            fetch_str()            // task<string>
 
554   
        );
443   
        );
555 - 
        // a is int, _ is monostate, b is string
444 + 
        // Returns task<void>, no result tuple
556   
    }
445   
    }
557   
    @endcode
446   
    @endcode
558  
447  
559   
    @see IoAwaitable, task
448   
    @see IoAwaitable, task
560   
*/
449   
*/
561   
template<IoAwaitable... As>
450   
template<IoAwaitable... As>
562   
[[nodiscard]] auto when_all(As... awaitables)
451   
[[nodiscard]] auto when_all(As... awaitables)
563 - 
    -> task<when_all_result_t<awaitable_result_t<As>...>>
452 + 
    -> task<non_void_tuple_t<awaitable_result_t<As>...>>
564   
{
453   
{
 
454 + 
    using result_type = non_void_tuple_t<awaitable_result_t<As>...>;
 
455 +
565   
    // State is stored in the coroutine frame, using the frame allocator
456   
    // State is stored in the coroutine frame, using the frame allocator
566   
    detail::when_all_state<awaitable_result_t<As>...> state;
457   
    detail::when_all_state<awaitable_result_t<As>...> state;
567  
458  
568   
    // Store awaitables in the frame
459   
    // Store awaitables in the frame
569   
    std::tuple<As...> awaitable_tuple(std::move(awaitables)...);
460   
    std::tuple<As...> awaitable_tuple(std::move(awaitables)...);
570  
461  
571   
    // Launch all awaitables and wait for completion
462   
    // Launch all awaitables and wait for completion
572   
    co_await detail::when_all_launcher<As...>(&awaitable_tuple, &state);
463   
    co_await detail::when_all_launcher<As...>(&awaitable_tuple, &state);
573  
464  
574   
    // Propagate first exception if any.
465   
    // Propagate first exception if any.
575   
    // Safe without explicit acquire: capture_exception() is sequenced-before
466   
    // Safe without explicit acquire: capture_exception() is sequenced-before
576   
    // signal_completion()'s acq_rel fetch_sub, which synchronizes-with the
467   
    // signal_completion()'s acq_rel fetch_sub, which synchronizes-with the
577   
    // last task's decrement that resumes this coroutine.
468   
    // last task's decrement that resumes this coroutine.
578 - 
    if(state.core_.first_exception_)
469 + 
    if(state.first_exception_)
579 - 
        std::rethrow_exception(state.core_.first_exception_);
470 + 
        std::rethrow_exception(state.first_exception_);
580 -
 
581 - 
    co_return detail::extract_results(state);
 
582 - 
}
 
583 -
 
584 - 
/** Execute a range of awaitables concurrently and collect their results.
 
585 -
 
586 - 
    Launches all awaitables in the range simultaneously and waits for all
 
587 - 
    to complete. Results are collected in a vector preserving input order.
 
588 - 
    If any awaitable throws, cancellation is requested for siblings and
 
589 - 
    the first exception is rethrown after all awaitables complete.
 
590 -
 
591 - 
    @li All child awaitables run concurrently on the caller's executor
 
592 - 
    @li Results are returned as a vector in input order
 
593 - 
    @li First exception wins; subsequent exceptions are discarded
 
594 - 
    @li Stop is requested for siblings on first error
 
595 - 
    @li Completes only after all children have finished
 
596 -
 
597 - 
    @par Thread Safety
 
598 - 
    The returned task must be awaited from a single execution context.
 
599 - 
    Child awaitables execute concurrently but complete through the caller's
 
600 - 
    executor.
 
601 -
 
602 - 
    @param awaitables Range of awaitables to execute concurrently (must
 
603 - 
        not be empty). Each element must satisfy @ref IoAwaitable and is
 
604 - 
        consumed (moved-from) when `when_all` is awaited.
 
605 -
 
606 - 
    @return A task yielding a vector where each element is the result of
 
607 - 
        the corresponding awaitable, in input order.
 
608 -
 
609 - 
    @throws std::invalid_argument if range is empty (thrown before
 
610 - 
        coroutine suspends).
 
611 - 
    @throws Rethrows the first child exception after all children
 
612 - 
        complete.
 
613 -
 
614 - 
    @par Example
 
615 - 
    @code
 
616 - 
    task<void> example()
 
617 - 
    {
 
618 - 
        std::vector<task<Response>> requests;
 
619 - 
        for (auto const& url : urls)
 
620 - 
            requests.push_back(fetch(url));
 
621 -
 
622 - 
        auto responses = co_await when_all(std::move(requests));
 
623 - 
    }
 
624 - 
    @endcode
 
625 -
 
626 - 
    @see IoAwaitableRange, when_all
 
627 - 
*/
 
628 - 
template<IoAwaitableRange R>
 
629 - 
    requires (!std::is_void_v<awaitable_result_t<std::ranges::range_value_t<R>>>)
 
630 - 
[[nodiscard]] auto when_all(R&& awaitables)
 
631 - 
    -> task<std::vector<awaitable_result_t<std::ranges::range_value_t<R>>>>
 
632 - 
{
 
633 - 
    using Awaitable = std::ranges::range_value_t<R>;
 
634 - 
    using T = awaitable_result_t<Awaitable>;
 
635 - 
    using OwnedRange = std::remove_cvref_t<R>;
 
636 -
 
637 - 
    auto count = std::ranges::size(awaitables);
 
638 - 
    if(count == 0)
 
639 - 
        throw std::invalid_argument("when_all requires at least one awaitable");
 
640 -
 
641 - 
    OwnedRange owned_awaitables = std::forward<R>(awaitables);
 
642 -
 
643 - 
    detail::when_all_homogeneous_state<T> state(count);
 
644 -
 
645 - 
    co_await detail::when_all_homogeneous_launcher<OwnedRange>(
 
646 - 
        &owned_awaitables, &state);
 
647 -
 
648 - 
    if(state.core_.first_exception_)
 
649 - 
        std::rethrow_exception(state.core_.first_exception_);
 
650 -
 
651 - 
    std::vector<T> results;
 
652 - 
    results.reserve(count);
 
653 - 
    for(auto& opt : state.results_)
 
654 - 
        results.push_back(std::move(*opt));
 
655 -
 
656 - 
    co_return results;
 
657 - 
}
 
658 -
 
659 - 
/** Execute a range of void awaitables concurrently.
 
660 -
 
661 - 
    Launches all awaitables in the range simultaneously and waits for all
 
662 - 
    to complete. Since all awaitables return void, no results are collected.
 
663 - 
    If any awaitable throws, cancellation is requested for siblings and
 
664 - 
    the first exception is rethrown after all awaitables complete.
 
665 -
 
666 - 
    @li All child awaitables run concurrently on the caller's executor
 
667 - 
    @li First exception wins; subsequent exceptions are discarded
 
668 - 
    @li Stop is requested for siblings on first error
 
669 - 
    @li Completes only after all children have finished
 
670 -
 
671 - 
    @par Thread Safety
 
672 - 
    The returned task must be awaited from a single execution context.
 
673 - 
    Child awaitables execute concurrently but complete through the caller's
 
674 - 
    executor.
 
675 -
 
676 - 
    @param awaitables Range of void awaitables to execute concurrently
 
677 - 
        (must not be empty).
 
678 -
 
679 - 
    @throws std::invalid_argument if range is empty (thrown before
 
680 - 
        coroutine suspends).
 
681 - 
    @throws Rethrows the first child exception after all children
 
682 - 
        complete.
 
683 -
 
684 - 
    @par Example
 
685 - 
    @code
 
686 - 
    task<void> example()
 
687 - 
    {
 
688 - 
        std::vector<task<void>> jobs;
 
689 - 
        for (int i = 0; i < n; ++i)
 
690 - 
            jobs.push_back(process(i));
 
691 -
 
692 - 
        co_await when_all(std::move(jobs));
 
693 - 
    }
 
694 - 
    @endcode
 
695 -
 
696 - 
    @see IoAwaitableRange, when_all
 
697 - 
*/
 
698 - 
template<IoAwaitableRange R>
 
699 - 
    requires std::is_void_v<awaitable_result_t<std::ranges::range_value_t<R>>>
 
700 - 
[[nodiscard]] auto when_all(R&& awaitables) -> task<void>
 
701 - 
{
 
702 - 
    using OwnedRange = std::remove_cvref_t<R>;
 
703 -
 
704 - 
    auto count = std::ranges::size(awaitables);
 
705 - 
    if(count == 0)
 
706 - 
        throw std::invalid_argument("when_all requires at least one awaitable");
 
707 -
 
708 - 
    OwnedRange owned_awaitables = std::forward<R>(awaitables);
 
709 -
 
710 - 
    detail::when_all_homogeneous_state<void> state(count);
 
711 -
 
712 - 
    co_await detail::when_all_homogeneous_launcher<OwnedRange>(
 
713 - 
        &owned_awaitables, &state);
 
714  
471  
715 - 
    if(state.core_.first_exception_)
472 + 
    // Extract and return results
716 - 
        std::rethrow_exception(state.core_.first_exception_);
473 + 
    if constexpr (std::is_void_v<result_type>)
 
474 + 
        co_return;
 
475 + 
    else
 
476 + 
        co_return detail::extract_results(state);
717   
}
477   
}
718  
478  
719   
} // namespace capy
479   
} // namespace capy
720   
} // namespace boost
480   
} // namespace boost
721  
481  
722   
#endif
482   
#endif