tf::ScalablePipeline
taskflow/algorithm/pipeline.hpp
tf::ScalablePipeline::Line
typename P
typename std::iterator_traits< P >::value_type using tf::ScalablePipeline< P >::pipe_t = typename std::iterator_traits<P>::value_type pipe_t tf::ScalablePipeline::pipe_t

pipe type

Graph Graph tf::ScalablePipeline< P >::_graph _graph tf::ScalablePipeline::_graph
size_t size_t tf::ScalablePipeline< P >::_num_tokens _num_tokens tf::ScalablePipeline::_num_tokens {0}
std::vector< P > std::vector<P> tf::ScalablePipeline< P >::_pipes _pipes tf::ScalablePipeline::_pipes
std::vector< Task > std::vector<Task> tf::ScalablePipeline< P >::_tasks _tasks tf::ScalablePipeline::_tasks
std::vector< Pipeflow > std::vector<Pipeflow> tf::ScalablePipeline< P >::_pipeflows _pipeflows tf::ScalablePipeline::_pipeflows
std::unique_ptr< Line[]> std::unique_ptr<Line[]> tf::ScalablePipeline< P >::_lines _lines tf::ScalablePipeline::_lines
std::queue< std::pair< size_t, size_t > > std::queue<std::pair<size_t, size_t> > tf::ScalablePipeline< P >::_ready_tokens _ready_tokens tf::ScalablePipeline::_ready_tokens
std::unordered_map< size_t, std::vector< size_t > > std::unordered_map<size_t, std::vector<size_t> > tf::ScalablePipeline< P >::_token_dependencies _token_dependencies tf::ScalablePipeline::_token_dependencies
std::unordered_map< size_t, DeferredPipeflow > std::unordered_map<size_t, DeferredPipeflow> tf::ScalablePipeline< P >::_deferred_tokens _deferred_tokens tf::ScalablePipeline::_deferred_tokens
size_t size_t tf::ScalablePipeline< P >::_longest_deferral _longest_deferral tf::ScalablePipeline::_longest_deferral = 0
tf::ScalablePipeline< P >::ScalablePipeline
()=default
ScalablePipeline
tf::ScalablePipeline::ScalablePipeline

default constructor

tf::ScalablePipeline< P >::ScalablePipeline
(size_t num_lines)
ScalablePipeline
tf::ScalablePipeline::ScalablePipeline
size_t num_lines

constructs an empty scalable pipeline object

num_lines

the number of parallel lines

An empty scalable pipeline does not have any pipes. The pipeline needs to be reset to a valid range of pipes before running.
tf::ScalablePipeline< P >::ScalablePipeline
(size_t num_lines, P first, P last)
ScalablePipeline
tf::ScalablePipeline::ScalablePipeline
size_t num_lines
P first
P last

constructs a scalable pipeline object

num_lines

the number of parallel lines

first

iterator to the beginning of the range

last

iterator to the end of the range

Constructs a pipeline from the given range of pipes specified in `[first, last)` using `num_lines` parallel lines. The first pipe must define a serial direction (tf::PipeType::SERIAL) or an exception will be thrown.

Internally, the scalable pipeline copies the iterators from the specified range. Those pipe callables pointed to by these iterators must remain valid during the execution of the pipeline.

tf::ScalablePipeline< P >::ScalablePipeline
(const ScalablePipeline &)=delete
ScalablePipeline
tf::ScalablePipeline::ScalablePipeline
const ScalablePipeline &

disabled copy constructor

tf::ScalablePipeline< P >::ScalablePipeline
(ScalablePipeline &&rhs)
ScalablePipeline
tf::ScalablePipeline::ScalablePipeline
ScalablePipeline && rhs

move constructor

Constructs a pipeline from the given `rhs` using move semantics (i.e. the data in `rhs` is moved into this pipeline). After the move, `rhs` is in a state as if it is just constructed. The behavior is undefined if `rhs` is running during the move.

ScalablePipeline & ScalablePipeline & tf::ScalablePipeline< P >::operator= (const ScalablePipeline &)=delete operator= tf::ScalablePipeline::operator= const ScalablePipeline &

disabled copy assignment operator

ScalablePipeline & ScalablePipeline< P > & tf::ScalablePipeline< P >::operator= (ScalablePipeline &&rhs) operator= tf::ScalablePipeline::operator= ScalablePipeline && rhs

move constructor

Replaces the contents with those of `rhs` using move semantics (i.e. the data in `rhs` is moved into this pipeline). After the move, `rhs` is in a state as if it is just constructed. The behavior is undefined if `rhs` is running during the move.

