flow_builder.hpp

#pragma once
 
#include "task.hpp"
#include "../algorithm/partitioner.hpp"

 
namespace tf {

class FlowBuilder {
 
  friend class Executor;
 
  public:

  FlowBuilder(Graph& graph);

  template <StaticTaskLike C>
  Task emplace(C&& callable);
  
  template <RuntimeTaskLike C>
  Task emplace(C&& callable);

  template <SubflowTaskLike C>
  Task emplace(C&& callable);

  template <ConditionTaskLike C>
  Task emplace(C&& callable);

  template <MultiConditionTaskLike C>
  Task emplace(C&& callable);

  template <typename... C> requires (sizeof...(C) > 1)
  auto emplace(C&&... callables);

  void erase(Task task);

  template <GraphLike T>
  Task composed_of(T& object);
  
  Task adopt(Graph&& graph);

  Task placeholder();

  void linearize(std::vector<Task>& tasks);

  void linearize(std::initializer_list<Task> tasks);
 
 
  // ------------------------------------------------------------------------
  // parallel iterations
  // ------------------------------------------------------------------------

  template <typename B, typename E, typename C, PartitionerLike P = DefaultPartitioner>
  Task for_each(B first, E last, C callable, P part = P());
  
  template <typename B, typename E, typename S, typename C, PartitionerLike P = DefaultPartitioner>
  Task for_each_index(B first, E last, S step, C callable, P part = P());

  template <IndexRange1DLike R, typename C, PartitionerLike P = DefaultPartitioner>
  Task for_each_by_index(R range, C callable, P part = P());
  
  template <IndexRangeMDLike R, typename C, PartitionerLike P = DefaultPartitioner>
  Task for_each_by_index(R range, C callable, P part = P());
 
  // ------------------------------------------------------------------------
  // transform
  // ------------------------------------------------------------------------

  template <typename B, typename E, typename O, typename C,
            PartitionerLike P = DefaultPartitioner>
  Task transform(B first1, E last1, O d_first, C c, P part = P());
  
  template <typename B1, typename E1, typename B2, typename O, typename C,
            PartitionerLike P = DefaultPartitioner>
  requires (!PartitionerLike<std::decay_t<C>>)
  Task transform(B1 first1, E1 last1, B2 first2, O d_first, C c, P part = P());
  
  // ------------------------------------------------------------------------
  // reduction
  // ------------------------------------------------------------------------

  template <typename B, typename E, typename T, typename O, PartitionerLike P = DefaultPartitioner>
  Task reduce(B first, E last, T& init, O bop, P part = P());

  template <IndexRangeLike R, typename T, typename L, typename G, PartitionerLike P = DefaultPartitioner>
  Task reduce_by_index(R range, T& init, L lop, G gop, P part = P());
  
  // ------------------------------------------------------------------------
  // transform and reduction
  // ------------------------------------------------------------------------

  template <typename B, typename E, typename T, typename BOP, typename UOP,
            PartitionerLike P = DefaultPartitioner>
  Task transform_reduce(B first, E last, T& init, BOP bop, UOP uop, P part = P());

  
  template <typename B1, typename E1, typename B2, typename T,
            typename BOP_R, typename BOP_T, PartitionerLike P = DefaultPartitioner>
  requires (!PartitionerLike<std::decay_t<BOP_T>>)
  Task transform_reduce(
    B1 first1, E1 last1, B2 first2, T& init, BOP_R bop_r, BOP_T bop_t, P part = P()
  );
 
  // ------------------------------------------------------------------------
  // scan
  // ------------------------------------------------------------------------

  template <typename B, typename E, typename D, typename BOP>
  Task inclusive_scan(B first, E last, D d_first, BOP bop);
  
  template <typename B, typename E, typename D, typename BOP, typename T>
  Task inclusive_scan(B first, E last, D d_first, BOP bop, T init);
  
  template <typename B, typename E, typename D, typename T, typename BOP>
  Task exclusive_scan(B first, E last, D d_first, T init, BOP bop);
  
  // ------------------------------------------------------------------------
  // transform scan
  // ------------------------------------------------------------------------
  
  template <typename B, typename E, typename D, typename BOP, typename UOP>
  Task transform_inclusive_scan(B first, E last, D d_first, BOP bop, UOP uop);
  
