docs/classtf_1_1ScalablePipeline.html
class to create a scalable pipeline object
| Template parameters |
|---|
| P |
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::Taskflow taskflow("pipeline");tf::Executor executor;const size\_t num\_lines = 4;// create data storagestd::array\<int, num\_lines\> buffer;// define the pipe callableauto pipe\_callable = [&buffer] (tf::Pipeflow& pf) mutable {switch(pf.pipe()) {// first stage generates only 5 scheduling tokens and saves the// token number into the buffer.case 0: {if(pf.token() == 5) {pf.stop();}else {printf("stage 1: input token = %zu\n", pf.token());buffer[pf.line()] = pf.token();}return;}break;// other stages propagate the previous result to this pipe and// increment it by onedefault: {printf("stage %zu: input buffer[%zu] = %d\n", pf.pipe(), pf.line(), buffer[pf.line()]);buffer[pf.line()] = buffer[pf.line()] + 1;}break;}};// create a vector of three pipesstd::vector\< tf::Pipe\<std::function\<void(tf::Pipeflow&)\>\> \> pipes;for(size\_t i=0; i\<3; i++) {pipes.emplace\_back(tf::PipeType::SERIAL, pipe\_callable);}// create a pipeline of four parallel lines based on the given vector of pipestf::ScalablePipeline pl(num\_lines, pipes.begin(), pipes.end());// build the pipeline graph using compositiontf::Task init = taskflow.emplace([](){ std::cout \<\< "ready\n"; }).name("starting pipeline");tf::Task task = taskflow.composed\_of(pl).name("pipeline");tf::Task stop = taskflow.emplace([](){ std::cout \<\< "stopped\n"; }).name("pipeline stopped");// create task dependencyinit.precede(task);task.precede(stop);// dump the pipeline graph structure (with composition)taskflow.dump(std::cout);// run the pipelineexecutor.run(taskflow).wait();// reset the pipeline to a new range of five pipes and starts from// the initial state (i.e., token counts from zero)for(size\_t i=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:
# initial construction of three serial pipeso -> o -> o|||v v v
o -> o -> o|||v v v
o -> o -> o|||v v v
o -> o -> o# resetting to a new range of five serial pipeso -> o -> o -> o -> o|||||v v v v v
o -> o -> o -> o -> o|||||v v v v v
o -> o -> o -> o -> o|||||v v v v v
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.
using pipe_t = typename std::iterator_traits<P>::value_type pipe type
ScalablePipeline() defaulteddefault constructorScalablePipeline(size_t num_lines)constructs an empty scalable pipeline objectScalablePipeline(size_t num_lines, P first, P last)constructs a scalable pipeline objectScalablePipeline(const ScalablePipeline&) deleteddisabled copy constructorScalablePipeline(ScalablePipeline&& rhs)move constructor
auto operator=(const ScalablePipeline&) -> ScalablePipeline& deleteddisabled copy assignment operatorauto operator=(ScalablePipeline&& rhs) -> ScalablePipeline&move constructorauto num_lines() const -> size_t noexceptqueries the number of parallel linesauto num_pipes() const -> size_t noexceptqueries the number of pipesvoid reset()resets the pipelinevoid reset(P first, P last)resets the pipeline with a new range of pipesvoid reset(size_t num_lines, P first, P last)resets the pipeline to a new line number and a new range of pipesauto num_tokens() const -> size_t noexceptqueries the number of generated tokens in the pipelineauto graph() -> Graph&obtains the graph object associated with the pipeline construct
constructs an empty scalable pipeline object
| Parameters |
|---|
| num_lines |
An empty scalable pipeline does not have any pipes. The pipeline needs to be reset to a valid range of pipes before running.
constructs a scalable pipeline object
| Parameters |
|---|
| num_lines |
| first |
| last |
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.
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.
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.
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.
queries the number of pipes
The Function returns the number of pipes given by the user upon the construction of the pipeline.
resets the pipeline
Resets the pipeline to the initial state. After resetting a pipeline, its token identifier will start from zero.
resets the pipeline with a new range of pipes
| Parameters |
|---|
| first |
| last |
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.
resets the pipeline to a new line number and a new range of pipes
| Parameters |
|---|
| num_lines |
| first |
| last |
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.
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.
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.