taskflow.hpp

#pragma once
 
#include "flow_builder.hpp"

 
namespace tf {
 
// ----------------------------------------------------------------------------

class Taskflow : public FlowBuilder {
 
  friend class Topology;
  friend class Executor;
  friend class FlowBuilder;
  friend class Subflow;
 
  struct Dumper {
    size_t id;
    std::stack<std::tuple<const Node*, const Graph*, size_t>> stack;  // added depth
    std::unordered_map<const Graph*, size_t> visited;
  };
 
  public:

  Taskflow(const std::string& name);

  Taskflow();

  Taskflow(Taskflow&& rhs);

  Taskflow& operator = (Taskflow&& rhs);

  ~Taskflow() = default;

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

  std::string dump() const;

  size_t num_tasks() const;

  bool empty() const;

  void name(const std::string&);

  const std::string& name() const;

  void clear();

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

  void remove_dependency(Task from, Task to);

  Graph& graph();
 
  private:
 
  mutable std::mutex _mutex;
 
  std::string _name;
 
  Graph _graph;
 
  std::queue<std::shared_ptr<Topology>> _topologies;
 
  void _dump(std::ostream&, const Graph*) const;
  void _dump(std::ostream&, const Node*, Dumper&, size_t level) const;
  void _dump(std::ostream&, const Graph*, Dumper&, size_t level) const;
 
