task.hpp

#pragma once
 
#include "graph.hpp"

 
namespace tf {
 
// ----------------------------------------------------------------------------
// Task Types
// ----------------------------------------------------------------------------

enum class TaskType : int {
  PLACEHOLDER = 0,
  STATIC,
  RUNTIME,
  SUBFLOW,
  CONDITION,
  MODULE,
  ASYNC,
  UNDEFINED
};

inline constexpr std::array<TaskType, 7> TASK_TYPES = {
  TaskType::PLACEHOLDER,
  TaskType::STATIC,
  TaskType::RUNTIME,
  TaskType::SUBFLOW,
  TaskType::CONDITION,
  TaskType::MODULE,
  TaskType::ASYNC,
};

inline const char* to_string(TaskType type) {
 
  const char* val;
 
  switch(type) {
    case TaskType::PLACEHOLDER: val = "placeholder";     break;
    case TaskType::STATIC:      val = "static";          break;
    case TaskType::RUNTIME:     val = "runtime";         break;
    case TaskType::SUBFLOW:     val = "subflow";         break;
    case TaskType::CONDITION:   val = "condition";       break;
    case TaskType::MODULE:      val = "module";          break;
    case TaskType::ASYNC:       val = "async";           break;
    default:                    val = "undefined";       break;
  }
 
  return val;
}
 
// ----------------------------------------------------------------------------
// Static Task Trait
// ----------------------------------------------------------------------------

template <typename C>
concept StaticTask = std::invocable<C> &&
                     std::same_as<std::invoke_result_t<C>, void>;

template <typename C>
constexpr bool is_static_task_v = StaticTask<C>;
 
// ----------------------------------------------------------------------------
// Subflow Task Trait
// ----------------------------------------------------------------------------

template <typename C>
concept SubflowTask = std::invocable<C, tf::Subflow&> &&
                      std::same_as<std::invoke_result_t<C, tf::Subflow&>, void>;

template <typename C>
constexpr bool is_subflow_task_v = SubflowTask<C>;
 
// ----------------------------------------------------------------------------
// Runtime Task Trait
// ----------------------------------------------------------------------------

template <typename C>
concept RuntimeTask =
  (std::invocable<C, tf::Runtime&> &&
   std::same_as<std::invoke_result_t<C, tf::Runtime&>, void>) ||
  (std::invocable<C, tf::NonpreemptiveRuntime&> &&
   std::same_as<std::invoke_result_t<C, tf::NonpreemptiveRuntime&>, void>);

template <typename C>
constexpr bool is_runtime_task_v = RuntimeTask<C>;
 
 
// ----------------------------------------------------------------------------
// Condition Task Trait
// ----------------------------------------------------------------------------

template <typename C>
concept ConditionTask = std::invocable<C> &&
                        std::convertible_to<std::invoke_result_t<C>, int>;

template <typename C>
constexpr bool is_condition_task_v = ConditionTask<C>;

template <typename C>
concept MultiConditionTask = std::invocable<C> &&
                             std::same_as<std::invoke_result_t<C>, SmallVector<int>>;

template <typename C>
constexpr bool is_multi_condition_task_v = MultiConditionTask<C>;
 
 
// ----------------------------------------------------------------------------
// Task
// ----------------------------------------------------------------------------

class Task {
 
  friend class FlowBuilder;
  friend class Runtime;
  friend class NonpreemptiveRuntime;
  friend class Taskflow;
  friend class TaskView;
  friend class Executor;
 
  public:

  Task() = default;

  Task(const Task& other);

  Task& operator = (const Task& other);

  Task& operator = (std::nullptr_t);

  bool operator == (const Task& rhs) const;

  bool operator != (const Task& rhs) const;

  const std::string& name() const;

  size_t num_successors() const;

  size_t num_predecessors() const;

  size_t num_strong_dependencies() const;

  size_t num_weak_dependencies() const;

  Task& name(const std::string& name);

  template <typename C>
  Task& work(C&& callable);

  template <HasGraph T>
  Task& composed_of(T& object);
  
  Task& adopt(tf::Graph&& graph);

  template <typename... Ts>
  Task& precede(Ts&&... tasks);

  template <typename... Ts>
  Task& succeed(Ts&&... tasks);
    
  template <typename... Ts>
  Task& remove_predecessors(Ts&&... tasks);

