when_helpers.h

Go to the documentation of this file.

// BSD 3-Clause License
// Copyright (c) 2024, 🍀☀🌕🌥 🌊
// See the LICENSE file in the project root for full license information.
 
#pragma once
 
#include <atomic>
#include <future>
#include <memory>
#include <thread>
#include <tuple>
#include <type_traits>
#include <utility>
#include <vector>
 
namespace kcenon::thread {
 
namespace detail {
 
template<typename Future>
struct future_value_type {
    using type = decltype(std::declval<Future>().get());
};
 
template<typename Future>
using future_value_type_t = typename future_value_type<Future>::type;
 
template<std::size_t... Is, typename Tuple, typename ResultTuple>
void get_all_impl(std::index_sequence<Is...>, Tuple& futures, ResultTuple& results) {
    ((std::get<Is>(results) = std::get<Is>(futures).get()), ...);
}
 
} // namespace detail
 
template<typename... Futures>
auto when_all(Futures&&... futures)
    -> std::future<std::tuple<detail::future_value_type_t<std::decay_t<Futures>>...>>
{
    using result_tuple = std::tuple<detail::future_value_type_t<std::decay_t<Futures>>...>;
 
    auto promise = std::make_shared<std::promise<result_tuple>>();
    auto result_future = promise->get_future();
 
    // Store futures in a tuple for processing
    auto futures_tuple = std::make_shared<std::tuple<std::decay_t<Futures>...>>(
        std::forward<Futures>(futures)...);
 
    // Launch async operation to wait for all
    std::thread([promise, futures_tuple]() mutable {
        try {
            result_tuple results;
            detail::get_all_impl(
                std::make_index_sequence<sizeof...(Futures)>{},
                *futures_tuple,
                results);
            promise->set_value(std::move(results));
        } catch (...) {
            promise->set_exception(std::current_exception());
        }
    }).detach();
 
    return result_future;
}
 
inline auto when_all_empty() -> std::future<std::tuple<>>
{
    std::promise<std::tuple<>> promise;
    promise.set_value(std::tuple<>{});
    return promise.get_future();
}
 
template<typename T>
auto when_any(std::vector<std::future<T>>&& futures)
    -> std::future<T>
{
    if (futures.empty()) {
        std::promise<T> promise;
        promise.set_exception(
            std::make_exception_ptr(std::invalid_argument("Empty futures vector"))
        );
        return promise.get_future();
    }
 
    auto promise = std::make_shared<std::promise<T>>();
    auto result_future = promise->get_future();
    auto completed = std::make_shared<std::atomic<bool>>(false);
    auto futures_ptr = std::make_shared<std::vector<std::future<T>>>(std::move(futures));
 
    // Launch threads for each future
    for (std::size_t i = 0; i < futures_ptr->size(); ++i) {
        std::thread([promise, completed, futures_ptr, i]() {
            try {
                auto& future = (*futures_ptr)[i];
                T result = future.get();
 
                bool expected = false;
                if (completed->compare_exchange_strong(expected, true)) {
                    promise->set_value(std::move(result));
                }
            } catch (...) {
                bool expected = false;
                if (completed->compare_exchange_strong(expected, true)) {
                    promise->set_exception(std::current_exception());
                }
            }
        }).detach();
    }
 
    return result_future;
}
 
inline auto when_any(std::vector<std::future<void>>&& futures)
    -> std::future<void>
{
    if (futures.empty()) {
        std::promise<void> promise;
        promise.set_exception(
            std::make_exception_ptr(std::invalid_argument("Empty futures vector"))
        );
        return promise.get_future();
    }
 
    auto promise = std::make_shared<std::promise<void>>();
    auto result_future = promise->get_future();
    auto completed = std::make_shared<std::atomic<bool>>(false);
    auto futures_ptr = std::make_shared<std::vector<std::future<void>>>(std::move(futures));
 
    for (std::size_t i = 0; i < futures_ptr->size(); ++i) {
        std::thread([promise, completed, futures_ptr, i]() {
            try {
                (*futures_ptr)[i].get();
 
                bool expected = false;
                if (completed->compare_exchange_strong(expected, true)) {
                    promise->set_value();
                }
            } catch (...) {
                bool expected = false;
                if (completed->compare_exchange_strong(expected, true)) {
                    promise->set_exception(std::current_exception());
                }
            }
        }).detach();
    }
 
    return result_future;
}
 
template<typename T>
auto when_any_with_index(std::vector<std::future<T>>&& futures)
    -> std::future<std::pair<std::size_t, T>>
{
    if (futures.empty()) {
        std::promise<std::pair<std::size_t, T>> promise;
        promise.set_exception(
            std::make_exception_ptr(std::invalid_argument("Empty futures vector"))
        );
        return promise.get_future();
    }
 
    auto promise = std::make_shared<std::promise<std::pair<std::size_t, T>>>();
    auto result_future = promise->get_future();
    auto completed = std::make_shared<std::atomic<bool>>(false);
    auto futures_ptr = std::make_shared<std::vector<std::future<T>>>(std::move(futures));
 
    for (std::size_t i = 0; i < futures_ptr->size(); ++i) {
        std::thread([promise, completed, futures_ptr, i]() {
            try {
                auto& future = (*futures_ptr)[i];
                T result = future.get();
 
                bool expected = false;
                if (completed->compare_exchange_strong(expected, true)) {
                    promise->set_value(std::make_pair(i, std::move(result)));
                }
            } catch (...) {
                bool expected = false;
                if (completed->compare_exchange_strong(expected, true)) {
                    promise->set_exception(std::current_exception());
                }
            }
        }).detach();
    }
 
    return result_future;
}
 
} // namespace kcenon::thread