futures.hpp

Go to the documentation of this file.

#pragma once

 
#include "expected.hpp"
 
#include <chrono>
#include <exception>
#include <future>
#include <random>
#include <stdexcept>
#include <thread>
#include <utility>
#include <vector>
 
namespace threadschedule
{

template <typename T>
auto when_all(std::vector<std::future<T>>& futures) -> std::vector<T>
{
    std::vector<T> results;
    results.reserve(futures.size());
    std::exception_ptr first_error;
 
    for (auto& f : futures)
    {
        try
        {
            results.push_back(f.get());
        }
        catch (...)
        {
            if (!first_error)
                first_error = std::current_exception();
        }
    }
 
    if (first_error)
        std::rethrow_exception(first_error);
 
    return results;
}

inline void when_all(std::vector<std::future<void>>& futures)
{
    std::exception_ptr first_error;
 
    for (auto& f : futures)
    {
        try
        {
            f.get();
        }
        catch (...)
        {
            if (!first_error)
                first_error = std::current_exception();
        }
    }
 
    if (first_error)
        std::rethrow_exception(first_error);
}

template <typename T>
auto when_all_settled(std::vector<std::future<T>>& futures) -> std::vector<expected<T, std::exception_ptr>>
{
    std::vector<expected<T, std::exception_ptr>> results;
    results.reserve(futures.size());
 
    for (auto& f : futures)
    {
        try
        {
            results.push_back(f.get());
        }
        catch (...)
        {
            results.push_back(unexpected(std::current_exception()));
        }
    }
 
    return results;
}

inline auto when_all_settled(std::vector<std::future<void>>& futures) -> std::vector<expected<void, std::exception_ptr>>
{
    std::vector<expected<void, std::exception_ptr>> results;
    results.reserve(futures.size());
 
    for (auto& f : futures)
    {
        try
        {
            f.get();
            results.emplace_back();
        }
        catch (...)
        {
            results.push_back(unexpected(std::current_exception()));
        }
    }
 
    return results;
}

template <typename T>
auto when_any(std::vector<std::future<T>>& futures) -> std::pair<size_t, T>
{
    if (futures.empty())
        throw std::invalid_argument("when_any: empty futures vector");
 
    thread_local std::mt19937 rng{std::random_device{}()};
    std::uniform_int_distribution<size_t> dist(0, futures.size() - 1);
    size_t const start = dist(rng);
    unsigned backoff_ms = 1;
 
    while (true)
    {
        for (size_t k = 0; k < futures.size(); ++k)
        {
            size_t const i = (start + k) % futures.size();
            if (futures[i].wait_for(std::chrono::milliseconds(1)) == std::future_status::ready)
                return {i, futures[i].get()};
        }
        std::this_thread::sleep_for(std::chrono::milliseconds(backoff_ms));
        if (backoff_ms < 16)
            backoff_ms *= 2;
    }
}

inline auto when_any(std::vector<std::future<void>>& futures) -> size_t
{
    if (futures.empty())
        throw std::invalid_argument("when_any: empty futures vector");
 
    thread_local std::mt19937 rng{std::random_device{}()};
    std::uniform_int_distribution<size_t> dist(0, futures.size() - 1);
    size_t const start = dist(rng);
    unsigned backoff_ms = 1;
 
    while (true)
    {
        for (size_t k = 0; k < futures.size(); ++k)
        {
            size_t const i = (start + k) % futures.size();
            if (futures[i].wait_for(std::chrono::milliseconds(1)) == std::future_status::ready)
            {
                futures[i].get();
                return i;
            }
        }
        std::this_thread::sleep_for(std::chrono::milliseconds(backoff_ms));
        if (backoff_ms < 16)
            backoff_ms *= 2;
    }
}
 
} // namespace threadschedule