tf::Pipeline
taskflow/algorithm/pipeline.hpp
tf::Pipeline::Line
tf::Pipeline::PipeMeta
typename... Ps Ps
Graph Graph tf::Pipeline< Ps >::_graph _graph tf::Pipeline::_graph
size_t size_t tf::Pipeline< Ps >::_num_tokens _num_tokens tf::Pipeline::_num_tokens
std::tuple< Ps... > std::tuple<Ps...> tf::Pipeline< Ps >::_pipes _pipes tf::Pipeline::_pipes
std::array< PipeMeta, sizeof...(Ps)> std::array<PipeMeta, sizeof...(Ps)> tf::Pipeline< Ps >::_meta _meta tf::Pipeline::_meta
std::vector< std::array< Line, sizeof...(Ps)> > std::vector<std::array<Line, sizeof...(Ps)> > tf::Pipeline< Ps >::_lines _lines tf::Pipeline::_lines
std::vector< Task > std::vector<Task> tf::Pipeline< Ps >::_tasks _tasks tf::Pipeline::_tasks
std::vector< Pipeflow > std::vector<Pipeflow> tf::Pipeline< Ps >::_pipeflows _pipeflows tf::Pipeline::_pipeflows
std::queue< std::pair< size_t, size_t > > std::queue<std::pair<size_t, size_t> > tf::Pipeline< Ps >::_ready_tokens _ready_tokens tf::Pipeline::_ready_tokens
std::unordered_map< size_t, std::vector< size_t > > std::unordered_map<size_t, std::vector<size_t> > tf::Pipeline< Ps >::_token_dependencies _token_dependencies tf::Pipeline::_token_dependencies
std::unordered_map< size_t, DeferredPipeflow > std::unordered_map<size_t, DeferredPipeflow> tf::Pipeline< Ps >::_deferred_tokens _deferred_tokens tf::Pipeline::_deferred_tokens
size_t size_t tf::Pipeline< Ps >::_longest_deferral _longest_deferral tf::Pipeline::_longest_deferral = 0
tf::Pipeline< Ps >::Pipeline
(size_t num_lines, Ps &&... ps)
Pipeline
tf::Pipeline::Pipeline
size_t num_lines
Ps &&... ps

constructs a pipeline object

num_lines

the number of parallel lines

ps

a list of pipes

Constructs a pipeline of up to `num_lines` parallel lines to schedule tokens through the given linear chain of pipes. The first pipe must define a serial direction (tf::PipeType::SERIAL) or an exception will be thrown.
tf::Pipeline< Ps >::Pipeline
(size_t num_lines, std::tuple< Ps... > &&ps)
Pipeline
tf::Pipeline::Pipeline
size_t num_lines
std::tuple< Ps... > && ps

constructs a pipeline object

num_lines

the number of parallel lines

ps

a tuple of pipes

Constructs a pipeline of up to `num_lines` parallel lines to schedule tokens through the given linear chain of pipes. The first pipe must define a serial direction (tf::PipeType::SERIAL) or an exception will be thrown.
size_t size_t tf::Pipeline< Ps >::num_lines () const noexcept num_lines tf::Pipeline::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::Pipeline< Ps >::num_pipes () const noexcept num_pipes tf::Pipeline::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::Pipeline< Ps >::reset () reset tf::Pipeline::reset

resets the pipeline

Resetting the pipeline to the initial state. After resetting a pipeline, its token identifier will start from zero as if the pipeline was just constructed.

size_t size_t tf::Pipeline< Ps >::num_tokens () const noexcept num_tokens tf::Pipeline::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::Pipeline< Ps >::graph () graph tf::Pipeline::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.

