runtime.hpp

#pragma once
 
#include "executor.hpp"
 
namespace tf {
 
// ------------------------------------------------------------------------------------------------
// class: Runtime
// ------------------------------------------------------------------------------------------------

class Runtime {
 
  friend class Executor;
  friend class FlowBuilder;
  friend class PreemptionGuard;
  friend class Algorithm;
  
  public:
  
  Executor& executor();
  
  inline Worker& worker();

  void schedule(Task task);
 
  // ----------------------------------------------------------------------------------------------
  // async methods
  // ----------------------------------------------------------------------------------------------

  template <typename F>
  auto async(F&& f);
  
  template <typename P, typename F>
  auto async(P&& params, F&& f);
  
  // ----------------------------------------------------------------------------------------------
  // silent async methods
  // ----------------------------------------------------------------------------------------------

  template <typename F>
  void silent_async(F&& f);
  
  template <typename P, typename F>
  void silent_async(P&& params, F&& f);
  
  // ----------------------------------------------------------------------------------------------
  // dependent async methods
  // ----------------------------------------------------------------------------------------------
  
  template <typename F, typename... Tasks>
  requires (std::same_as<std::decay_t<Tasks>, AsyncTask> && ...)
  auto dependent_async(F&& func, Tasks&&... tasks);
  
  template <TaskParameters P, typename F, typename... Tasks>
  requires (std::same_as<std::decay_t<Tasks>, AsyncTask> && ...)
  auto dependent_async(P&& params, F&& func, Tasks&&... tasks);
  
  template <typename F, typename I>
  requires (!std::same_as<std::decay_t<I>, AsyncTask>)
  auto dependent_async(F&& func, I first, I last);
  
  template <TaskParameters P, typename F, typename I>
  requires (!std::same_as<std::decay_t<I>, AsyncTask>)
  auto dependent_async(P&& params, F&& func, I first, I last);
  
  // ----------------------------------------------------------------------------------------------
  // silent dependent async methods
  // ----------------------------------------------------------------------------------------------

  template <typename F, typename... Tasks>
  requires (std::same_as<std::decay_t<Tasks>, AsyncTask> && ...)
  tf::AsyncTask silent_dependent_async(F&& func, Tasks&&... tasks);
  
  template <TaskParameters P, typename F, typename... Tasks>
  requires (std::same_as<std::decay_t<Tasks>, AsyncTask> && ...)
  tf::AsyncTask silent_dependent_async(P&& params, F&& func, Tasks&&... tasks);
  
  template <typename F, typename I>
  requires (!std::same_as<std::decay_t<I>, AsyncTask>)
  tf::AsyncTask silent_dependent_async(F&& func, I first, I last);
  
  template <TaskParameters P, typename F, typename I>
  requires (!std::same_as<std::decay_t<I>, AsyncTask>)
  tf::AsyncTask silent_dependent_async(P&& params, F&& func, I first, I last);
 
 
 
  // ----------------------------------------------------------------------------------------------
  // cooperative execution methods
  // ----------------------------------------------------------------------------------------------
  
  void corun();

  void corun_all();

  bool is_cancelled();
 
  private:

  explicit Runtime(Executor&, Worker&, Node*);
  
  Executor& _executor;
  
  Worker& _worker;
  
