pipeline.hpp

#pragma once
 
#include "../taskflow.hpp"

 
namespace tf {
 
 
// ----------------------------------------------------------------------------
// Class Definition: Pipeflow
// ----------------------------------------------------------------------------

class Pipeflow {
 
  template <typename... Ps>
  friend class Pipeline;
 
  template <typename P>
  friend class ScalablePipeline;
 
  template <typename... Ps>
  friend class DataPipeline;
 
  public:

  Pipeflow() = default;

  size_t line() const {
    return _line;
  }

  size_t pipe() const {
    return _pipe;
  }

  size_t token() const {
    return _token;
  }

  void stop() {
    if(_pipe != 0) {
      TF_THROW("only the first pipe can stop the token");
    }
    _stop = true;
  }
 
  private:
 
  // Regular data
  size_t _line;
  size_t _pipe;
  size_t _token;
  bool   _stop;
};
 
// ----------------------------------------------------------------------------
// Class Definition: PipeType
// ----------------------------------------------------------------------------

enum class PipeType : int {
  PARALLEL = 1,
  SERIAL   = 2
};
 
// ----------------------------------------------------------------------------
// Class Definition: Pipe
// ----------------------------------------------------------------------------

template <typename C = std::function<void(tf::Pipeflow&)>>
class Pipe {
 
  template <typename... Ps>
  friend class Pipeline;
 
  template <typename P>
  friend class ScalablePipeline;
 
  public:

  using callable_t = C;

  Pipe() = default;

  Pipe(PipeType d, C&& callable) :
    _type{d}, _callable{std::forward<C>(callable)} {
  }

  PipeType type() const {
    return _type;
  }

  void type(PipeType type) {
    _type = type;
  }

  template <typename U>
  void callable(U&& callable) {
    _callable = std::forward<U>(callable);
  }
 
  private:
 
  PipeType _type;
 
  C _callable;
};
 
// ----------------------------------------------------------------------------
// Class Definition: Pipeline
// ----------------------------------------------------------------------------

template <typename... Ps>
class Pipeline {
 
  static_assert(sizeof...(Ps)>0, "must have at least one pipe");

  struct Line {
    std::atomic<size_t> join_counter;
  };

  struct PipeMeta {
    PipeType type;
  };
 
  public:

  Pipeline(size_t num_lines, Ps&&... ps);

  Pipeline(size_t num_lines, std::tuple<Ps...>&& ps);

  size_t num_lines() const noexcept;

  constexpr size_t num_pipes() const;

  void reset();

  size_t num_tokens() const noexcept;

  Graph& graph();
 
 
  private:
 
  Graph _graph;
 
  size_t _num_tokens;
 
  std::tuple<Ps...> _pipes;
  std::array<PipeMeta, sizeof...(Ps)> _meta;
  std::vector<std::array<Line, sizeof...(Ps)>> _lines;
  std::vector<Task> _tasks;
  std::vector<Pipeflow> _pipeflows;
  
  template <size_t... I>
  auto _gen_meta(std::tuple<Ps...>&&, std::index_sequence<I...>);
 
