flow_builder.hpp

Go to the documentation of this file.

#pragma once
 
#include "task.hpp"
 
namespace tf {
 
class FlowBuilder {
 
  friend class Executor;
 
  public:
 
    FlowBuilder(Graph& graph);
 
    template <typename C,
      std::enable_if_t<is_static_task_v<C>, void>* = nullptr
    >
    Task emplace(C&& callable);
 
    template <typename C,
      std::enable_if_t<is_dynamic_task_v<C>, void>* = nullptr
    >
    Task emplace(C&& callable);
 
    template <typename C,
      std::enable_if_t<is_condition_task_v<C>, void>* = nullptr
    >
    Task emplace(C&& callable);
 
    template <typename C,
      std::enable_if_t<is_multi_condition_task_v<C>, void>* = nullptr
    >
    Task emplace(C&& callable);
 
    template <typename... C, std::enable_if_t<(sizeof...(C)>1), void>* = nullptr>
    auto emplace(C&&... callables);
 
    void erase(Task task);
 
    template <typename T>
    Task composed_of(T& object);
 
    Task placeholder();
 
    template <typename C,
      std::enable_if_t<is_cudaflow_task_v<C>, void>* = nullptr
    >
    Task emplace(C&& callable);
 
    template <typename C, typename D,
      std::enable_if_t<is_cudaflow_task_v<C>, void>* = nullptr
    >
    Task emplace_on(C&& callable, D&& device);
 
    template <typename C, std::enable_if_t<is_syclflow_task_v<C>, void>* = nullptr>
    Task emplace(C&& callable);
 
    template <typename C, typename Q,
      std::enable_if_t<is_syclflow_task_v<C>, void>* = nullptr
    >
    Task emplace_on(C&& callable, Q&& queue);
 
    template <typename C,
      std::enable_if_t<is_runtime_task_v<C>, void>* = nullptr
    >
    Task emplace(C&& callable);
 
    void linearize(std::vector<Task>& tasks);
 
    void linearize(std::initializer_list<Task> tasks);
 
    // ------------------------------------------------------------------------
    // parallel iterations
    // ------------------------------------------------------------------------
 
    template <typename B, typename E, typename C>
    Task for_each(B first, E last, C callable);
 
    template <typename B, typename E, typename S, typename C>
    Task for_each_index(B first, E last, S step, C callable);
 
    // ------------------------------------------------------------------------
    // transform
    // ------------------------------------------------------------------------
 
    template <typename B, typename E, typename O, typename C>
    Task transform(B first1, E last1, O d_first, C c);
 
    template <typename B1, typename E1, typename B2, typename O, typename C>
    Task transform(B1 first1, E1 last1, B2 first2, O d_first, C c);
 
    // ------------------------------------------------------------------------
    // reduction
    // ------------------------------------------------------------------------
 
    template <typename B, typename E, typename T, typename O>
    Task reduce(B first, E last, T& init, O bop);
 
    // ------------------------------------------------------------------------
    // transfrom and reduction
    // ------------------------------------------------------------------------
 
    template <typename B, typename E, typename T, typename BOP, typename UOP>
    Task transform_reduce(B first, E last, T& init, BOP bop, UOP uop);
 
    // ------------------------------------------------------------------------
    // sort
    // ------------------------------------------------------------------------
 
    template <typename B, typename E, typename C>
    Task sort(B first, E last, C cmp);
 
    template <typename B, typename E>
    Task sort(B first, E last);
 
  protected:
 
    Graph& _graph;
 
  private:
 