  Node* _node;
};
 
// constructor
inline Runtime::Runtime(Executor& executor, Worker& worker, Node* node) :
  _executor {executor},
  _worker   {worker},
  _node     {node} {
}
 
// Function: executor
inline Executor& Runtime::executor() {
  return _executor;
}
 
// Function: worker
inline Worker& Runtime::worker() {
  return _worker;
}
 
// Procedure: schedule
inline void Runtime::schedule(Task task) {
 
  auto node = task._node;
  // need to keep the invariant: when scheduling a task, the task must have
  // zero dependency (join counter is 0)
  // or we can encounter bug when inserting a nested flow (e.g., module task)
  node->_join_counter.store(0, std::memory_order_relaxed);
 
  auto& j = node->_parent ? node->_parent->_join_counter :
                            node->_topology->_join_counter;
  j.fetch_add(1, std::memory_order_relaxed);
  _executor._schedule(_worker, node);
}
 
// Function: corun
inline void Runtime::corun() {
  {
    ExplicitAnchorGuard anchor(_node);
    _executor._corun_until(_worker, [this] () -> bool {
      return _node->_join_counter.load(std::memory_order_acquire) == 1;
    });
  }
  _node->_rethrow_exception();
}
 
// Function: corun_all
inline void Runtime::corun_all() {
  corun();
}
 
inline bool Runtime::is_cancelled() { 
  return _node->_is_parent_cancelled(); 
}
 
// ------------------------------------------------------------------------------------------------
// Runtime::silent_async
// ------------------------------------------------------------------------------------------------
 
// Function: silent_async
template <typename F>
void Runtime::silent_async(F&& f) {
  silent_async(DefaultTaskParams{}, std::forward<F>(f));
}
 
// Function: silent_async
template <typename P, typename F>
void Runtime::silent_async(P&& params, F&& f) {
  _node->_join_counter.fetch_add(1, std::memory_order_relaxed);
  _executor._silent_async(
    std::forward<P>(params), std::forward<F>(f), _node->_topology, _node
  );
}
 
// ------------------------------------------------------------------------------------------------
// Runtime::async 
// ------------------------------------------------------------------------------------------------
 
// Function: async
template <typename F>
auto Runtime::async(F&& f) {
  return async(DefaultTaskParams{}, std::forward<F>(f));
}
 
// Function: async
template <typename P, typename F>
auto Runtime::async(P&& params, F&& f) {
  _node->_join_counter.fetch_add(1, std::memory_order_relaxed);
  return _executor._async(
    std::forward<P>(params), std::forward<F>(f), _node->_topology, _node
  );
}
 
// ------------------------------------------------------------------------------------------------
// silent dependent async
// ------------------------------------------------------------------------------------------------
 
// Function: silent_dependent_async
template <typename F, typename... Tasks>
requires (std::same_as<std::decay_t<Tasks>, AsyncTask> && ...)
tf::AsyncTask Runtime::silent_dependent_async(F&& func, Tasks&&... tasks) {
  return silent_dependent_async(
    DefaultTaskParams{}, std::forward<F>(func), std::forward<Tasks>(tasks)...
  );
}
 
// Function: silent_dependent_async
template <TaskParameters P, typename F, typename... Tasks>
requires (std::same_as<std::decay_t<Tasks>, AsyncTask> && ...)
tf::AsyncTask Runtime::silent_dependent_async(
  P&& params, F&& func, Tasks&&... tasks 
){
  std::array<AsyncTask, sizeof...(Tasks)> array = { std::forward<Tasks>(tasks)... };
  return silent_dependent_async(
    std::forward<P>(params), std::forward<F>(func), array.begin(), array.end()
  );
}
 
// Function: silent_dependent_async
template <typename F, typename I>
requires (!std::same_as<std::decay_t<I>, AsyncTask>)
tf::AsyncTask Runtime::silent_dependent_async(F&& func, I first, I last) {
  return silent_dependent_async(DefaultTaskParams{}, std::forward<F>(func), first, last);
}
 
// Function: silent_dependent_async
template <TaskParameters P, typename F, typename I>
requires (!std::same_as<std::decay_t<I>, AsyncTask>)
tf::AsyncTask Runtime::silent_dependent_async(
  P&& params, F&& func, I first, I last
) {
  _node->_join_counter.fetch_add(1, std::memory_order_relaxed);
  return _executor._silent_dependent_async(
    std::forward<P>(params), std::forward<F>(func), first, last, _node->_topology, _node
  );
}
 
// ------------------------------------------------------------------------------------------------
// dependent async
// ------------------------------------------------------------------------------------------------
 
// Function: dependent_async
template <typename F, typename... Tasks>
requires (std::same_as<std::decay_t<Tasks>, AsyncTask> && ...)
auto Runtime::dependent_async(F&& func, Tasks&&... tasks) {
  return dependent_async(DefaultTaskParams{}, std::forward<F>(func), std::forward<Tasks>(tasks)...);
}
 
// Function: dependent_async
template <TaskParameters P, typename F, typename... Tasks>
requires (std::same_as<std::decay_t<Tasks>, AsyncTask> && ...)
auto Runtime::dependent_async(P&& params, F&& func, Tasks&&... tasks) {
  std::array<AsyncTask, sizeof...(Tasks)> array = { std::forward<Tasks>(tasks)... };
  return dependent_async(
    std::forward<P>(params), std::forward<F>(func), array.begin(), array.end()
  );
}
 
// Function: dependent_async
template <typename F, typename I>
requires (!std::same_as<std::decay_t<I>, AsyncTask>)
auto Runtime::dependent_async(F&& func, I first, I last) {
  return dependent_async(DefaultTaskParams{}, std::forward<F>(func), first, last);
}
 
// Function: dependent_async
template <TaskParameters P, typename F, typename I>
requires (!std::same_as<std::decay_t<I>, AsyncTask>)
auto Runtime::dependent_async(P&& params, F&& func, I first, I last) {
  _node->_join_counter.fetch_add(1, std::memory_order_relaxed);
  return _executor._dependent_async(
    std::forward<P>(params), std::forward<F>(func), first, last, _node->_topology, _node
  );
}
 
// ----------------------------------------------------------------------------
// Executor Forward Declaration
// ----------------------------------------------------------------------------
 
// Procedure: _invoke_runtime_task
inline bool Executor::_invoke_runtime_task(Worker& worker, Node* node) {
  return _invoke_runtime_task_impl(
    worker, node, std::get_if<Node::Runtime>(&node->_handle)->work
  );
}
 
// Function: _invoke_runtime_task_impl
inline bool Executor::_invoke_runtime_task_impl(
  Worker& worker, Node* node, std::function<void(Runtime&)>& work
) {
  // first time
  if((node->_nstate & NSTATE::PREEMPTED) == 0) {
 
    Runtime rt(*this, worker, node);
 
    node->_nstate |= (NSTATE::PREEMPTED | NSTATE::IMPLICITLY_ANCHORED);
 
    node->_join_counter.fetch_add(1, std::memory_order_release);
 
    _observer_prologue(worker, node);
    TF_EXECUTOR_EXCEPTION_HANDLER(worker, node, {
      work(rt);
    });
    _observer_epilogue(worker, node);
    
    // Last one to leave the runtime; no need to preempt this runtime.
    if(node->_join_counter.fetch_sub(1, std::memory_order_acq_rel) == 1) {
      node->_nstate &= ~(NSTATE::PREEMPTED | NSTATE::IMPLICITLY_ANCHORED);
    }
    // There are still child tasks running; need to preempt this runtime.
    // Here, we cannot let caller check the state from node->_nstate due to data race,
    // but return a stateless boolean to indicate preemption.
    // Ex: if preempted, another task may finish real quck and insert this parent task
    // again into the scheduling queue. When running this parent task, it will jump to
    // else branch below and modify tne nstate, thus incuring data race.
    else {
      return true;
    }
  }
  // second time - previously preempted
  else {
    node->_nstate &= ~(NSTATE::PREEMPTED | NSTATE::IMPLICITLY_ANCHORED);
  }
  return false;
}
 
// Function: _invoke_runtime_task_impl
inline bool Executor::_invoke_runtime_task_impl(
  Worker& worker, Node* node, std::function<void(Runtime&, bool)>& work
) {
    
  Runtime rt(*this, worker, node);
 
  // first time
  if((node->_nstate & NSTATE::PREEMPTED) == 0) {
    
    node->_nstate |= (NSTATE::PREEMPTED | NSTATE::IMPLICITLY_ANCHORED);
    node->_join_counter.fetch_add(1, std::memory_order_release);
 
    _observer_prologue(worker, node);
    TF_EXECUTOR_EXCEPTION_HANDLER(worker, node, {
      work(rt, false);
    });
    _observer_epilogue(worker, node);
    
    // Last one to leave this runtime; no need to preempt this runtime
    if(node->_join_counter.fetch_sub(1, std::memory_order_acq_rel) == 1) {
      node->_nstate &= ~(NSTATE::PREEMPTED | NSTATE::IMPLICITLY_ANCHORED);
    }
    // Here, we cannot let caller check the state from node->_nstate due to data race,
    // but return a stateless boolean to indicate preemption.
    // Ex: if preempted, another task may finish real quck and insert this parent task
    // again into the scheduling queue. When running this parent task, it will jump to
    // else branch below and modify tne nstate, thus incuring data race.
    else {
      return true;
    }
  }
  // second time - previously preempted
  else {
    node->_nstate &= ~(NSTATE::PREEMPTED | NSTATE::IMPLICITLY_ANCHORED);
  }
 
  // clean up outstanding work (e.g., exception)
  work(rt, true);
 
  return false;
}
 
// ------------------------------------------------------------------------------------------------
// class: NonpreemptiveRuntime (internal use only)
// ------------------------------------------------------------------------------------------------

class NonpreemptiveRuntime {
 
