Taskflow

Why Taskflow?

Start Your First Taskflow Program

#include<taskflow/taskflow.hpp>// Taskflow is header-onlyintmain(){    tf::Executor executor;  tf::Taskflow taskflow;auto [A, B, C, D] = taskflow.emplace(// create four tasks    [] () { std::cout <<"TaskA\n"; },    [] () { std::cout <<"TaskB\n"; },    [] () { std::cout <<"TaskC\n"; },    [] () { std::cout <<"TaskD\n"; }   );                                                                          A.precede(B, C);// A runs before B and C  D.succeed(B, C);// D runs after  B and C                                        executor.run(taskflow).wait();return0;}

~$ git clone https://github.com/taskflow/taskflow.git# clone it only once~$ g++ -std=c++20 examples/simple.cpp -I. -O2 -pthread -o simple~$ ./simpleTaskATaskC TaskB TaskD

Visualize Your First Taskflow Program

# run the program with the environment variable TF_ENABLE_PROFILER enabled~$ TF_ENABLE_PROFILER=simple.json ./simple~$ cat simple.json[{"executor":"0","data":[{"worker":0,"level":0,"data":[{"span":[172,186],"name":"0_0","type":"static"},{"span":[187,189],"name":"0_1","type":"static"}]},{"worker":2,"level":0,"data":[{"span":[93,164],"name":"2_0","type":"static"},{"span":[170,179],"name":"2_1","type":"static"}]}]}]# paste the profiling json data to https://taskflow.github.io/tfprof/

// dump the taskflow graph to a DOT format through std::couttaskflow.dump(std::cout);

Express Task Graph Parallelism

Create a Subflow Graph

tf::Task A = taskflow.emplace([](){}).name("A");  tf::Task C = taskflow.emplace([](){}).name("C");  tf::Task D = taskflow.emplace([](){}).name("D");  tf::Task B = taskflow.emplace([] (tf::Subflow& subflow) {   tf::Task B1 = subflow.emplace([](){}).name("B1");    tf::Task B2 = subflow.emplace([](){}).name("B2");    tf::Task B3 = subflow.emplace([](){}).name("B3");    B3.succeed(B1, B2);// B3 runs after B1 and B2}).name("B");A.precede(B, C);// A runs before B and CD.succeed(B, C);// D runs after  B and C

Integrate Control Flow to a Task Graph

tf::Task init = taskflow.emplace([](){}).name("init");tf::Task stop = taskflow.emplace([](){}).name("stop");// creates a condition task that returns a random binarytf::Task cond = taskflow.emplace(  [](){returnstd::rand() %2; }).name("cond");init.precede(cond);// creates a feedback loop {0: cond, 1: stop}cond.precede(cond, stop);

Offload a Task to a GPU

__global__voidsaxpy(size_t N,float alpha,float* dx,float* dy) {int i = blockIdx.x*blockDim.x + threadIdx.x;if (i < n) {    y[i] = a*x[i] + y[i];  }}// create a CUDA Graph tasktf::Task cudaflow = taskflow.emplace([&]() {  tf::cudaGraph cg;  tf::cudaTask h2d_x = cg.copy(dx, hx.data(), N);  tf::cudaTask h2d_y = cg.copy(dy, hy.data(), N);  tf::cudaTask d2h_x = cg.copy(hx.data(), dx, N);  tf::cudaTask d2h_y = cg.copy(hy.data(), dy, N);  tf::cudaTask saxpy = cg.kernel((N+255)/256,256,0, saxpy, N,2.0f, dx, dy);  saxpy.succeed(h2d_x, h2d_y)       .precede(d2h_x, d2h_y);// instantiate an executable CUDA graph and run it through a stream  tf::cudaGraphExecexec(cg);  tf::cudaStream stream;  stream.run(exec).synchronize();}).name("CUDA Graph Task");

Compose Task Graphs

tf::Taskflow f1, f2;// create taskflow f1 of two taskstf::Task f1A = f1.emplace([]() { std::cout <<"Task f1A\n"; })                 .name("f1A");tf::Task f1B = f1.emplace([]() { std::cout <<"Task f1B\n"; })                 .name("f1B");// create taskflow f2 with one module task composed of f1tf::Task f2A = f2.emplace([]() { std::cout <<"Task f2A\n"; })                 .name("f2A");tf::Task f2B = f2.emplace([]() { std::cout <<"Task f2B\n"; })                 .name("f2B");tf::Task f2C = f2.emplace([]() { std::cout <<"Task f2C\n"; })                 .name("f2C");tf::Task f1_module_task = f2.composed_of(f1)                            .name("module");f1_module_task.succeed(f2A, f2B)              .precede(f2C);

Launch Asynchronous Tasks

tf::Executor executor;// create asynchronous tasks directly from an executorstd::future<int> future = executor.async([](){   std::cout <<"async task returns 1\n";return1;}); executor.silent_async([](){ std::cout <<"async task does not return\n"; });// create asynchronous tasks with dynamic dependenciestf::AsyncTask A = executor.silent_dependent_async([](){printf("A\n"); });tf::AsyncTask B = executor.silent_dependent_async([](){printf("B\n"); }, A);tf::AsyncTask C = executor.silent_dependent_async([](){printf("C\n"); }, A);tf::AsyncTask D = executor.silent_dependent_async([](){printf("D\n"); }, B, C);executor.wait_for_all();

Execute a Taskflow

// runs the taskflow oncetf::Future<void> run_once = executor.run(taskflow);// wait on this run to finishrun_once.get();// run the taskflow four timesexecutor.run_n(taskflow,4);// runs the taskflow five timesexecutor.run_until(taskflow, [counter=5](){return --counter ==0; });// block the executor until all submitted taskflows completeexecutor.wait_for_all();

Leverage Standard Parallel Algorithms

tf::Task task1 = taskflow.for_each(// assign each element to 100 in parallel  first, last, [] (auto& i) { i =100; }    );tf::Task task2 = taskflow.reduce(// reduce a range of items in parallel  first, last, init, [] (auto a,auto b) {return a + b; });tf::Task task3 = taskflow.sort(// sort a range of items in parallel  first, last, [] (auto a,auto b) {return a < b; });

// create a pipeline to propagate five tokens through three serial stagestf::Pipelinepl(num_parallel_lines,  tf::Pipe{tf::PipeType::SERIAL, [](tf::Pipeflow& pf) {if(pf.token() ==5) {      pf.stop();    }  }},  tf::Pipe{tf::PipeType::SERIAL, [](tf::Pipeflow& pf) {printf("stage 2: input buffer[%zu] = %d\n", pf.line(), buffer[pf.line()]);  }},  tf::Pipe{tf::PipeType::SERIAL, [](tf::Pipeflow& pf) {printf("stage 3: input buffer[%zu] = %d\n", pf.line(), buffer[pf.line()]);  }});taskflow.composed_of(pl)executor.run(taskflow).wait();

Supported Compilers

Learn More about Taskflow

License