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::advanced
{
 
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)
    {
      size_t deferred = futures.size();
      for (size_t k = 0; k < futures.size(); ++k)
        {
          size_t const i = (start + k) % futures.size();
          auto const status = futures[i].wait_for(std::chrono::milliseconds(1));
          if (status == std::future_status::ready)
            return { i, futures[i].get() };
          if (status == std::future_status::deferred && deferred == futures.size())
            deferred = i;
        }
      if (deferred != futures.size())
        return { deferred, futures[deferred].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)
    {
      size_t deferred = futures.size();
      for (size_t k = 0; k < futures.size(); ++k)
        {
          size_t const i = (start + k) % futures.size();
          auto const status = futures[i].wait_for(std::chrono::milliseconds(1));
          if (status == std::future_status::ready)
            {
              futures[i].get();
              return i;
            }
          if (status == std::future_status::deferred && deferred == futures.size())
            deferred = i;
        }
      if (deferred != futures.size())
        {
          futures[deferred].get();
          return deferred;
        }
      std::this_thread::sleep_for(std::chrono::milliseconds(backoff_ms));
      if (backoff_ms < 16)
        backoff_ms *= 2;
    }
}
 
} // namespace threadschedule::advanced