  template <typename... Ts>
  Task& remove_successors(Ts&&... tasks);

  Task& release(Semaphore& semaphore);
  
  template <typename I>
  Task& release(I first, I last);

  Task& acquire(Semaphore& semaphore);

  template <typename I>
  Task& acquire(I first, I last);

  Task& data(void* data);
  
  void reset();

  void reset_work();

  bool empty() const;

  bool has_work() const;

  template <typename V>
  void for_each_successor(V&& visitor) const;

  template <typename V>
  void for_each_predecessor(V&& visitor) const;

  template <typename V>
  void for_each_subflow_task(V&& visitor) const;

  size_t hash_value() const;

  TaskType type() const;

  void dump(std::ostream& ostream) const;

  void* data() const;

  std::exception_ptr exception_ptr() const;
  
  bool has_exception_ptr() const;
 
  private:
 
  Task(Node*);
 
  Node* _node {nullptr};
};
 
// Constructor
inline Task::Task(Node* node) : _node {node} {
}
 
// Constructor
inline Task::Task(const Task& rhs) : _node {rhs._node} {
}
 
// Function: precede
template <typename... Ts>
Task& Task::precede(Ts&&... tasks) {
  (_node->_precede(tasks._node), ...);
  //_precede(std::forward<Ts>(tasks)...);
  return *this;
}
 
// Function: succeed
template <typename... Ts>
Task& Task::succeed(Ts&&... tasks) {
  (tasks._node->_precede(_node), ...);
  //_succeed(std::forward<Ts>(tasks)...);
  return *this;
}
 
// Function: remove_predecessors
template <typename... Ts>
Task& Task::remove_predecessors(Ts&&... tasks) {
  (tasks._node->_remove_successors(_node), ...);
  (_node->_remove_predecessors(tasks._node), ...);
  return *this;
}
 
// Function: remove_successors
template <typename... Ts>
Task& Task::remove_successors(Ts&&... tasks) {
  (_node->_remove_successors(tasks._node), ...);
  (tasks._node->_remove_predecessors(_node), ...);
  return *this;
}
 
// Function: composed_of
template <HasGraph T>
Task& Task::composed_of(T& target) {
  _node->_handle.emplace<Node::Module>(retrieve_graph(target));
  return *this;
}
 
// Function: adopt
inline Task& Task::adopt(Graph&& graph) {
  _node->_handle.emplace<Node::AdoptedModule>(std::move(graph));
  return *this;
}
 
// Operator =
inline Task& Task::operator = (const Task& rhs) {
  _node = rhs._node;
  return *this;
}
 
// Operator =
inline Task& Task::operator = (std::nullptr_t ptr) {
  _node = ptr;
  return *this;
}
 
// Operator ==
inline bool Task::operator == (const Task& rhs) const {
  return _node == rhs._node;
}
 
// Operator !=
inline bool Task::operator != (const Task& rhs) const {
  return _node != rhs._node;
}
 
// Function: name
inline Task& Task::name(const std::string& name) {
  _node->_name = name;
  return *this;
}
 
// Function: acquire
inline Task& Task::acquire(Semaphore& s) {
  if(!_node->_semaphores) {
    _node->_semaphores = std::make_unique<Node::Semaphores>();
  }
  _node->_semaphores->to_acquire.push_back(&s);
  return *this;
}
 
// Function: acquire
template <typename I>
Task& Task::acquire(I first, I last) {
  if(!_node->_semaphores) {
    _node->_semaphores = std::make_unique<Node::Semaphores>();
  }
  _node->_semaphores->to_acquire.reserve(
    _node->_semaphores->to_acquire.size() + std::distance(first, last)
  );
  for(auto s = first; s != last; ++s){
    _node->_semaphores->to_acquire.push_back(&(*s));
  }
  return *this;
}
 
// Function: release
inline Task& Task::release(Semaphore& s) {
  if(!_node->_semaphores) {
    _node->_semaphores = std::make_unique<Node::Semaphores>();
  }
  _node->_semaphores->to_release.push_back(&s);
  return *this;
}
 
// Function: release
template <typename I>
Task& Task::release(I first, I last) {
  if(!_node->_semaphores) {
    _node->_semaphores = std::make_unique<Node::Semaphores>();
  }
  _node->_semaphores->to_release.reserve(
    _node->_semaphores->to_release.size() + std::distance(first, last)
  );
  for(auto s = first; s != last; ++s) {
    _node->_semaphores->to_release.push_back(&(*s));
  }
  return *this;
}
 
// Procedure: reset
inline void Task::reset() {
  _node = nullptr;
}
 
// Procedure: reset_work
inline void Task::reset_work() {
  _node->_handle.emplace<std::monostate>();
}
 
// Function: name
inline const std::string& Task::name() const {
  return _node->_name;
}
 
// Function: num_predecessors
inline size_t Task::num_predecessors() const {
  return _node->num_predecessors();
}
 
// Function: num_strong_dependencies
inline size_t Task::num_strong_dependencies() const {
  return _node->num_strong_dependencies();
}
 
// Function: num_weak_dependencies
inline size_t Task::num_weak_dependencies() const {
  return _node->num_weak_dependencies();
}
 
// Function: num_successors
inline size_t Task::num_successors() const {
  return _node->num_successors();
}
 
// Function: empty
inline bool Task::empty() const {
  return _node == nullptr;
}
 
// Function: has_work
inline bool Task::has_work() const {
  return _node ? _node->_handle.index() != 0 : false;
}
 
// Function: exception
inline std::exception_ptr Task::exception_ptr() const {
  return _node ? _node->_exception_ptr : nullptr;
}
 
// Function: has_exception
inline bool Task::has_exception_ptr() const {
  return _node ? (_node->_exception_ptr != nullptr) : false;
}
 
// Function: task_type
inline TaskType Task::type() const {
  switch(_node->_handle.index()) {
    case Node::PLACEHOLDER:           return TaskType::PLACEHOLDER;
    case Node::STATIC:                return TaskType::STATIC;
    case Node::RUNTIME:               return TaskType::RUNTIME;
    case Node::NONPREEMPTIVE_RUNTIME: return TaskType::RUNTIME;
    case Node::SUBFLOW:               return TaskType::SUBFLOW;
    case Node::CONDITION:             return TaskType::CONDITION;
    case Node::MULTI_CONDITION:       return TaskType::CONDITION;
    case Node::MODULE:                return TaskType::MODULE;
    case Node::ADOPTED_MODULE:        return TaskType::MODULE;
    case Node::ASYNC:                 return TaskType::ASYNC;
    case Node::DEPENDENT_ASYNC:       return TaskType::ASYNC;
    default:                          return TaskType::UNDEFINED;
  }
}
 
// Function: for_each_successor
template <typename V>
void Task::for_each_successor(V&& visitor) const {
  for(size_t i=0; i<_node->_num_successors; ++i) {
    visitor(Task(_node->_edges[i]));
  }
}
 
// Function: for_each_predecessor
template <typename V>
void Task::for_each_predecessor(V&& visitor) const {
  for(size_t i=_node->_num_successors; i<_node->_edges.size(); ++i) {
    visitor(Task(_node->_edges[i]));
  }
}
 
// Function: for_each_subflow_task
template <typename V>
void Task::for_each_subflow_task(V&& visitor) const {
  if(auto ptr = std::get_if<Node::Subflow>(&_node->_handle); ptr) {
    for(auto itr = ptr->subgraph.begin(); itr != ptr->subgraph.end(); ++itr) {
      visitor(Task(*itr));
    }
  }
}
 
// Function: hash_value
inline size_t Task::hash_value() const {
  return std::hash<Node*>{}(_node);
}
 
// Procedure: dump
inline void Task::dump(std::ostream& os) const {
  os << "task ";
  if(name().empty()) os << _node;
  else os << name();
  os << " [type=" << to_string(type()) << ']';
}
 
// Function: work
template <typename C>
Task& Task::work(C&& c) {
 
  if constexpr(is_static_task_v<C>) {
    _node->_handle.emplace<Node::Static>(std::forward<C>(c));
  }
  else if constexpr(is_runtime_task_v<C>) {
    _node->_handle.emplace<Node::Runtime>(std::forward<C>(c));
  }
  else if constexpr(is_subflow_task_v<C>) {
    _node->_handle.emplace<Node::Subflow>(std::forward<C>(c));
  }
  else if constexpr(is_condition_task_v<C>) {
    _node->_handle.emplace<Node::Condition>(std::forward<C>(c));
  }
  else if constexpr(is_multi_condition_task_v<C>) {
    _node->_handle.emplace<Node::MultiCondition>(std::forward<C>(c));
  }
  else {
    static_assert(dependent_false_v<C>, "invalid task callable");
  }
  return *this;
}
 
// Function: data
inline void* Task::data() const {
  return _node->_data;
}
 
// Function: data
inline Task& Task::data(void* data) {
  _node->_data = data;
  return *this;
}
 
// ----------------------------------------------------------------------------
// global ostream
// ----------------------------------------------------------------------------

inline std::ostream& operator << (std::ostream& os, const Task& task) {
  task.dump(os);
  return os;
}
 
// ----------------------------------------------------------------------------
// Task View
// ----------------------------------------------------------------------------

class TaskView {
 