size_t size_t tf::ScalablePipeline< P >::num_lines () const noexcept num_lines tf::ScalablePipeline::num_lines

queries the number of parallel lines

The function returns the number of parallel lines given by the user upon the construction of the pipeline. The number of lines represents the maximum parallelism this pipeline can achieve.

size_t size_t tf::ScalablePipeline< P >::num_pipes () const noexcept num_pipes tf::ScalablePipeline::num_pipes

queries the number of pipes

The Function returns the number of pipes given by the user upon the construction of the pipeline.

void void tf::ScalablePipeline< P >::reset () reset tf::ScalablePipeline::reset

resets the pipeline

Resets the pipeline to the initial state. After resetting a pipeline, its token identifier will start from zero.

void void tf::ScalablePipeline< P >::reset (P first, P last) reset tf::ScalablePipeline::reset P first P last

resets the pipeline with a new range of pipes

first

iterator to the beginning of the range

last

iterator to the end of the range

The member function assigns the pipeline to a new range of pipes specified in `[first, last)` and resets the pipeline to the initial state. After resetting a pipeline, its token identifier will start from zero.

Internally, the scalable pipeline copies the iterators from the specified range. Those pipe callables pointed to by these iterators must remain valid during the execution of the pipeline.

void void tf::ScalablePipeline< P >::reset (size_t num_lines, P first, P last) reset tf::ScalablePipeline::reset size_t num_lines P first P last

resets the pipeline to a new line number and a new range of pipes

num_lines

number of parallel lines

first

iterator to the beginning of the range

last

iterator to the end of the range

The member function resets the pipeline to a new number of parallel lines and a new range of pipes specified in `[first, last)`, as if the pipeline is just constructed. After resetting a pipeline, its token identifier will start from zero.

Internally, the scalable pipeline copies the iterators from the specified range. Those pipe callables pointed to by these iterators must remain valid during the execution of the pipeline.

size_t size_t tf::ScalablePipeline< P >::num_tokens () const noexcept num_tokens tf::ScalablePipeline::num_tokens

queries the number of generated tokens in the pipeline

The number represents the total scheduling tokens that has been generated by the pipeline so far.

Graph & Graph & tf::ScalablePipeline< P >::graph () graph tf::ScalablePipeline::graph

obtains the graph object associated with the pipeline construct

This method is primarily used as an opaque data structure for creating a module task of the this pipeline.

void void tf::ScalablePipeline< P >::_check_dependents (Pipeflow &) _check_dependents tf::ScalablePipeline::_check_dependents Pipeflow & pf
void void tf::ScalablePipeline< P >::_construct_deferred_tokens (Pipeflow &) _construct_deferred_tokens tf::ScalablePipeline::_construct_deferred_tokens Pipeflow & pf
void void tf::ScalablePipeline< P >::_resolve_token_dependencies (Pipeflow &) _resolve_token_dependencies tf::ScalablePipeline::_resolve_token_dependencies Pipeflow & pf
void void tf::ScalablePipeline< P >::_on_pipe (Pipeflow &, Runtime &) _on_pipe tf::ScalablePipeline::_on_pipe Pipeflow & pf Runtime & rt
void void tf::ScalablePipeline< P >::_build () _build tf::ScalablePipeline::_build
Line & ScalablePipeline< P >::Line & tf::ScalablePipeline< P >::_line (size_t, size_t) _line tf::ScalablePipeline::_line size_t l size_t p

class to create a scalable pipeline object

P

type of the iterator to a range of pipes

