| Taskflow: A General-purpose Task-parallel Programming System

Loading...
Searching...
No Matches
Public Member Functions | List of all members
tf::Pipeline< Ps > Class Template Reference

class to create a pipeline scheduling framework More...

`#include <taskflow/algorithm/pipeline.hpp>`

## Public Member Functions
| Pipeline (size_t num_lines, Ps &&... ps)
| constructs a pipeline object
| Pipeline (size_t num_lines, std::tuple< Ps... > &&ps)
| constructs a pipeline object
size_t | num_lines () const noexcept
| queries the number of parallel lines
constexpr size_t | num_pipes () const
| queries the number of pipes
void | reset ()
| resets the pipeline
size_t | num_tokens () const noexcept
| queries the number of generated tokens in the pipeline
Graph & | graph ()
| obtains the graph object associated with the pipeline construct

## Detailed Description

template<typename... Ps>
class tf::Pipeline< Ps >

class to create a pipeline scheduling framework

Template Parameters

  Ps | pipe types

A tf::Pipeline is a composable graph object that allows users to parallelize an application using pipeline parallelism. Unlike conventional pipeline programming frameworks (e.g., Intel TBB), tf::Pipeline does not provide any data abstraction but a task-parallel framework for users to customize their data layout when leveraging pipeline parallelism. The following code creates a pipeline with four parallel lines that schedule five tokens through three pipes of a serial-parallel-serial direction:

tf::Taskflow taskflow;
tf::Executor executor;
const size_t num_lines = 4;
const size_t num_pipes = 3;
// create a custom data buffer
std::array<std::array<int, num_pipes>, num_lines> buffer;
// create a pipeline graph of four concurrent lines and three serial pipes
tf::Pipeline pipeline(num_lines,
// first pipe must define a serial direction
tf::Pipe{tf::PipeType::SERIAL, [&buffer](tf::Pipeflow& pf) {
// generate only 5 scheduling tokens
if(pf.token() == 5) {
pf.stop();
}
// save the token id into the buffer
else {
buffer[pf.line()][pf.pipe()] = pf.token();
}
}},
tf::Pipe{tf::PipeType::PARALLEL, [&buffer] (tf::Pipeflow& pf) {
// propagate the previous result to this pipe by adding one
buffer[pf.line()][pf.pipe()] = buffer[pf.line()][pf.pipe()-1] + 1;
}},
tf::Pipe{tf::PipeType::SERIAL, [&buffer](tf::Pipeflow& pf){
// propagate the previous result to this pipe by adding one
buffer[pf.line()][pf.pipe()] = buffer[pf.line()][pf.pipe()-1] + 1;
}}
);
// build the pipeline graph using composition
tf::Task init = taskflow.emplace([](){ std::cout << "ready\n"; })
.name("starting pipeline");
tf::Task task = taskflow.composed_of(pipeline)
.name("pipeline");
tf::Task stop = taskflow.emplace([](){ std::cout << "stopped\n"; })
.name("pipeline stopped");
// create task dependency
init.precede(task);
task.precede(stop);
// run the pipeline
executor.run(taskflow).wait();
tf::Executor
class to create an executor
Definition executor.hpp:62
tf::Executor::run
tf::Future< void > run(Taskflow &taskflow)
runs a taskflow once
tf::FlowBuilder::emplace
Task emplace(C &&callable)
creates a static task
Definition flow_builder.hpp:1738
tf::FlowBuilder::composed_of
Task composed_of(T &object)
creates a module task for the target object
Definition flow_builder.hpp:1788
tf::Pipe
class to create a pipe object for a pipeline stage
Definition pipeline.hpp:144
tf::Pipeflow
class to create a pipeflow object used by the pipe callable
Definition pipeline.hpp:43
tf::Pipeline
class to create a pipeline scheduling framework
Definition pipeline.hpp:307
tf::Pipeline::num_lines
size_t num_lines() const noexcept
queries the number of parallel lines
Definition pipeline.hpp:469
tf::Pipeline::num_pipes
constexpr size_t num_pipes() const
queries the number of pipes
Definition pipeline.hpp:475
tf::Task
class to create a task handle over a taskflow node
Definition task.hpp:569
tf::Task::name
const std::string & name() const
queries the name of the task
Definition task.hpp:1388
tf::Task::precede
Task & precede(Ts &&... tasks)
adds precedence links from this to other tasks
Definition task.hpp:1258
tf::Taskflow
class to create a taskflow object
Definition taskflow.hpp:64
tf::PipeType::SERIAL
@ SERIAL
serial type
tf::PipeType::PARALLEL
@ PARALLEL
parallel type

The five tokens will be scheduled across four parallel lines in a circular fashion, as depicted below:

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.

## Constructor & Destructor Documentation

## ◆ Pipeline() [1/2]

template<typename... Ps>

tf::Pipeline< Ps >::Pipeline | ( | size_t | num_lines,
|  | Ps &&... | ps )

constructs a pipeline object

Parameters

  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.

## ◆ Pipeline() [2/2]

template<typename... Ps>

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

constructs a pipeline object

Parameters

  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.

## Member Function Documentation

## ◆ graph()

template<typename... Ps>

Graph & tf::Pipeline< Ps >::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.

## ◆ num_lines()

template<typename... Ps>

size_t tf::Pipeline< Ps >::num_lines | ( |  | ) | const
| noexcept

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.

## ◆ num_pipes()

template<typename... Ps>

size_t tf::Pipeline< Ps >::num_pipes | ( |  | ) | const
| constexpr

queries the number of pipes

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

## ◆ num_tokens()

template<typename... Ps>

size_t tf::Pipeline< Ps >::num_tokens | ( |  | ) | const
| noexcept

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.

## ◆ reset()

template<typename... Ps>

void tf::Pipeline< Ps >::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.

The documentation for this class was generated from the following file:

- taskflow/algorithm/pipeline.hpp