  void _on_pipe(Pipeflow&, NonpreemptiveRuntime&);
  void _build();
};
 
// constructor
template <typename... Ps>
Pipeline<Ps...>::Pipeline(size_t num_lines, Ps&&... ps) :
  _pipes     {std::make_tuple(std::forward<Ps>(ps)...)},
  _meta      {PipeMeta{ps.type()}...},
  _lines     (num_lines),
  _tasks     (num_lines + 1),
  _pipeflows (num_lines) {
 
  if(num_lines == 0) {
    TF_THROW("must have at least one line");
  }
 
  if(std::get<0>(_pipes).type() != PipeType::SERIAL) {
    TF_THROW("first pipe must be serial");
  }
 
  reset();
  _build();
}
 
// constructor
template <typename... Ps>
Pipeline<Ps...>::Pipeline(size_t num_lines, std::tuple<Ps...>&& ps) :
  _pipes     {std::forward<std::tuple<Ps...>>(ps)},
  _meta      {_gen_meta(
    std::forward<std::tuple<Ps...>>(ps), std::make_index_sequence<sizeof...(Ps)>{}
  )},
  _lines     (num_lines),
  _tasks     (num_lines + 1),
  _pipeflows (num_lines) {
 
  if(num_lines == 0) {
    TF_THROW("must have at least one line");
  }
 
  if(std::get<0>(_pipes).type() != PipeType::SERIAL) {
    TF_THROW("first pipe must be serial");
  }
 
  reset();
  _build();
}
 
// Function: _get_meta
template <typename... Ps>
template <size_t... I>
auto Pipeline<Ps...>::_gen_meta(std::tuple<Ps...>&& ps, std::index_sequence<I...>) {
  return std::array{PipeMeta{std::get<I>(ps).type()}...};
}
 
// Function: num_lines
template <typename... Ps>
size_t Pipeline<Ps...>::num_lines() const noexcept {
  return _pipeflows.size();
}
 
// Function: num_pipes
template <typename... Ps>
constexpr size_t Pipeline<Ps...>::num_pipes() const {
  return sizeof...(Ps);
}
 
// Function: num_tokens
template <typename... Ps>
size_t Pipeline<Ps...>::num_tokens() const noexcept {
  return _num_tokens;
}
 
// Function: graph
template <typename... Ps>
Graph& Pipeline<Ps...>::graph() {
  return _graph;
}
 
// Function: reset
template <typename... Ps>
void Pipeline<Ps...>::reset() {
 
  _num_tokens = 0;
 
  for(size_t l = 0; l<num_lines(); l++) {
    _pipeflows[l]._pipe = 0;
    _pipeflows[l]._line = l;
  }
  
  _lines[0][0].join_counter.store(0, std::memory_order_relaxed);
 
  for(size_t l=1; l<num_lines(); l++) {
    for(size_t f=1; f<num_pipes(); f++) {
      _lines[l][f].join_counter.store(
        static_cast<size_t>(_meta[f].type), std::memory_order_relaxed
      );
    }
  }
 
  for(size_t f=1; f<num_pipes(); f++) {
    _lines[0][f].join_counter.store(1, std::memory_order_relaxed);
  }
 
  for(size_t l=1; l<num_lines(); l++) {
    _lines[l][0].join_counter.store(
      static_cast<size_t>(_meta[0].type) - 1, std::memory_order_relaxed
    );
  }
}
 
// Procedure: _on_pipe
template <typename... Ps>
void Pipeline<Ps...>::_on_pipe(Pipeflow& pf, NonpreemptiveRuntime& rt) {
  visit_tuple([&](auto&& pipe){
    using callable_t = typename std::decay_t<decltype(pipe)>::callable_t;
    if constexpr (std::is_invocable_v<callable_t, Pipeflow&>) {
      pipe._callable(pf);
    }
    else if constexpr(std::is_invocable_v<callable_t, Pipeflow&, NonpreemptiveRuntime&>) {
      pipe._callable(pf, rt);
    }
    else {
      static_assert(dependent_false_v<callable_t>, "un-supported pipe callable type");
    }
  }, _pipes, pf._pipe);
}
 
// Procedure: _build
template <typename... Ps>
void Pipeline<Ps...>::_build() {
 
  using namespace std::literals::string_literals;
 
  FlowBuilder fb(_graph);
 
  // init task
  _tasks[0] = fb.emplace([this]() {
    return static_cast<int>(_num_tokens % num_lines());
  }).name("cond");
 
  // line task
  for(size_t l = 0; l < num_lines(); l++) {
 
    _tasks[l + 1] = fb.emplace([this, l] (tf::NonpreemptiveRuntime& rt) mutable {
 
      auto pf = &_pipeflows[l];
 
      pipeline:
 
      _lines[pf->_line][pf->_pipe].join_counter.store(
        static_cast<size_t>(_meta[pf->_pipe].type), std::memory_order_relaxed
      );
      
      // First pipe does all jobs of initialization and token dependencies
      if (pf->_pipe == 0) {
        pf->_token = _num_tokens;
        if (pf->_stop = false, _on_pipe(*pf, rt); pf->_stop == true) {
          // here, the pipeline is not stopped yet because other
          // lines of tasks may still be running their last stages
          return;
        }
        ++_num_tokens;
      }
      else {
        _on_pipe(*pf, rt);
      }
 
      size_t c_f = pf->_pipe;
      size_t n_f = (pf->_pipe + 1) % num_pipes();
      size_t n_l = (pf->_line + 1) % num_lines();
 
      pf->_pipe = n_f;
 
      // ---- scheduling starts here ----
      // Notice that the shared variable f must not be changed after this
      // point because it can result in data race due to the following
      // condition:
      //
      // a -> b
      // |    |
      // v    v
      // c -> d
      //
      // d will be spawned by either c or b, so if c changes f but b spawns d
      // then data race on f will happen
 
      std::array<int, 2> retval;
      size_t n = 0;
 
      // downward dependency
      if(_meta[c_f].type == PipeType::SERIAL &&
         _lines[n_l][c_f].join_counter.fetch_sub(
           1, std::memory_order_acq_rel) == 1
        ) {
        retval[n++] = 1;
      }
 
      // forward dependency
      if(_lines[pf->_line][n_f].join_counter.fetch_sub(
          1, std::memory_order_acq_rel) == 1
        ) {
        retval[n++] = 0;
      }
      
      // notice that the task index starts from 1
      switch(n) {
        case 2: {
          rt.schedule(_tasks[n_l+1]);
          goto pipeline;
        }
        case 1: {
          // downward dependency 
          if (retval[0] == 1) {
            pf = &_pipeflows[n_l];
          }
          // forward dependency
          goto pipeline;
        }
      }
    }).name("nprt-"s + std::to_string(l));
 
    _tasks[0].precede(_tasks[l+1]);
  }
}
 
// ----------------------------------------------------------------------------
// Class Definition: ScalablePipeline
// ----------------------------------------------------------------------------

template <typename P>
class ScalablePipeline {