  template <typename B, typename E, typename D, typename BOP, typename UOP, typename T>
  Task transform_inclusive_scan(B first, E last, D d_first, BOP bop, UOP uop, T init);
  
  template <typename B, typename E, typename D, typename T, typename BOP, typename UOP>
  Task transform_exclusive_scan(B first, E last, D d_first, T init, BOP bop, UOP uop);
 
  // ------------------------------------------------------------------------
  // find
  // ------------------------------------------------------------------------
 
  template <typename B, typename E, typename T, typename UOP, PartitionerLike P = DefaultPartitioner>
  Task find_if(B first, E last, T &result, UOP predicate, P part = P());

  template <typename B, typename E, typename T, typename UOP, PartitionerLike P = DefaultPartitioner>
  Task find_if_not(B first, E last, T &result, UOP predicate, P part = P());

  template <typename B, typename E, typename T, typename C, PartitionerLike P>
  Task min_element(B first, E last, T& result, C comp, P part);
  
  template <typename B, typename E, typename T, typename C, PartitionerLike P>
  Task max_element(B first, E last, T& result, C comp, P part);
 
  // ------------------------------------------------------------------------
  // 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);

 
  template <typename B1, typename E1, typename B2, typename E2,
            typename O, PartitionerLike P = DefaultPartitioner>
  Task merge(B1 first1, E1 last1, B2 first2, E2 last2, O d_first, P part = P());

 
  template <typename B1, typename E1, typename B2, typename E2,
            typename O, typename C, PartitionerLike P = DefaultPartitioner>
  requires (!PartitionerLike<std::decay_t<C>>)
  Task merge(B1 first1, E1 last1, B2 first2, E2 last2, O d_first, C cmp, P part = P());

  template<typename B, typename E, typename V, PartitionerLike P = DefaultPartitioner>
  Task fill(B first, E last, V value, P part = P());

  template<typename B, std::integral C, typename V, PartitionerLike P = DefaultPartitioner>
  Task fill_n(B first, C count, V value, P part = P());
  
  template <typename B, typename E, typename G, PartitionerLike P= DefaultPartitioner>
  Task generate(B first, E last, G gen, P part = P());

  template <typename B, std::integral C, typename G, PartitionerLike P = DefaultPartitioner>
  Task generate_n(B first, C count, G gen, P part = P());
 
  protected:
  
  Graph& _graph;
 
  private:
 
