core/taskflow.hpp

Go to the documentation of this file.

#pragma once
 
#include "flow_builder.hpp"
 
namespace tf {
 
// ----------------------------------------------------------------------------
 
class Taskflow : public FlowBuilder {
 
  friend class Topology;
  friend class Executor;
  friend class FlowBuilder;
 
  struct Dumper {
    size_t id;
    std::stack<std::pair<const Node*, const Graph*>> stack;
    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;
 
    Graph& graph();
 
  private:
 
    mutable std::mutex _mutex;
 
    std::string _name;
 
    Graph _graph;
 
    std::queue<std::shared_ptr<Topology>> _topologies;
 
    std::optional<std::list<Taskflow>::iterator> _satellite;
 
    void _dump(std::ostream&, const Graph*) const;
    void _dump(std::ostream&, const Node*, Dumper&) const;
    void _dump(std::ostream&, const Graph*, Dumper&) const;
};
 
// 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);
  _satellite = rhs._satellite;
 
  rhs._satellite.reset();
}
 
// 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);
    _satellite = rhs._satellite;
    rhs._satellite.reset();
  }
  return *this;
}
 
// Procedure:
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(size_t i=0; i<_graph._nodes.size(); ++i) {
    visitor(Task(_graph._nodes[i]));
  }
}
 
// Procedure: 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";
  _dump(os, &_graph);
  os << "}\n";
}
 
// Procedure: _dump
inline void Taskflow::_dump(std::ostream& os, const Graph* top) const {
 
  Dumper dumper;
 
  dumper.id = 0;
  dumper.stack.push({nullptr, top});
  dumper.visited[top] = dumper.id++;
 
  while(!dumper.stack.empty()) {
 
    auto [p, f] = dumper.stack.top();
    dumper.stack.pop();
 
    os << "subgraph cluster_p" << f << " {\nlabel=\"";
 
    // n-level module
    if(p) {
      os << 'm' << dumper.visited[f];
    }
    // top-level taskflow graph
    else {
      os << "Taskflow: ";
      if(_name.empty()) os << 'p' << this;
      else os << _name;
    }
 
    os << "\";\n";
 
    _dump(os, f, dumper);
    os << "}\n";
  }
}
 
// Procedure: _dump
inline void Taskflow::_dump(
  std::ostream& os, const Node* node, Dumper& dumper
) const {
 
  os << 'p' << node << "[label=\"";
  if(node->_name.empty()) os << 'p' << node;
  else os << node->_name;
  os << "\" ";
 
  // shape for node
  switch(node->_handle.index()) {
 
    case Node::CONDITION:
    case Node::MULTI_CONDITION:
      os << "shape=diamond color=black fillcolor=aquamarine style=filled";
    break;
 
    case Node::RUNTIME:
      os << "shape=component";
    break;
 
    case Node::CUDAFLOW:
      os << " style=\"filled\""
         << " color=\"black\" fillcolor=\"purple\""
         << " fontcolor=\"white\""
         << " shape=\"folder\"";
    break;
 
    case Node::SYCLFLOW:
      os << " style=\"filled\""
         << " color=\"black\" fillcolor=\"red\""
         << " fontcolor=\"white\""
         << " shape=\"folder\"";
    break;
 
    default:
    break;
  }
 
  os << "];\n";
 
  for(size_t s=0; s<node->_successors.size(); ++s) {
    if(node->_is_conditioner()) {
      // case edge is dashed
      os << 'p' << node << " -> p" << node->_successors[s]
         << " [style=dashed label=\"" << s << "\"];\n";
    } else {
      os << 'p' << node << " -> p" << node->_successors[s] << ";\n";
    }
  }
 
  // subflow join node
  if(node->_parent && node->_parent->_handle.index() == Node::DYNAMIC &&
     node->_successors.size() == 0
    ) {
    os << 'p' << node << " -> p" << node->_parent << ";\n";
  }
 
  // node info
  switch(node->_handle.index()) {
 
    case Node::DYNAMIC: {
      auto& sbg = std::get_if<Node::Dynamic>(&node->_handle)->subgraph;
      if(!sbg.empty()) {
        os << "subgraph cluster_p" << node << " {\nlabel=\"Subflow: ";
        if(node->_name.empty()) os << 'p' << node;
        else os << node->_name;
 
        os << "\";\n" << "color=blue\n";
        _dump(os, &sbg, dumper);
        os << "}\n";
      }
    }
    break;
 
    case Node::CUDAFLOW: {
      std::get_if<Node::cudaFlow>(&node->_handle)->graph->dump(
        os, node, node->_name
      );
    }
    break;
 
    case Node::SYCLFLOW: {
      std::get_if<Node::syclFlow>(&node->_handle)->graph->dump(
        os, node, node->_name
      );
    }
    break;
 
    default:
    break;
  }
}
 
// Procedure: _dump
inline void Taskflow::_dump(
  std::ostream& os, const Graph* graph, Dumper& dumper
) const {
 
  for(const auto& n : graph->_nodes) {
 
    // regular task
    if(n->_handle.index() != Node::MODULE) {
      _dump(os, n, dumper);
    }
    // module task
    else {
      //auto module = &(std::get_if<Node::Module>(&n->_handle)->module);
      auto module = &(std::get_if<Node::Module>(&n->_handle)->graph);
 
      os << 'p' << n << "[shape=box3d, color=blue, label=\"";
      if(n->_name.empty()) os << 'p' << n;
      else os << n->_name;
 
      if(dumper.visited.find(module) == dumper.visited.end()) {
        dumper.visited[module] = dumper.id++;
        dumper.stack.push({n, module});
      }
 
      os << " [m" << dumper.visited[module] << "]\"];\n";
 
      for(const auto s : n->_successors) {
        os << 'p' << n << "->" << 'p' << s << ";\n";
      }
    }
  }
}
 
// ----------------------------------------------------------------------------
// class definition: Future
// ----------------------------------------------------------------------------
 
template <typename T>
class Future : public std::future<T>  {
 
  friend class Executor;
  friend class Subflow;
 
  using handle_t = std::variant<
    std::monostate, std::weak_ptr<Topology>, std::weak_ptr<AsyncTopology>
  >;
 
  // variant index
  constexpr static auto ASYNC = get_index_v<std::weak_ptr<AsyncTopology>, handle_t>;
  constexpr static auto TASKFLOW = get_index_v<std::weak_ptr<Topology>, handle_t>;
 
  public:
 
    Future() = default;
 
    Future(const Future&) = delete;
 
    Future(Future&&) = default;
 
    Future& operator = (const Future&) = delete;
 
    Future& operator = (Future&&) = default;
 
    bool cancel();
 
  private:
 
    handle_t _handle;
 
    template <typename P>
    Future(std::future<T>&&, P&&);
};
 
template <typename T>
template <typename P>
Future<T>::Future(std::future<T>&& fu, P&& p) :
  std::future<T> {std::move(fu)},
  _handle        {std::forward<P>(p)} {
}
 
// Function: cancel
template <typename T>
bool Future<T>::cancel() {
  return std::visit([](auto&& arg){
    using P = std::decay_t<decltype(arg)>;
    if constexpr(std::is_same_v<P, std::monostate>) {
      return false;
    }
    else {
      auto ptr = arg.lock();
      if(ptr) {
        ptr->_is_cancelled.store(true, std::memory_order_relaxed);
        return true;
      }
      return false;
    }
  }, _handle);
}
 
 
}  // end of namespace tf. ---------------------------------------------------