  struct Line {
    std::atomic<size_t> join_counter;
  };
 
 
  public:

  using pipe_t = typename std::iterator_traits<P>::value_type;

  ScalablePipeline() = default;

  ScalablePipeline(size_t num_lines);

  ScalablePipeline(size_t num_lines, P first, P last);

  ScalablePipeline(const ScalablePipeline&) = delete;

  ScalablePipeline(ScalablePipeline&& rhs);

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

  ScalablePipeline& operator = (ScalablePipeline&& rhs);

  size_t num_lines() const noexcept;

  size_t num_pipes() const noexcept;

  void reset();

  void reset(P first, P last);

  void reset(size_t num_lines, P first, P last);

  size_t num_tokens() const noexcept;

  Graph& graph();
 
  private:
 
  Graph _graph;
 
  size_t _num_tokens{0};
 
  std::vector<P> _pipes;
  std::vector<Task> _tasks;
  std::vector<Pipeflow> _pipeflows;
  std::unique_ptr<Line[]> _lines;
 
  void _on_pipe(Pipeflow&, NonpreemptiveRuntime&);
  void _build();
 
  Line& _line(size_t, size_t);
};
 
// constructor
template <typename P>
ScalablePipeline<P>::ScalablePipeline(size_t num_lines) :
  _tasks     (num_lines + 1),
  _pipeflows (num_lines) {
 
  if(num_lines == 0) {
    TF_THROW("must have at least one line");
  }
 
  _build();
}
 
// constructor
template <typename P>
ScalablePipeline<P>::ScalablePipeline(size_t num_lines, P first, P last) :
  _tasks     (num_lines + 1),
  _pipeflows (num_lines) {
 
  if(num_lines == 0) {
    TF_THROW("must have at least one line");
  }
 
  reset(first, last);
  _build();
}
 
// move constructor
template <typename P>
ScalablePipeline<P>::ScalablePipeline(ScalablePipeline&& rhs):
  _num_tokens {rhs._num_tokens},
  _pipes      {std::move(rhs._pipes)},
  _pipeflows  {std::move(rhs._pipeflows)},
  _lines      {std::move(rhs._lines)} {
 
  _graph.clear();
  _tasks.resize(_pipeflows.size()+1);
  rhs._num_tokens = 0;
  rhs._tasks.clear();
  _build();
}
 
// move assignment operator
template <typename P>
ScalablePipeline<P>& ScalablePipeline<P>::operator = (ScalablePipeline&& rhs) {
  _num_tokens = rhs._num_tokens;
  _pipes      = std::move(rhs._pipes);
  _pipeflows  = std::move(rhs._pipeflows);
  _lines      = std::move(rhs._lines);
 
  _graph.clear();
  _tasks.resize(_pipeflows.size()+1);
 
  rhs._num_tokens = 0;
  rhs._tasks.clear();
  _build();
  return *this;
}
 
// Function: num_lines
template <typename P>
size_t ScalablePipeline<P>::num_lines() const noexcept {
  return _pipeflows.size();
}
 
// Function: num_pipes
template <typename P>
size_t ScalablePipeline<P>::num_pipes() const noexcept {
  return _pipes.size();
}
 
// Function: num_tokens
template <typename P>
size_t ScalablePipeline<P>::num_tokens() const noexcept {
  return _num_tokens;
}
 
// Function: graph
template <typename P>
Graph& ScalablePipeline<P>::graph() {
  return _graph;
}
 
// Function: _line
template <typename P>
typename ScalablePipeline<P>::Line& ScalablePipeline<P>::_line(size_t l, size_t p) {
  return _lines[l*num_pipes() + p];
}
 
template <typename P>
void ScalablePipeline<P>::reset(size_t num_lines, P first, P last) {
 
  if(num_lines == 0) {
    TF_THROW("must have at least one line");
  }
 
  _graph.clear();
  _tasks.resize(num_lines + 1);
  _pipeflows.resize(num_lines);
 
  reset(first, last);
 
  _build();
}
 
// Function: reset
template <typename P>
void ScalablePipeline<P>::reset(P first, P last) {
 
  size_t num_pipes = static_cast<size_t>(std::distance(first, last));
 
  if(num_pipes == 0) {
    TF_THROW("pipeline cannot be empty");
  }
 
  if(first->type() != PipeType::SERIAL) {
    TF_THROW("first pipe must be serial");
  }
 
  _pipes.resize(num_pipes);
 
  size_t i=0;
  for(auto itr = first; itr != last; itr++) {
    _pipes[i++] = itr;
  }
 
  _lines = std::make_unique<Line[]>(num_lines() * _pipes.size());
 
  reset();
}
 
// Function: reset
template <typename P>
void ScalablePipeline<P>::reset() {
 
  _num_tokens = 0;
 
  for(size_t l = 0; l<num_lines(); l++) {
    _pipeflows[l]._pipe = 0;
    _pipeflows[l]._line = l;
  }
 
  _line(0, 0).join_counter.store(0, std::memory_order_relaxed);
 
  for(size_t l=1; l<num_lines(); l++) {
    for(size_t f=1; f<num_pipes(); f++) {
      _line(l, f).join_counter.store(
        static_cast<size_t>(_pipes[f]->type()), std::memory_order_relaxed
      );
    }
  }
 
  for(size_t f=1; f<num_pipes(); f++) {
    _line(0, f).join_counter.store(1, std::memory_order_relaxed);
  }
 
  for(size_t l=1; l<num_lines(); l++) {
    _line(l, 0).join_counter.store(
      static_cast<size_t>(_pipes[0]->type()) - 1, std::memory_order_relaxed
    );
  }
}
 
// Procedure: _on_pipe
template <typename P>
void ScalablePipeline<P>::_on_pipe(Pipeflow& pf, NonpreemptiveRuntime& rt) {
    
  using callable_t = typename pipe_t::callable_t;
 
  if constexpr (std::is_invocable_v<callable_t, Pipeflow&>) {
    _pipes[pf._pipe]->_callable(pf);
  }
  else if constexpr(std::is_invocable_v<callable_t, Pipeflow&, NonpreemptiveRuntime&>) {
    _pipes[pf._pipe]->_callable(pf, rt);
  }
  else {
    static_assert(dependent_false_v<callable_t>, "un-supported pipe callable type");
  }
}
 
// Procedure: _build
template <typename P>
void ScalablePipeline<P>::_build() {
 
  using namespace std::literals::string_literals;
 
  FlowBuilder fb(_graph);
  
  // init task
  _tasks[0] = fb.emplace([this]() {
    return static_cast<int>(_num_tokens % num_lines());
  }).name("cond");
 
  // line task
  for(size_t l = 0; l < num_lines(); l++) {
    
    _tasks[l + 1] = fb.emplace([this, l] (tf::NonpreemptiveRuntime& rt) mutable {
 
      auto pf = &_pipeflows[l];
 
      pipeline:
 
      _line(pf->_line, pf->_pipe).join_counter.store(
        static_cast<size_t>(_pipes[pf->_pipe]->type()), std::memory_order_relaxed
      );
 
      // First pipe does all jobs of initialization and token dependencies
      if (pf->_pipe == 0) {
        pf->_token = _num_tokens;
        if (pf->_stop = false, _on_pipe(*pf, rt); pf->_stop == true) {
          // here, the pipeline is not stopped yet because other
          // lines of tasks may still be running their last stages
          return;
        }
        ++_num_tokens;
      }
      else {
        _on_pipe(*pf, rt);
      }
      
      size_t c_f = pf->_pipe;
      size_t n_f = (pf->_pipe + 1) % num_pipes();
      size_t n_l = (pf->_line + 1) % num_lines();
      
      pf->_pipe = n_f;
 
      // ---- scheduling starts here ----
      // Notice that the shared variable f must not be changed after this
      // point because it can result in data race due to the following
      // condition:
      //
      // a -> b
      // |    |
      // v    v
      // c -> d
      //
      // d will be spawned by either c or b, so if c changes f but b spawns d
      // then data race on f will happen
 
      std::array<int, 2> retval;
      size_t n = 0;
 
      // downward dependency
      if(_pipes[c_f]->type() == PipeType::SERIAL &&
         _line(n_l, c_f).join_counter.fetch_sub(
           1, std::memory_order_acq_rel) == 1
        ) {
        retval[n++] = 1;
      }
 
      // forward dependency
      if(_line(pf->_line, n_f).join_counter.fetch_sub(
          1, std::memory_order_acq_rel) == 1
        ) {
        retval[n++] = 0;
      }
 
      // notice that the task index starts from 1
      switch(n) {
        case 2: {
          rt.schedule(_tasks[n_l+1]);
          goto pipeline;
        }
        case 1: {
          if (retval[0] == 1) {
            pf = &_pipeflows[n_l];
          }
          goto pipeline;
        }
      }
    }).name("nprt-"s + std::to_string(l));
 
    _tasks[0].precede(_tasks[l+1]);
  }
}
 
}  // end of namespace tf -----------------------------------------------------