  template <typename L>
  void _linearize(L&);
};
 
// Constructor
inline FlowBuilder::FlowBuilder(Graph& graph) :
  _graph {graph} {
}
 
// Function: emplace
template <StaticTaskLike C>
Task FlowBuilder::emplace(C&& c) {
  return Task(_graph._emplace_back(NSTATE::NONE, ESTATE::NONE, DefaultTaskParams{}, nullptr, nullptr, 0,
    std::in_place_type_t<Node::Static>{}, std::forward<C>(c)
  ));
}
 
// Function: emplace
template <RuntimeTaskLike C>
Task FlowBuilder::emplace(C&& c) {
  if constexpr (std::is_invocable_v<C, tf::Runtime&>) {
    return Task(_graph._emplace_back(NSTATE::NONE, ESTATE::NONE, DefaultTaskParams{}, nullptr, nullptr, 0,
      std::in_place_type_t<Node::Runtime>{}, std::forward<C>(c)
    ));
  }
  else if constexpr (std::is_invocable_v<C, tf::NonpreemptiveRuntime&>) {
    return Task(_graph._emplace_back(NSTATE::NONE, ESTATE::NONE, DefaultTaskParams{}, nullptr, nullptr, 0,
      std::in_place_type_t<Node::NonpreemptiveRuntime>{}, std::forward<C>(c)
    ));
  }
  else {
    static_assert(dependent_false_v<C>, "invalid runtime task callable");
  }
}
 
// Function: emplace
template <SubflowTaskLike C>
Task FlowBuilder::emplace(C&& c) {
  return Task(_graph._emplace_back(NSTATE::NONE, ESTATE::NONE, DefaultTaskParams{}, nullptr, nullptr, 0,
    std::in_place_type_t<Node::Subflow>{}, std::forward<C>(c)
  ));
}
 
// Function: emplace
template <ConditionTaskLike C>
Task FlowBuilder::emplace(C&& c) {
  return Task(_graph._emplace_back(NSTATE::NONE, ESTATE::NONE, DefaultTaskParams{}, nullptr, nullptr, 0,
    std::in_place_type_t<Node::Condition>{}, std::forward<C>(c)
  ));
}
 
// Function: emplace
template <MultiConditionTaskLike C>
Task FlowBuilder::emplace(C&& c) {
  return Task(_graph._emplace_back(NSTATE::NONE, ESTATE::NONE, DefaultTaskParams{}, nullptr, nullptr, 0,
    std::in_place_type_t<Node::MultiCondition>{}, std::forward<C>(c)
  ));
}
 
// Function: composed_of
template <GraphLike T>
Task FlowBuilder::composed_of(T& target) {
  return Task(_graph._emplace_back(NSTATE::NONE, ESTATE::NONE, DefaultTaskParams{}, nullptr, nullptr, 0,
    std::in_place_type_t<Node::Module>{}, retrieve_graph(target)
  ));
}
 
// Function: adopt
inline Task FlowBuilder::adopt(Graph&& graph) {
  return Task(_graph._emplace_back(NSTATE::NONE, ESTATE::NONE, DefaultTaskParams{}, nullptr, nullptr, 0,
    std::in_place_type_t<Node::AdoptedModule>{}, std::move(graph)
  ));
}
 
// Function: placeholder
inline Task FlowBuilder::placeholder() {
  auto node = _graph._emplace_back(NSTATE::NONE, ESTATE::NONE, DefaultTaskParams{}, nullptr, nullptr, 0,
    std::in_place_type_t<Node::Placeholder>{}
  );
  return Task(node);
}
 
// Function: emplace
template <typename... C> requires (sizeof...(C) > 1)
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;
  }
 
  // remove task from its successors' predecessor list
  for(size_t i=0; i<task._node->_num_successors; ++i) {
    task._node->_edges[i]->_remove_predecessors(task._node);
  }
 
  // remove task from its precedessors' successor list
  for(size_t i=task._node->_num_successors; i<task._node->_edges.size(); ++i) {
    task._node->_edges[i]->_remove_successors(task._node);
  }
 
  _graph._erase(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;
 
  public:
    
    void join();

    bool joinable() const noexcept;

    Executor& executor() noexcept;
    
    Graph& graph() { return _graph; }
    
    void retain(bool flag) noexcept;

    bool retain() const;
 
  private:
    
    Subflow(Executor&, Worker&, Node*, Graph&);
    
    Subflow() = delete;
    Subflow(const Subflow&) = delete;
    Subflow(Subflow&&) = delete;
 
    Executor& _executor;
    Worker& _worker;
    Node* _node;
};
 
// Constructor
inline Subflow::Subflow(Executor& executor, Worker& worker, Node* node, Graph& graph) :
  FlowBuilder {graph}, 
  _executor   {executor}, 
  _worker     {worker}, 
  _node       {node} {
  
  // need to reset since there could have iterative control flow
  _node->_nstate &= ~(NSTATE::JOINED_SUBFLOW | NSTATE::RETAIN_SUBFLOW);
 
  // clear the graph
  graph.clear();
}
 
// Function: joinable
inline bool Subflow::joinable() const noexcept {
  return !(_node->_nstate & NSTATE::JOINED_SUBFLOW);
}
 
// Function: executor
inline Executor& Subflow::executor() noexcept {
  return _executor;
}
 
// Function: retain
inline void Subflow::retain(bool flag) noexcept {
  // default value is not to retain 
  if(flag == true) {
    _node->_nstate |= NSTATE::RETAIN_SUBFLOW;
  }
  else {
    _node->_nstate &= ~NSTATE::RETAIN_SUBFLOW;
  }
 
  //_node->_nstate = (_node->_nstate & ~NSTATE::RETAIN_SUBFLOW) | 
  //                 (-static_cast<int>(flag) & NSTATE::RETAIN_SUBFLOW);
}
 
// Function: retain
inline bool Subflow::retain() const {
  return _node->_nstate & NSTATE::RETAIN_SUBFLOW;
}
 
}  // end of namespace tf. ---------------------------------------------------