A scalable pipeline is a composable graph object for users to create a pipeline scheduling framework using a module task in a taskflow. Unlike tf::Pipeline that instantiates all pipes upon the construction time, tf::ScalablePipeline allows variable assignments of pipes using range iterators. Users can also reset a scalable pipeline to a different range of pipes between runs. The following code creates a scalable pipeline of four parallel lines to schedule tokens through three serial pipes in a custom storage, then resetting the pipeline to a new range of five serial pipes:

    tf::Taskflowtaskflow("pipeline");
    tf::Executorexecutor;

    constsize_tnum_lines=4;

    //createdatastorage
    std::array<int,num_lines>buffer;

    //definethepipecallable
    autopipe_callable=[&buffer](tf::Pipeflow&pf)mutable{
    switch(pf.pipe()){
    //firststagegeneratesonly5schedulingtokensandsavesthe
    //tokennumberintothebuffer.
    case0:{
    if(pf.token()==5){
    pf.stop();
    }
    else{
    printf("stage1:inputtoken=%zu\n",pf.token());
    buffer[pf.line()]=pf.token();
    }
    return;
    }
    break;

    //otherstagespropagatethepreviousresulttothispipeand
    //incrementitbyone
    default:{
    printf(
    "stage%zu:inputbuffer[%zu]=%d\n",pf.pipe(),pf.line(),buffer[pf.line()]
    );
    buffer[pf.line()]=buffer[pf.line()]+1;
    }
    break;
    }
    };

    //createavectorofthreepipes
    std::vector<tf::Pipe<std::function<void(tf::Pipeflow&)>>>pipes;

    for(size_ti=0;i<3;i++){
    pipes.emplace_back(tf::PipeType::SERIAL,pipe_callable);
    }

    //createapipelineoffourparallellinesbasedonthegivenvectorofpipes
    tf::ScalablePipelinepl(num_lines,pipes.begin(),pipes.end());

    //buildthepipelinegraphusingcomposition
    tf::Taskinit=taskflow.emplace([](){std::cout<<"ready\n";})
    .name("startingpipeline");
    tf::Tasktask=taskflow.composed_of(pl)
    .name("pipeline");
    tf::Taskstop=taskflow.emplace([](){std::cout<<"stopped\n";})
    .name("pipelinestopped");

    //createtaskdependency
    init.precede(task);
    task.precede(stop);

    //dumpthepipelinegraphstructure(withcomposition)
    taskflow.dump(std::cout);

    //runthepipeline
    executor.run(taskflow).wait();

    //resetthepipelinetoanewrangeoffivepipesandstartsfrom
    //theinitialstate(i.e.,tokencountsfromzero)
    for(size_ti=0;i<2;i++){
    pipes.emplace_back(tf::PipeType::SERIAL,pipe_callable);
    }
    pl.reset(pipes.begin(),pipes.end());

    executor.run(taskflow).wait();

The above example creates a pipeline graph that schedules five tokens over four parallel lines in a circular fashion, first going through three serial pipes and then five serial pipes:

    {.shell-session}
    #initialconstructionofthreeserialpipes
    o->o->o
    |||
    vvv
    o->o->o
    |||
    vvv
    o->o->o
    |||
    vvv
    o->o->o

    #resettingtoanewrangeoffiveserialpipes
    o->o->o->o->o
    |||||
    vvvvv
    o->o->o->o->o
    |||||
    vvvvv
    o->o->o->o->o
    |||||
    vvvvv
    o->o->o->o->o

Each pipe has the same type of `tf::Pipe<std::function<void(tf::Pipeflow&)>>` and is kept in a vector that is amenable to change. We construct the scalable pipeline using two range iterators pointing to the beginning and the end of the vector. At each pipe stage, the program propagates the result to the next pipe by adding one to the result stored in a custom data storage, `buffer`. The pipeline scheduler will generate five scheduling tokens and then stop.

A scalable pipeline is move-only.

- tf::ScalablePipeline
  _build
- tf::ScalablePipeline
  _check_dependents
- tf::ScalablePipeline
  _construct_deferred_tokens
- tf::ScalablePipeline
  _deferred_tokens
- tf::ScalablePipeline
  _graph
- tf::ScalablePipeline
  _line
- tf::ScalablePipeline
  _lines
- tf::ScalablePipeline
  _longest_deferral
- tf::ScalablePipeline
  _num_tokens
- tf::ScalablePipeline
  _on_pipe
- tf::ScalablePipeline
  _pipeflows
- tf::ScalablePipeline
  _pipes
- tf::ScalablePipeline
  _ready_tokens
- tf::ScalablePipeline
  _resolve_token_dependencies
- tf::ScalablePipeline
  _tasks
- tf::ScalablePipeline
  _token_dependencies
- tf::ScalablePipeline
  graph
- tf::ScalablePipeline
  num_lines
- tf::ScalablePipeline
  num_pipes
- tf::ScalablePipeline
  num_tokens
- tf::ScalablePipeline
  operator=
- tf::ScalablePipeline
  operator=
- tf::ScalablePipeline
  pipe_t
- tf::ScalablePipeline
  reset
- tf::ScalablePipeline
  reset
- tf::ScalablePipeline
  reset
- tf::ScalablePipeline
  ScalablePipeline
- tf::ScalablePipeline
  ScalablePipeline
- tf::ScalablePipeline
  ScalablePipeline
- tf::ScalablePipeline
  ScalablePipeline
- tf::ScalablePipeline
  ScalablePipeline