TaskParallelScalablePipeline

# Task-parallel Scalable Pipeline

Include the Header

Include the Header

TaskParallelScalablePipeline_1IncludeTheScalablePipelineHeader
Create a Scalable Pipeline Module Task

Create a Scalable Pipeline Module Task

TaskParallelScalablePipeline_1CreateAScalablePipelineModuleTask
Reset a Placeholder Scalable Pipeline

Reset a Placeholder Scalable Pipeline

TaskParallelScalablePipeline_1ResetAPlaceholderScalablePipeline
Use Other Iterator Types

Use Other Iterator Types

TaskParallelScalablePipeline_1ScalablePipelineUseOtherIteratorTypes
Learn More about Taskflow Pipeline

Learn More about Taskflow Pipeline

TaskParallelScalablePipeline_1ParallelScalablePipelineLearnMore

Unlike tf::Pipeline (see Task-parallel Pipeline) that instantiates all pipes at the construction time, Taskflow provides a scalable alternative called tf::ScalablePipeline to allow variable assignments of pipes using range iterators. A scalable pipeline is thus more flexible for applications to create a pipeline scheduling framework whose pipeline structure depends on runtime variables.

# Include the Header

You need to include the header file, `taskflow/algorithm/pipeline.hpp`, for creating a scalable pipeline scheduling framework.

    #include<taskflow/algorithm/pipeline.hpp>

# Create a Scalable Pipeline Module Task

Similar to tf::Pipeline, tf::ScalablePipeline is a composable graph object to implement a pipeline scheduling framework in a taskflow. The key difference between tf::Pipeline and tf::ScalablePipeline is that a scalable pipeline can accept variable assignments of pipes rather than instantiating all pipes at construction or programming time. Users define a linear range of pipes, each of the same callable type, and apply that range to construct a scalable pipeline. Between successive runs, users can reset the pipeline to a different range of pipes. The following code creates a scalable pipeline that uses four parallel lines to schedule tokens through three serial pipes in the given vector, then resetting that 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 program defines a uniform pipe type of `tf::Pipe<std::function<void(tf::Pipeflow&)>>` and keep all pipes in a vector that is amenable to change. Then, it constructs a scalable pipeline using two range iterators, `[first, last)`, that point to the beginning and the end of the pipe vector, resulting in a pipeline of three serial stages:

Then, the program appends another two pipes into the vector and resets the pipeline to the new range of two additional pipes, resulting in a pipeline of five serial stages:

When resetting a scalable pipeline to a new range, it will start from the initial state as if it has just been constructed, i.e., the token number counts from zero.

Unlike tf::Pipeline that keeps the given pipes in a std::tuple object, tf::ScalablePipeline does not own the given pipe but maintains a vector of iterators to each pipe in the given range. It is your responsibility to keep those pipe objects alive during the execution of the pipeline task.

# Reset a Placeholder Scalable Pipeline

It is possible to create a scalable pipeline as a placeholder using the constructor tf::ScalablePipeline(size_t num_lines) and reset it to another range later in the application. The following code creates a task to emplace a range of pipes and reset the pipeline to that range, before running the pipeline task:

    tf::Executorexecutor;
    tf::Taskflowtaskflow;

    size_tnum_pipes=10;
    size_tnum_lines=10;

    std::vector<tf::Pipe<std::function<void(tf::Pipeflow&)>>>pipes;
    tf::ScalablePipeline<typenamedecltype(pipes)::iterator>spl(num_lines);

    tf::Taskinit=taskflow.emplace([&](){
    for(size_ti=0;i<num_pipes;i++){
    pipes.emplace_back(tf::PipeType::SERIAL,[&](tf::Pipeflow&pf){
    if(pf.pipe()==0&&pf.token()==1024){
    pf.stop();
    return;
    }
    });
    }
    spl.reset(pipes.begin(),pipes.end());
    }).name("init");

    tf::Taskpipeline=taskflow.composed_of(spl).name("pipeline");
    pipeline.succeed(init);
    executor.run(taskflow).wait();

The task graph of this program is shown below:

It is your responsibility to ensure a scalable pipeline has a valid structure before running it. A valid pipeline must have at least one parallel line and one pipe, where the first pipe is a serial type.

Similarly, you can create an empty scalable pipeline using the default constructor tf::ScalablePipeline() and reset it later in your program.

    std::vector<tf::Pipe<std::function<void(tf::Pipeflow&)>>>pipes;
    tf::ScalablePipeline<typenamedecltype(pipes)::iterator>spl;
    //createpipes...
    spl.reset(num_lines,pipes.begin(),pipes.end());

# Use Other Iterator Types

When assigning a range to a scalable pipeline, the pipeline fetches all pipe iterators in that range to an internal vector. This organization allows invoking a pipe callable to be a random accessible operation, regardless of the pipe container type. Taskflow does not have much restriction on the iterator type, as long as these pipes can be iterated in a sequential order using the postfix increment operator, `++`.

    //usevectortostorepipes
    std::vector<tf::Pipe<std::function<void(tf::Pipeflow&)>>>vector;
    tf::ScalablePipelinespl1(num_lines,vector.begin(),vector.end());

    //uselisttostorepipes
    std::list<tf::Pipe<std::function<void(tf::Pipeflow&)>>>list;
    tf::ScalablePipelinespl2(num_lines,list.begin(),list.end());

# Learn More about Taskflow Pipeline

Visit the following pages to learn more about pipeline:

- Task-parallel Pipeline

- Data-parallel Pipeline

- Text Processing Pipeline

- Graph Processing Pipeline

- Taskflow Processing Pipeline