scheduler.hpp

Go to the documentation of this file.

 
 
// see https://github.com/dbittman/waitfree-mpsc-queue/blob/master/mpsc.c
 
#pragma once
 
#include "async_task.hpp"
 
#include <concepts>
#include <cstddef>
#include <functional>
#include <memory>
#include <thread>
#include <type_traits>
#include <utility>
 
namespace uni20::async
{
 
class AsyncTask;
class TaskPromiseBase;
 
class LightweightTaskBatch {
  public:
    template <class Function>
      requires std::invocable<Function&, std::size_t>
    static auto bind(std::size_t size, Function& function) -> LightweightTaskBatch
    {
      using function_type = std::remove_reference_t<Function>;
      return LightweightTaskBatch{
          size, const_cast<void*>(static_cast<void const*>(std::addressof(function))),
          [](void* context, std::size_t index) { std::invoke(*static_cast<function_type*>(context), index); }};
    }
 
    [[nodiscard]] auto size() const noexcept -> std::size_t { return size_; }
 
    void operator()(std::size_t index) const { invoke_(context_, index); }
 
  private:
    using invoke_type = void (*)(void*, std::size_t);
 
    LightweightTaskBatch(std::size_t size, void* context, invoke_type invoke) noexcept
        : size_(size), context_(context), invoke_(invoke)
    {}
 
    std::size_t size_;
    void* context_;
    invoke_type invoke_;
};
 
class IScheduler {
  public:
    virtual ~IScheduler() = default;
 
  private:
    friend class BasicTask;
    friend class TaskPromiseBase;
 
    virtual bool accepts_route(TaskRoute route) const noexcept = 0;
 
    virtual bool can_direct_transfer(TaskHandle from, TaskHandle to) const noexcept = 0;
 
    virtual void reschedule(BasicTask&& task) = 0;
};
 
class IAsyncScheduler : public virtual IScheduler {
  public:
    ~IAsyncScheduler() override = default;
 
    virtual void schedule(AsyncTask&& task) = 0;
 
    template <class Function>
      requires std::invocable<Function&, std::size_t>
    void execute_batch(std::size_t size, Function&& function)
    {
      auto batch = LightweightTaskBatch::bind(size, function);
      this->execute_batch_impl(batch);
    }
 
    virtual void pause() = 0;
 
    virtual void resume() = 0;
 
    using WaitPredicate = std::function<bool()>;
    using WaitWakeup = std::function<void()>;
    using WaitWakeupRegistration = std::function<void(WaitWakeup)>;
 
    struct WaitRequest
    {
        WaitPredicate is_ready;
        WaitWakeupRegistration notify_when_ready{};
    };
 
    virtual void help_while_waiting(const WaitPredicate& is_ready)
    {
      while (!is_ready())
      {
        std::this_thread::yield();
      }
    }
 
    virtual void wait_for(const WaitPredicate& is_ready)
    {
      while (!is_ready())
      {
        this->help_while_waiting(is_ready);
      }
    }
 
    virtual void wait_for(WaitRequest const& request) { this->wait_for(request.is_ready); }
 
  private:
    virtual void execute_batch_impl(LightweightTaskBatch const& batch)
    {
      for (std::size_t index = 0; index < batch.size(); ++index)
        batch(index);
    }
};
 
} // namespace uni20::async