  friend class Executor;
 
  public:
  
  void schedule(Task task);
  
  private:

  explicit NonpreemptiveRuntime(Executor& executor, Worker& worker) :
    _executor {executor}, _worker {worker}{
  }
  
  Executor& _executor;
  
  Worker& _worker;
};
 
// Procedure: schedule
inline void NonpreemptiveRuntime::schedule(Task task) {
 
  auto node = task._node;
  // need to keep the invariant: when scheduling a task, the task must have
  // zero dependency (join counter is 0)
  // or we can encounter bug when inserting a nested flow (e.g., module task)
  node->_join_counter.store(0, std::memory_order_relaxed);
 
  auto& j = node->_parent ? node->_parent->_join_counter :
                            node->_topology->_join_counter;
  j.fetch_add(1, std::memory_order_relaxed);
  _executor._schedule(_worker, node);
}
 
// ----------------------------------------------------------------------------
// Executor Forward Declaration
// ----------------------------------------------------------------------------
 
// Procedure: _invoke_nonpreemptive_runtime_task
inline void Executor::_invoke_nonpreemptive_runtime_task(Worker& worker, Node* node) {
  _observer_prologue(worker, node);
  tf::NonpreemptiveRuntime nprt(*this, worker);
  TF_EXECUTOR_EXCEPTION_HANDLER(worker, node, {
    std::get_if<Node::NonpreemptiveRuntime>(&node->_handle)->work(nprt);
  });
  _observer_epilogue(worker, node);
}
 
 
// Function: run_until
template <typename P, typename C>
tf::Future<void> Executor::run_until(Taskflow&& f, P&& p, C&& c) {
 
  // No need to create a real topology but returns an dummy future for invariant.
  if(f.empty() || p()) {
    c();
    std::promise<void> promise;
    promise.set_value();
    return tf::Future<void>(promise.get_future());
  }
  
  _increment_topology();
 
  auto g = std::make_unique<Taskflow>(std::move(f)); 
 
  // creates a topology for this run
  auto t = std::make_shared<Topology>(*g, std::forward<P>(p), std::forward<C>(c));
  //auto t = std::make_shared<DerivedTopology<P, C>>(*g, std::forward<P>(p), std::forward<C>(c));
 
  // need to create future before the topology got torn down quickly
  tf::Future<void> future(t->_promise.get_future(), t);
 
  // creates a silent-async that holds the taskflow
  silent_async([g=MoC{std::move(g)}, t](tf::Runtime& rt) mutable {
    t->_parent = rt._node;
    t->_parent->_join_counter.fetch_add(1, std::memory_order_release);
    if(g.object->_fetch_enqueue(t) == 0) {
      rt._executor._schedule_graph(
        rt._worker, g.object->_graph, t.get(), t.get()
      );
    }
  });
 
  return future;
}
 
}  // end of namespace tf -----------------------------------------------------