  size_t _fetch_enqueue(std::shared_ptr<Topology>);
};
 
// Constructor
inline Taskflow::Taskflow(const std::string& name) :
  FlowBuilder {_graph},
  _name       {name} {
}
 
// Constructor
inline Taskflow::Taskflow() : FlowBuilder{_graph} {
}
 
// Move constructor
inline Taskflow::Taskflow(Taskflow&& rhs) : FlowBuilder{_graph} {
  std::scoped_lock<std::mutex> lock(rhs._mutex);
  _name = std::move(rhs._name);
  _graph = std::move(rhs._graph);
  _topologies = std::move(rhs._topologies);
}
 
// Move assignment
inline Taskflow& Taskflow::operator = (Taskflow&& rhs) {
  if(this != &rhs) {
    std::scoped_lock<std::mutex, std::mutex> lock(_mutex, rhs._mutex);
    _name = std::move(rhs._name);
    _graph = std::move(rhs._graph);
    _topologies = std::move(rhs._topologies);
  }
  return *this;
}
 
// Function:
inline void Taskflow::clear() {
  _graph.clear();
}
 
// Function: num_tasks
inline size_t Taskflow::num_tasks() const {
  return _graph.size();
}
 
// Function: empty
inline bool Taskflow::empty() const {
  return _graph.empty();
}
 
// Function: name
inline void Taskflow::name(const std::string &name) {
  _name = name;
}
 
// Function: name
inline const std::string& Taskflow::name() const {
  return _name;
}
 
// Function: graph
inline Graph& Taskflow::graph() {
  return _graph;
}
 
// Function: for_each_task
template <typename V>
void Taskflow::for_each_task(V&& visitor) const {
  for(auto itr = _graph.begin(); itr != _graph.end(); ++itr) {
    visitor(Task(*itr));
  }
}
 
// Function: remove_dependency
inline void Taskflow::remove_dependency(Task from, Task to) {
  // remove "to" from the succcessor list of "from"
  from._node->_remove_successors(to._node);
 
  // remove "from" from the predecessor list of "to"
  to._node->_remove_predecessors(from._node);
}
 
// Function: _fetch_enqueue
inline size_t Taskflow::_fetch_enqueue(std::shared_ptr<Topology> tpg) {
  std::lock_guard<std::mutex> lock(_mutex);
  auto pre_size = _topologies.size();
  _topologies.emplace(std::move(tpg));
  return pre_size;
}
 
// Function: dump
inline std::string Taskflow::dump() const {
  std::ostringstream oss;
  dump(oss);
  return oss.str();
}
 
// Function: dump
inline void Taskflow::dump(std::ostream& os) const {
  os << "digraph Taskflow {\n";
  os << "  compound=true;\n";  // required for lhead/ltail cluster edges
  _dump(os, &_graph);
  os << "}\n";
}
 
// Function: _dump (top-level — iterates module stack)
inline void Taskflow::_dump(std::ostream& os, const Graph* top) const {
 
  Dumper dumper;
 
  dumper.id = 0;
  dumper.stack.push({nullptr, top, 1});
  dumper.visited[top] = dumper.id++;
 
  while(!dumper.stack.empty()) {
 
    auto [p, f, depth] = dumper.stack.top();
    dumper.stack.pop();
 
    std::string ind(depth * 2, ' ');
    std::string ind2((depth + 1) * 2, ' ');
 
    os << ind << "subgraph cluster_p" << f << " {\n";
    os << ind2 << "label=\"";
 
    // n-level module
    if(p) {
      if (p->_name.empty()) os << 'm' << dumper.visited[f];
      else os << p->name();
    }
    // top-level taskflow graph
    else {
      os << "Taskflow: ";
      if(_name.empty()) os << 'p' << this;
      else os << _name;
    }
 
    os << "\";\n";
 
    _dump(os, f, dumper, depth + 1);
    os << ind << "}\n";
  }
}
 
// Function: _dump (single node)
inline void Taskflow::_dump(
  std::ostream& os, const Node* node, Dumper& dumper, size_t level
) const {
 
  std::string ind(level * 2, ' ');
 
  // label of the node
  os << ind << 'p' << node << "[label=\"";
  if(node->_name.empty()) os << 'p' << node;
  else os << node->_name;
  os << "\" ";
 
  // shape of the node
  switch(node->_handle.index()) {
 
    case Node::CONDITION:
    case Node::MULTI_CONDITION:
      os << "shape=diamond color=black fillcolor=aquamarine style=filled";
    break;
 
    default:
    break;
  }
 
  os << "];\n";
 
  for(size_t s=0; s<node->_num_successors; ++s) {
    if(node->_is_conditioner()) {
      os << ind << 'p' << node << " -> p" << node->_edges[s]
         << " [style=dashed label=\"" << s << "\"];\n";
    } else {
      os << ind << 'p' << node << " -> p" << node->_edges[s] << ";\n";
    }
  }
 
  // node info
  switch(auto hid = node->_handle.index(); hid) {
 
    case Node::SUBFLOW:
    case Node::ADOPTED_MODULE: {
 
      auto& g = (hid == Node::SUBFLOW) ? 
                std::get_if<Node::Subflow>(&node->_handle)->subgraph :
                std::get_if<Node::AdoptedModule>(&node->_handle)->graph;
 
      if(!g.empty()) {
        std::string ind2((level + 1) * 2, ' ');
 
        os << ind  << "subgraph cluster_p" << node << " {\n";
        os << ind2 << ((hid == Node::SUBFLOW) ? "label=\"Subflow: " 
                                              : "label=\"AdoptedModule: ");
        if(node->_name.empty()) os << 'p' << node;
        else os << node->_name;
        os << "\";\n";
        os << ind2 << "color=blue;\n";
 
        _dump(os, &g, dumper, level + 1);
        os << ind << "}\n";
 
        // Single cluster-level join edge: subflow cluster → parent node.
        // ltail clips the arrow tail at the cluster boundary.
        auto first = *g.begin();
        os << ind << 'p' << first << " -> p" << node
           << " [ltail=cluster_p" << node
           << " style=dashed color=blue];\n";
      }
    }
    break;
    
    default:
    break;
  }
}
 
// Function: _dump (graph — iterates nodes)
inline void Taskflow::_dump(
  std::ostream& os, const Graph* graph, Dumper& dumper, size_t level
) const {
 
  std::string ind(level * 2, ' ');
 
  for(auto itr = graph->begin(); itr != graph->end(); ++itr) {
 
    Node* n = *itr;
 
    // regular task
    if(n->_handle.index() != Node::MODULE) {
      _dump(os, n, dumper, level);
    }
    // module task
    else {
      auto mgraph = &(std::get_if<Node::Module>(&n->_handle)->graph);
 
      os << ind << 'p' << n << "[shape=box3d, color=blue, label=\"";
      if(n->_name.empty()) os << 'p' << n;
      else os << n->_name;
 
      if(dumper.visited.find(mgraph) == dumper.visited.end()) {
        dumper.visited[mgraph] = dumper.id++;
        dumper.stack.push({n, mgraph, level});
      }
 
      if(n->_name.empty()) os << " [m" << dumper.visited[mgraph] << "]";
      os << "\"];\n";
 
      for(size_t i=0; i<n->_num_successors; ++i) {
        os << ind << 'p' << n << " -> " << 'p' << n->_edges[i] << ";\n";
      }
    }
  }
}
 
// ----------------------------------------------------------------------------
// class definition: Future
// ----------------------------------------------------------------------------

template <typename T>
class Future : public std::future<T>  {
 
  friend class Executor;
  friend class Subflow;
  friend class Runtime;
 
  public:

  Future() = default;

  Future(const Future&) = delete;

  Future(Future&&) = default;

  Future(std::future<T>&&);

  Future& operator = (const Future&) = delete;

  Future& operator = (Future&&) = default;

  bool cancel();
 
  private:
 
  std::weak_ptr<Topology> _topology;
 
  Future(std::future<T>&&, std::weak_ptr<Topology>);
};
 
template <typename T>
Future<T>::Future(std::future<T>&& f, std::weak_ptr<Topology> p) :
  std::future<T> {std::move(f)},
  _topology      {std::move(p)} {
}
 
template <typename T>
Future<T>::Future(std::future<T>&& f) : std::future<T> {std::move(f)} {
}
 
// Function: cancel
template <typename T>
bool Future<T>::cancel() {
  if(auto ptr = _topology.lock(); ptr) {
    ptr->_estate.fetch_or(ESTATE::CANCELLED, std::memory_order_relaxed);
    return true;
  }
  return false;
}
 
 
}  // end of namespace tf. ---------------------------------------------------