  friend class Executor;
 
  public:

  const std::string& name() const;

  size_t num_successors() const;

  size_t num_predecessors() const;

  size_t num_strong_dependencies() const;

  size_t num_weak_dependencies() const;

  template <typename V>
  void for_each_successor(V&& visitor) const;

  template <typename V>
  void for_each_predecessor(V&& visitor) const;

  TaskType type() const;

  size_t hash_value() const;
 
  private:
 
  TaskView(const Node&);
  TaskView(const TaskView&) = default;
 
  const Node& _node;
};
 
// Constructor
inline TaskView::TaskView(const Node& node) : _node {node} {
}
 
// Function: name
inline const std::string& TaskView::name() const {
  return _node._name;
}
 
// Function: num_predecessors
inline size_t TaskView::num_predecessors() const {
  return _node.num_predecessors();
}
 
// Function: num_strong_dependencies
inline size_t TaskView::num_strong_dependencies() const {
  return _node.num_strong_dependencies();
}
 
// Function: num_weak_dependencies
inline size_t TaskView::num_weak_dependencies() const {
  return _node.num_weak_dependencies();
}
 
// Function: num_successors
inline size_t TaskView::num_successors() const {
  return _node.num_successors();
}
 
// Function: type
inline TaskType TaskView::type() const {
  switch(_node._handle.index()) {
    case Node::PLACEHOLDER:           return TaskType::PLACEHOLDER;
    case Node::STATIC:                return TaskType::STATIC;
    case Node::RUNTIME:               return TaskType::RUNTIME;
    case Node::NONPREEMPTIVE_RUNTIME: return TaskType::RUNTIME;
    case Node::SUBFLOW:               return TaskType::SUBFLOW;
    case Node::CONDITION:             return TaskType::CONDITION;
    case Node::MULTI_CONDITION:       return TaskType::CONDITION;
    case Node::MODULE:                return TaskType::MODULE;
    case Node::ADOPTED_MODULE:        return TaskType::MODULE;
    case Node::ASYNC:                 return TaskType::ASYNC;
    case Node::DEPENDENT_ASYNC:       return TaskType::ASYNC;
    default:                          return TaskType::UNDEFINED;
  }
}
 
// Function: hash_value
inline size_t TaskView::hash_value() const {
  return std::hash<const Node*>{}(&_node);
}
 
// Function: for_each_successor
template <typename V>
void TaskView::for_each_successor(V&& visitor) const {
  for(size_t i=0; i<_node._num_successors; ++i) {
    visitor(TaskView(*_node._edges[i]));
  }
  //for(size_t i=0; i<_node._successors.size(); ++i) {
  //  visitor(TaskView(*_node._successors[i]));
  //}
}
 
// Function: for_each_predecessor
template <typename V>
void TaskView::for_each_predecessor(V&& visitor) const {
  for(size_t i=_node._num_successors; i<_node._edges.size(); ++i) {
    visitor(TaskView(*_node._edges[i]));
  }
  //for(size_t i=0; i<_node._predecessors.size(); ++i) {
  //  visitor(TaskView(*_node._predecessors[i]));
  //}
}
 
}  // end of namespace tf. ----------------------------------------------------
 
namespace std {

template <>
struct hash<tf::Task> {
  auto operator() (const tf::Task& task) const noexcept {
    return task.hash_value();
  }
};

template <>
struct hash<tf::TaskView> {
  auto operator() (const tf::TaskView& task_view) const noexcept {
    return task_view.hash_value();
  }
};
 
}  // end of namespace std ----------------------------------------------------