pipeline.hpp

Go to the documentation of this file.

#pragma once
 
#include "../taskflow.hpp"
 
namespace tf {
 
// ----------------------------------------------------------------------------
// Class Definition: Pipeflow
// ----------------------------------------------------------------------------
 
class Pipeflow {
 
  template <typename... Ps>
  friend class Pipeline;
 
  template <typename P>
  friend class ScalablePipeline;
 
  public:
 
  Pipeflow() = default;
 
  size_t line() const {
    return _line;
  }
 
  size_t pipe() const {
    return _pipe;
  }
 
  size_t token() const {
    return _token;
  }
 
  void stop() {
    _stop = true;
  }
 
  private:
 
  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 noexcept;
 
  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&, Runtime&);
  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 noexcept {
  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, Runtime&) {
  visit_tuple([&](auto&& pipe){
    pipe._callable(pf);
  }, _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::Runtime& 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
      );
 
      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: {
          if (retval[0] == 1) {
            pf = &_pipeflows[n_l];
          }
          goto pipeline;
        }
      }
    }).name("rt-"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_type = 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&, Runtime&);
  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) :
  _graph      {std::move(rhs._graph)},
  _num_tokens {rhs._num_tokens},
  _pipes      {std::move(rhs._pipes)},
  _tasks      {std::move(rhs._tasks)},
  _pipeflows  {std::move(rhs._pipeflows)},
  _lines      {std::move(rhs._lines)} {
 
  rhs._num_tokens = 0;
}
 
// move assignment operator
template <typename P>
ScalablePipeline<P>& ScalablePipeline<P>::operator = (ScalablePipeline&& rhs) {
  _graph      = std::move(rhs._graph);
  _num_tokens = rhs._num_tokens;
  _pipes      = std::move(rhs._pipes);
  _tasks      = std::move(rhs._tasks);
  _pipeflows  = std::move(rhs._pipeflows);
  _lines      = std::move(rhs._lines);
  rhs._num_tokens = 0;
  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, Runtime&) {
  _pipes[pf._pipe]->_callable(pf);
}
 
// 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::Runtime& 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
      );
 
      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("rt-"s + std::to_string(l));
 
    _tasks[0].precede(_tasks[l+1]);
  }
}
 
}  // end of namespace tf -----------------------------------------------------