task_group.hpp

#pragma once
 
#include "executor.hpp"
 
namespace tf {
 
// ------------------------------------------------------------------------------------------------
// class: TaskGroup
// ------------------------------------------------------------------------------------------------
 
class TaskGroup {
 
  friend class Executor;
  
  public:
  
  // ----------------------------------------------------------------------------------------------
  // deleted members
  // ----------------------------------------------------------------------------------------------
 
  TaskGroup(const TaskGroup&) = delete;
  
  TaskGroup(TaskGroup&&) = delete;
 
  TaskGroup& operator = (TaskGroup&&) = delete;
 
  TaskGroup& operator = (const TaskGroup&) = delete;
  
  Executor& executor();
  
  // ----------------------------------------------------------------------------------------------
  // async methods
  // ----------------------------------------------------------------------------------------------
 
  template <typename F>
  auto async(F&& f);
  
  template <TaskParamsLike P, typename F>
  auto async(P&& params, F&& f);
  
  // ----------------------------------------------------------------------------------------------
  // silent async methods
  // ----------------------------------------------------------------------------------------------
 
  template <typename F>
  void silent_async(F&& f);
  
  template <TaskParamsLike P, typename F>
  void silent_async(P&& params, F&& f);
  
  // ----------------------------------------------------------------------------------------------
  // dependent async methods
  // ----------------------------------------------------------------------------------------------
  
  template <typename F, AsyncTaskHandleLike... Tasks>
  auto dependent_async(F&& func, Tasks&&... tasks);
  
  template <TaskParamsLike P, typename F, AsyncTaskHandleLike... Tasks>
  auto dependent_async(P&& params, F&& func, Tasks&&... tasks);
  
  template <typename F, std::input_iterator I>
  auto dependent_async(F&& func, I first, I last);
  
  template <TaskParamsLike P, typename F, std::input_iterator I>
  auto dependent_async(P&& params, F&& func, I first, I last);
 
  
  // ----------------------------------------------------------------------------------------------
  // silent dependent async methods
  // ----------------------------------------------------------------------------------------------
 
  template <typename F, AsyncTaskHandleLike... Tasks>
  tf::AsyncTask silent_dependent_async(F&& func, Tasks&&... tasks);
  
  template <TaskParamsLike P, typename F, AsyncTaskHandleLike... Tasks>
  tf::AsyncTask silent_dependent_async(P&& params, F&& func, Tasks&&... tasks);
  
  template <typename F, std::input_iterator I>
  tf::AsyncTask silent_dependent_async(F&& func, I first, I last);
  
  template <TaskParamsLike P, typename F, std::input_iterator I>
  tf::AsyncTask silent_dependent_async(P&& params, F&& func, I first, I last);
 
 
  // ----------------------------------------------------------------------------------------------
  // cooperative execution methods
  // ----------------------------------------------------------------------------------------------
  
  void corun();
  
  void cancel();
  
  bool is_cancelled();
  
  size_t size() const;
 
  private:
 
  explicit TaskGroup(Executor&, Worker&);
  
  Executor& _executor;
 
  Worker& _worker;
  