    template <typename L>
    void _linearize(L&);
};
 
// Constructor
inline FlowBuilder::FlowBuilder(Graph& graph) :
  _graph {graph} {
}
 
// Function: emplace
template <typename C, std::enable_if_t<is_static_task_v<C>, void>*>
Task FlowBuilder::emplace(C&& c) {
  return Task(_graph._emplace_back(
    std::in_place_type_t<Node::Static>{}, std::forward<C>(c)
  ));
}
 
// Function: emplace
template <typename C, std::enable_if_t<is_dynamic_task_v<C>, void>*>
Task FlowBuilder::emplace(C&& c) {
  return Task(_graph._emplace_back(
    std::in_place_type_t<Node::Dynamic>{}, std::forward<C>(c)
  ));
}
 
// Function: emplace
template <typename C, std::enable_if_t<is_condition_task_v<C>, void>*>
Task FlowBuilder::emplace(C&& c) {
  return Task(_graph._emplace_back(
    std::in_place_type_t<Node::Condition>{}, std::forward<C>(c)
  ));
}
 
// Function: emplace
template <typename C, std::enable_if_t<is_multi_condition_task_v<C>, void>*>
Task FlowBuilder::emplace(C&& c) {
  return Task(_graph._emplace_back(
    std::in_place_type_t<Node::MultiCondition>{}, std::forward<C>(c)
  ));
}
 
// Function: emplace
template <typename C, std::enable_if_t<is_runtime_task_v<C>, void>*>
Task FlowBuilder::emplace(C&& c) {
  return Task(_graph._emplace_back(
    std::in_place_type_t<Node::Runtime>{}, std::forward<C>(c)
  ));
}
 
// Function: emplace
template <typename... C, std::enable_if_t<(sizeof...(C)>1), void>*>
auto FlowBuilder::emplace(C&&... cs) {
  return std::make_tuple(emplace(std::forward<C>(cs))...);
}
 
// Function: erase
inline void FlowBuilder::erase(Task task) {
 
  if (!task._node) {
    return;
  }
 
  task.for_each_dependent([&] (Task dependent) {
    auto& S = dependent._node->_successors;
    if(auto I = std::find(S.begin(), S.end(), task._node); I != S.end()) {
      S.erase(I);
    }
  });
 
  task.for_each_successor([&] (Task dependent) {
    auto& D = dependent._node->_dependents;
    if(auto I = std::find(D.begin(), D.end(), task._node); I != D.end()) {
      D.erase(I);
    }
  });
 
  _graph._erase(task._node);
}
 
// Function: composed_of
template <typename T>
Task FlowBuilder::composed_of(T& object) {
  auto node = _graph._emplace_back(
    std::in_place_type_t<Node::Module>{}, object
  );
  return Task(node);
}
 
// Function: placeholder
inline Task FlowBuilder::placeholder() {
  auto node = _graph._emplace_back();
  return Task(node);
}
 
// Procedure: _linearize
template <typename L>
void FlowBuilder::_linearize(L& keys) {
 
  auto itr = keys.begin();
  auto end = keys.end();
 
  if(itr == end) {
    return;
  }
 
  auto nxt = itr;
 
  for(++nxt; nxt != end; ++nxt, ++itr) {
    itr->_node->_precede(nxt->_node);
  }
}
 
// Procedure: linearize
inline void FlowBuilder::linearize(std::vector<Task>& keys) {
  _linearize(keys);
}
 
// Procedure: linearize
inline void FlowBuilder::linearize(std::initializer_list<Task> keys) {
  _linearize(keys);
}
 
// ----------------------------------------------------------------------------
 
class Subflow : public FlowBuilder {
 
  friend class Executor;
  friend class FlowBuilder;
  friend class Runtime;
 
  public:
 
    void join();
 
    void detach();
 
    void reset();
 
    bool joinable() const noexcept;
 
    template <typename F, typename... ArgsT>
    auto async(F&& f, ArgsT&&... args);
 
    template <typename F, typename... ArgsT>
    auto named_async(const std::string& name, F&& f, ArgsT&&... args);
 
    template <typename F, typename... ArgsT>
    void silent_async(F&& f, ArgsT&&... args);
 
    template <typename F, typename... ArgsT>
    void named_silent_async(const std::string& name, F&& f, ArgsT&&... args);
 
    inline Executor& executor();
 
  private:
 
    Executor& _executor;
    Worker& _worker;
    Node* _parent;
    bool _joinable {true};
 
    Subflow(Executor&, Worker&, Node*, Graph&);
 
    template <typename F, typename... ArgsT>
    auto _named_async(Worker& w, const std::string& name, F&& f, ArgsT&&... args);
 
    template <typename F, typename... ArgsT>
    void _named_silent_async(Worker& w, const std::string& name, F&& f, ArgsT&&... args);
};
 
// Constructor
inline Subflow::Subflow(
  Executor& executor, Worker& worker, Node* parent, Graph& graph
) :
  FlowBuilder {graph},
  _executor   {executor},
  _worker     {worker},
  _parent     {parent} {
  // assert(_parent != nullptr);
}
 
// Function: joined
inline bool Subflow::joinable() const noexcept {
  return _joinable;
}
 
// Function: executor
inline Executor& Subflow::executor() {
  return _executor;
}
 
// Procedure: reset
inline void Subflow::reset() {
  _graph._clear();
  _joinable = true;
}
 
}  // end of namespace tf. ---------------------------------------------------