size_t... I I
auto auto tf::Pipeline< Ps >::_gen_meta (std::tuple< Ps... > &&, std::index_sequence< I... >) _gen_meta tf::Pipeline::_gen_meta std::tuple< Ps... > && ps std::index_sequence< I... >
void void tf::Pipeline< Ps >::_on_pipe (Pipeflow &, Runtime &) _on_pipe tf::Pipeline::_on_pipe Pipeflow & pf Runtime & rt
void void tf::Pipeline< Ps >::_build () _build tf::Pipeline::_build
void void tf::Pipeline< Ps >::_check_dependents (Pipeflow &) _check_dependents tf::Pipeline::_check_dependents Pipeflow & pf
void void tf::Pipeline< Ps >::_construct_deferred_tokens (Pipeflow &) _construct_deferred_tokens tf::Pipeline::_construct_deferred_tokens Pipeflow & pf
void void tf::Pipeline< Ps >::_resolve_token_dependencies (Pipeflow &) _resolve_token_dependencies tf::Pipeline::_resolve_token_dependencies Pipeflow & pf

class to create a pipeline scheduling framework

Ps

pipe types

A pipeline is a composable graph object for users to create a pipeline scheduling framework using a module task in a taskflow. Unlike the conventional pipeline programming frameworks (e.g., Intel TBB), Taskflow's pipeline algorithm does not provide any data abstraction, which often restricts users from optimizing data layouts in their applications, but a flexible framework for users to customize their application data atop our pipeline scheduling. The following code creates a pipeline of four parallel lines to schedule tokens through three serial pipes:

    tf::Taskflowtaskflow;
    tf::Executorexecutor;

    constsize_tnum_lines=4;
    constsize_tnum_pipes=3;

    //createacustomdatabuffer
    std::array<std::array<int,num_pipes>,num_lines>buffer;

    //createapipelinegraphoffourconcurrentlinesandthreeserialpipes
    tf::Pipelinepipeline(num_lines,
    //firstpipemustdefineaserialdirection
    tf::Pipe{tf::PipeType::SERIAL,[&buffer](tf::Pipeflow&pf){
    //generateonly5schedulingtokens
    if(pf.token()==5){
    pf.stop();
    }
    //savethetokenidintothebuffer
    else{
    buffer[pf.line()][pf.pipe()]=pf.token();
    }
    }},
    tf::Pipe{tf::PipeType::SERIAL,[&buffer](tf::Pipeflow&pf){
    //propagatethepreviousresulttothispipebyaddingone
    buffer[pf.line()][pf.pipe()]=buffer[pf.line()][pf.pipe()-1]+1;
    }},
    tf::Pipe{tf::PipeType::SERIAL,[&buffer](tf::Pipeflow&pf){
    //propagatethepreviousresulttothispipebyaddingone
    buffer[pf.line()][pf.pipe()]=buffer[pf.line()][pf.pipe()-1]+1;
    }}
    );

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

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

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

The above example creates a pipeline graph that schedules five tokens over four parallel lines in a circular fashion, as depicted below:

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

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.

Internally, tf::Pipeline uses std::tuple to store the given sequence of pipes. The definition of each pipe can be different, completely decided by the compiler to optimize the object layout. After a pipeline is constructed, it is not possible to change its pipes. If applications need to change these pipes, please use tf::ScalablePipeline.

- tf::Pipeline
  _build
- tf::Pipeline
  _check_dependents
- tf::Pipeline
  _construct_deferred_tokens
- tf::Pipeline
  _deferred_tokens
- tf::Pipeline
  _gen_meta
- tf::Pipeline
  _graph
- tf::Pipeline
  _lines
- tf::Pipeline
  _longest_deferral
- tf::Pipeline
  _meta
- tf::Pipeline
  _num_tokens
- tf::Pipeline
  _on_pipe
- tf::Pipeline
  _pipeflows
- tf::Pipeline
  _pipes
- tf::Pipeline
  _ready_tokens
- tf::Pipeline
  _resolve_token_dependencies
- tf::Pipeline
  _tasks
- tf::Pipeline
  _token_dependencies
- tf::Pipeline
  graph
- tf::Pipeline
  num_lines
- tf::Pipeline
  num_pipes
- tf::Pipeline
  num_tokens
- tf::Pipeline
  Pipeline
- tf::Pipeline
  Pipeline
- tf::Pipeline
  reset