  NodeBase _node_base;
};
 
// constructor
inline TaskGroup::TaskGroup(Executor& executor, Worker& worker) : 
  _executor  {executor}, 
  _worker    {worker}, 
  _node_base {NSTATE::IMPLICITLY_ANCHORED, ESTATE::NONE, nullptr, 0} {
}
 
// Function: executor
inline Executor& TaskGroup::executor() {
  return _executor;
}
 
// Function: corun
inline void TaskGroup::corun() {
  {
    ExplicitAnchorGuard anchor(&_node_base);
    _executor._corun_until(_worker, [this] () -> bool {
      return _node_base._join_counter.load(std::memory_order_acquire) == 0;
    });
  }
  _node_base._rethrow_exception();
}
 
// Function: cancel
inline void TaskGroup::cancel() {
  _node_base._estate.fetch_or(ESTATE::CANCELLED, std::memory_order_relaxed);
}
 
// Function: is_cancelled
inline bool TaskGroup::is_cancelled() { 
  return _node_base._estate.load(std::memory_order_relaxed) & ESTATE::CANCELLED;
}
 
// Function: size
inline size_t TaskGroup::size() const {
  return _node_base._join_counter.load(std::memory_order_relaxed);
}
 
// ------------------------------------------------------------------------------------------------
// TaskGroup::silent_async
// ------------------------------------------------------------------------------------------------
 
// Function: silent_async
template <typename F>
void TaskGroup::silent_async(F&& f) {
  silent_async(DefaultTaskParams{}, std::forward<F>(f));
}
 
// Function: silent_async
template <TaskParamsLike P, typename F>
void TaskGroup::silent_async(P&& params, F&& f) {
  _node_base._join_counter.fetch_add(1, std::memory_order_relaxed);
  _executor._silent_async(
    std::forward<P>(params), std::forward<F>(f), nullptr, &_node_base
  );
}
 
// ------------------------------------------------------------------------------------------------
// TaskGroup::async 
// ------------------------------------------------------------------------------------------------
 
// Function: async
template <typename F>
auto TaskGroup::async(F&& f) {
  return async(DefaultTaskParams{}, std::forward<F>(f));
}
 
// Function: async
template <TaskParamsLike P, typename F>
auto TaskGroup::async(P&& params, F&& f) {
  _node_base._join_counter.fetch_add(1, std::memory_order_relaxed);
  return _executor._async(
    std::forward<P>(params), std::forward<F>(f), nullptr, &_node_base
  );
}
 
// ------------------------------------------------------------------------------------------------
// silent dependent async
// ------------------------------------------------------------------------------------------------
 
// Function: silent_dependent_async
template <typename F, AsyncTaskHandleLike... Tasks>
tf::AsyncTask TaskGroup::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 <TaskParamsLike P, typename F, AsyncTaskHandleLike... Tasks>
tf::AsyncTask TaskGroup::silent_dependent_async(
  P&& params, F&& func, Tasks&&... tasks 
){
  std::array<AsyncTask*, sizeof...(Tasks)> array = { (&tasks)... };
  return silent_dependent_async(
    std::forward<P>(params), std::forward<F>(func), array.begin(), array.end()
  );
}
 
// Function: silent_dependent_async
template <typename F, std::input_iterator I>
tf::AsyncTask TaskGroup::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 <TaskParamsLike P, typename F, std::input_iterator I>
tf::AsyncTask TaskGroup::silent_dependent_async(
  P&& params, F&& func, I first, I last
) {
  _node_base._join_counter.fetch_add(1, std::memory_order_relaxed);
  return _executor._silent_dependent_async(
    std::forward<P>(params), std::forward<F>(func), first, last, nullptr, &_node_base
  );
}
 
// ------------------------------------------------------------------------------------------------
// dependent async
// ------------------------------------------------------------------------------------------------
 
// Function: dependent_async
template <typename F, AsyncTaskHandleLike... Tasks>
auto TaskGroup::dependent_async(F&& func, Tasks&&... tasks) {
  return dependent_async(DefaultTaskParams{}, std::forward<F>(func), std::forward<Tasks>(tasks)...);
}
 
// Function: dependent_async
template <TaskParamsLike P, typename F, AsyncTaskHandleLike... Tasks>
auto TaskGroup::dependent_async(P&& params, F&& func, Tasks&&... tasks) {
  std::array<AsyncTask*, sizeof...(Tasks)> array = { (&tasks)... };
  return dependent_async(
    std::forward<P>(params), std::forward<F>(func), array.begin(), array.end()
  );
}
 
// Function: dependent_async
template <typename F, std::input_iterator I>
auto TaskGroup::dependent_async(F&& func, I first, I last) {
  return dependent_async(DefaultTaskParams{}, std::forward<F>(func), first, last);
}
 
// Function: dependent_async
template <TaskParamsLike P, typename F, std::input_iterator I>
auto TaskGroup::dependent_async(P&& params, F&& func, I first, I last) {
  _node_base._join_counter.fetch_add(1, std::memory_order_relaxed);
  return _executor._dependent_async(
    std::forward<P>(params), std::forward<F>(func), first, last, nullptr, &_node_base
  );
}
 
// ----------------------------------------------------------------------------
// Executor Forward Declaration
// ----------------------------------------------------------------------------
 
// Procedure: task_group
inline TaskGroup Executor::task_group() {
  Worker* w = this_worker();
  if(w == nullptr) {
    TF_THROW("task_group can only created by a worker of the executor");
  }
  return TaskGroup(*this, *w);
}
 
 
}  // end of namespace tf -----------------------------------------------------