inline_pool_backend.hpp

Go to the documentation of this file.

#pragma once
 
#include "../../expected.hpp"
#include "../callable/bind.hpp"
#include "backend.hpp"
#include "worker_context_guard.hpp"
#include <functional>
#include <future>
#include <system_error>
#include <type_traits>
#include <vector>
 
namespace threadschedule::detail
{
 
class inline_pool_backend
{
public:
  explicit inline_pool_backend(size_t /*num_threads*/ = 0) {}
 
  template <typename F, typename... Args>
  auto
  submit(F&& f, Args&&... args) -> std::future<bind_result_t<F, Args...>>
  {
    auto result = try_submit(std::forward<F>(f), std::forward<Args>(args)...);
    if (!result.has_value())
      throw std::runtime_error("inline_pool_backend is shut down");
    return std::move(result.value());
  }
 
  template <typename F, typename... Args>
  auto
  try_submit(F&& f, Args&&... args) -> expected<std::future<bind_result_t<F, Args...>>, std::error_code>
  {
    using return_type = bind_result_t<F, Args...>;
    auto bound = bind_args(std::forward<F>(f), std::forward<Args>(args)...);
    if (stop_)
      return unexpected(std::make_error_code(std::errc::operation_canceled));
 
    std::promise<return_type> p;
    auto future = p.get_future();
    try
      {
        if constexpr (std::is_void_v<return_type>)
          {
            bound();
            p.set_value();
          }
        else
          {
            p.set_value(bound());
          }
      }
    catch (...)
      {
        p.set_exception(std::current_exception());
      }
    return future;
  }
 
  template <typename F, typename... Args>
  void
  post(F&& f, Args&&... args)
  {
    auto r = try_post(std::forward<F>(f), std::forward<Args>(args)...);
    if (!r.has_value())
      throw std::runtime_error("inline_pool_backend is shut down");
  }
 
  template <typename F, typename... Args>
  auto
  try_post(F&& f, Args&&... args) -> expected<void, std::error_code>
  {
    auto bound = bind_args(std::forward<F>(f), std::forward<Args>(args)...);
    if (stop_)
      return unexpected(std::make_error_code(std::errc::operation_canceled));
    try
      {
        bound();
      }
    catch (...)
      {
      }
    return {};
  }
 
  template <typename Iterator>
  auto
  try_submit_batch(Iterator begin, Iterator end) -> expected<std::vector<std::future<void>>, std::error_code>
  {
    if (stop_)
      return unexpected(std::make_error_code(std::errc::operation_canceled));
 
    std::vector<std::future<void>> futures;
    for (auto it = begin; it != end; ++it)
      {
        std::promise<void> p;
        futures.push_back(p.get_future());
        try
          {
            (*it)();
            p.set_value();
          }
        catch (...)
          {
            p.set_exception(std::current_exception());
          }
      }
    return futures;
  }
 
  template <typename Iterator>
  auto
  submit_batch(Iterator begin, Iterator end) -> std::vector<std::future<void>>
  {
    auto result = try_submit_batch(begin, end);
    if (!result.has_value())
      throw std::runtime_error("inline_pool_backend is shut down");
    return std::move(result.value());
  }
 
  template <typename Iterator, typename F>
  void
  parallel_for_each(Iterator begin, Iterator end, F&& func)
  {
    static_assert(detail::is_forward_iterator_v<Iterator>, "parallel_for_each requires at least a forward iterator");
    if (stop_)
      throw std::system_error(std::make_error_code(std::errc::operation_canceled),
                              "inline_pool_backend::parallel_for_each");
    for (auto it = begin; it != end; ++it)
      func(*it);
  }
 
  [[nodiscard]] auto
  size() const noexcept -> size_t
  {
    return 0;
  }
  [[nodiscard]] auto
  pending_tasks() const noexcept -> size_t
  {
    return 0;
  }
 
  void
  wait_for_tasks()
  {
  }
  void
  shutdown(shutdown_policy_backend /*policy*/ = shutdown_policy_backend::drain)
  {
    stop_ = true;
  }
 
private:
  bool stop_{ false };
};
 
} // namespace threadschedule::detail