docs/GraphProcessingPipeline.html
We study a graph processing pipeline that propagates a sequence of linearly dependent tasks over a dependency graph. In this particular workload, we will learn how to transform task graph parallelism into pipeline parallelism.
Given a directed acyclic graph (DAG), where each node encapsulates a sequence of linearly dependent tasks, namely stage tasks, and each edge represents a dependency between two tasks at the same stages of adjacent nodes. For example, assuming fi(u) represents the ith-stage task of node u, a dependency from u to v requires fi(u) to run before fi(v). The following figures shows an example of three stage tasks in a DAG of three nodes (A, B, and C) and two dependencies (A->B and A->C):
La[Node A]Step 1: f1(A)Step 2: f2(A)Step 3: f3(A)b[Node B]Step 1: f1(B)Step 2: f2(B)Step 3: f3(B)a->bc[Node C]Step 1: f1(C)Step 2: f2(C)Step 3: f3(C)a->c
While we can directly create a taskflow for the DAG (i.e., each task in the taskflow runs f1, f2, and f3 sequentially), we can describe the parallelism as a three-stage pipeline that propagates a topological order of the DAG through three stage tasks. Consider a valid topological order of this DAG, A, B, C, its pipeline parallelism can be illustrated in the following figure:
Rf1_Af1(A)f1_Bf1(B)f1_A->f1_Bf2_Af2(A)f2_Bf2(B)f2_A->f2_Bf3_Af3(A)f3_Bf3(B)f3_A->f3_Bf1_B->f2_Af1_Cf1(C)f1_B->f1_Cf2_B->f3_Af2_Cf2(C)f2_B->f2_Cf3_Cf3(C)f3_B->f3_Cf1_C->f2_Bf2_C->f3_B
At the beginning, f1(A) runs first. When f1(A) completes, it moves on to f2(A) and, meanwhile, f1(B) can start to run together with f2(A), and so on so forth. The straight line represents two parallel tasks that can overlap in time in the pipeline. For example, f3(A), f2(B), and f1(C) can run simultaneously. The following figures shows the task dependency graph of this pipeline workload:
Rf1_Af1(A)f1_Bf1(B)f1_A->f1_Bf2_Af2(A)f1_A->f2_Af1_Cf1(C)f1_B->f1_Cf2_Bf2(B)f1_B->f2_Bf2_Cf2(C)f1_C->f2_Cf2_A->f2_Bf3_Af3(A)f2_A->f3_Af2_B->f2_Cf3_Bf3(B)f2_B->f3_Bf3_Cf3(C)f2_C->f3_Cf3_A->f3_Bf3_B->f3_C
As we can see, tasks in diagonal lines (lower-left to upper-right) can run in parallel. This type of parallelism is also referred to as wavefront parallelism, which sweeps parallel elements in a diagonal direction.
Using the example from the previous section, we create a three-stage pipeline that encapsulates the three stage tasks (f1, f2, f3) in three pipes. By finding a topological order of the graph, we can transform the node dependency into a sequence of linearly dependent data tokens to feed into the pipeline. The overall implementation is shown below:
#include \<taskflow/taskflow.hpp\>#include \<taskflow/algorithm/pipeline.hpp\>// 1st-stage functionvoid f1(const std::string& node) {printf("f1(%s)\n", node.c\_str());}// 2nd-stage functionvoid f2(const std::string& node) {printf("f2(%s)\n", node.c\_str());}// 3rd-stage functionvoid f3(const std::string& node) {printf("f3(%s)\n", node.c\_str());}int main() {tf::Taskflow taskflow("graph processing pipeline");tf::Executor executor;const size\_t num\_lines = 2;// a topological order of the graph// |-\> B// A--|// |-\> Cconst std::vector\<std::string\> nodes = {"A", "B", "C"};// the pipeline consists of three serial pipes// and up to two concurrent scheduling tokenstf::Pipeline pl(num\_lines,// first pipe calls f1tf::Pipe{tf::PipeType::SERIAL, [&](tf::Pipeflow& pf) {if(pf.token() == nodes.size()) {pf.stop();}else {f1(nodes[pf.token()]);}}},// second pipe calls f2tf::Pipe{tf::PipeType::SERIAL, [&](tf::Pipeflow& pf) {f2(nodes[pf.token()]);}},// third pipe calls f3tf::Pipe{tf::PipeType::SERIAL, [&](tf::Pipeflow& pf) {f3(nodes[pf.token()]);}});// 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();return 0;}
The first step is to find a valid topological order of the graph, such that we can transform the graph dependency into a linear sequence. In this example, we simply hard-code it:
const std::vector\<std::string\> nodes = {"A", "B", "C"};
This particular workload does not propagate data directly through the pipeline. In most situations, data is directly stored in a custom graph data structure, and the stage function will just need to know which node to process. For demo's sake, we simply output a message to show which stage function is processing which node:
// 1st-stage functionvoid f1(const std::string& node) {printf("f1(%s)\n", node.c\_str());}// 2nd-stage functionvoid f2(const std::string& node) {printf("f2(%s)\n", node.c\_str());}// 3rd-stage functionvoid f3(const std::string& node) {printf("f3(%s)\n", node.c\_str());}
The pipe structure is straightforward. Each pipe encapsulates the corresponding stage function and passes the node into the function argument. The first pipe will cease the pipeline scheduling when it has processed all nodes. To identify which node is being processed at a running pipe, we use tf::Pipeflow::token to find the index:
// first pipe calls f1tf::Pipe{tf::PipeType::SERIAL, [&](tf::Pipeflow& pf) {if(pf.token() == nodes.size()) {pf.stop();}else {f1(nodes[pf.token()]);}}},// second pipe calls f2tf::Pipe{tf::PipeType::SERIAL, [&](tf::Pipeflow& pf) {f2(nodes[pf.token()]);}},// third pipe calls f3tf::Pipe{tf::PipeType::SERIAL, [&](tf::Pipeflow& pf) {f3(nodes[pf.token()]);}}
To build up the taskflow for the pipeline, we create a module task with the defined pipeline structure and connect it with two tasks that output helper messages before and after the pipeline:
tf::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");init.precede(task);task.precede(stop);
Taskflowcluster_p0x7ffd7418c200Graph Processing Pipelinecluster_p0x7ffd7418c110m1p0x7bc4000142e8starting pipelinep0x7bc4000143d0pipeline [m1]p0x7bc4000142e8->p0x7bc4000143d0p0x7bc4000144b8pipeline stoppedp0x7bc4000143d0->p0x7bc4000144b8p0x7bc400014030condp0x7bc400014118rt-0p0x7bc400014030->p0x7bc4000141180p0x7bc400014200rt-1p0x7bc400014030->p0x7bc4000142001
Finally, we submit the taskflow to the execution and run it once:
executor.run(taskflow).wait();
Three possible outputs are shown below:
# possible output 1ready
f1(A)f2(A)f1(B)f2(B)f3(A)f1(C)f2(C)f3(B)f3(C)stopped# possible output 2f1(A)f2(A)f3(A)f1(B)f2(B)f3(B)f1(C)f2(C)f3(C)stopped# possible output 3ready
f1(A)f2(A)f3(A)f1(B)f2(B)f1(C)f2(C)f3(B)f3(C)stopped
We have applied the graph processing pipeline technique to speed up a circuit analysis problem. Details can be referred to our publication below: