Taskflow 简单使用

Hello World

#include <taskflow/taskflow.hpp>

int main() {
    tf::Executor executor; 
    tf::Taskflow taskflow;
    // 返回一个std::tuple<tf::Task, tf::Task, tf::Task, tf::Task> 
    auto [A, B, C, D] = taskflow.emplace(
        [](){std::cout<<"A"<<std::endl;},
        [](){std::cout<<"B"<<std::endl;},
        [](){std::cout<<"C"<<std::endl;},
        [](){std::cout<<"D"<<std::endl;}
    );

    A.precede(B, C);
    D.succeed(B, C);
    executor.run(taskflow).wait();
    

    return 0;
}

内置一个性能profiler，使用方式：

TF_ENABLE_PROFILER=simple.json ./simple # 可执行文件
cat simple.json 

然后把内容复制到TFProf。

创建一个Subflow Graph

流程如下：

#include <taskflow/taskflow.hpp>

void put(const std::string& str) {
    std::cout<<str<<std::endl;
}

int main() {
    tf::Executor executor; 
    tf::Taskflow taskflow;
    // 构建三个空任务,并命名
    tf::Task A = taskflow.emplace([](){put("A");}).name("A");
    tf::Task C = taskflow.emplace([](){put("C");}).name("C");
    tf::Task D = taskflow.emplace([](){put("D");}).name("D");

    // 构建一个子流,并命名
    tf::Task B = taskflow.emplace([](tf::Subflow& subflow){
        auto [B1, B2, B3] = subflow.emplace(
            [](){put("B1");},
            [](){put("B2");},
            [](){put("B3");} 
        );
        B3.succeed(B1, B2);
    }).name("B");

    A.precede(B, C);
    D.succeed(B, C);
    executor.run(taskflow).wait();
    

    return 0;
}

控制流

循环执行条件，直到返回true，才执行下一步：

#include <taskflow/taskflow.hpp>

void put(const std::string& str) {
    std::cout<<str<<std::endl;
}

int main() {
    tf::Executor executor; 
    tf::Taskflow taskflow;
    // 构建三个空任务,并命名
    tf::Task init = taskflow.emplace([](){put("init");}).name("init");
    tf::Task stop = taskflow.emplace([](){put("stop");}).name("stop");
    tf::Task cond = taskflow.emplace(
        [](){
            int p = std::rand() % 2;
            put(std::to_string(p).c_str());
            return p;
        }
    ).name("cond");

    init.precede(cond);
    cond.precede(cond,stop); // cond 需要返回True后才解除对cond的依赖
    executor.run(taskflow).wait();
    return 0;
}

任务组

一个taskflow的流程中，还可以嵌入另一个taskflow。

#include "taskflow/core/taskflow.hpp"
#include <taskflow/taskflow.hpp>

void put(const std::string& str) {
    std::cout<<str<<std::endl;
}


int main() {
    tf::Executor executor; 
    tf::Taskflow f1, f2;
    
    tf::Task f1a = f1.emplace([](){put("f1a");}).name("f1a");
    tf::Task f1b = f1.emplace([](){put("f1b");}).name("f1b");
    tf::Task f1_task_module = f2.composed_of(f1).name("f1_task_module"); // 表示f1是f2的一个task

    tf::Task f2a = f2.emplace([](){put("f2a");}).name("f2a");
    tf::Task f2b = f2.emplace([](){put("f2b");}).name("f2b");
    tf::Task f2c = f2.emplace([](){put("f2c");}).name("f2c");

    f1_task_module.succeed(f2a, f2b).precede(f2c); // 在f2c之前，在f2a, f2b之后
    
    executor.run(f2).wait();
  
    return 0;
}

异步任务

taskflow 支持开启异步任务，动态探索并行度：

#include "taskflow/core/async_task.hpp"
#include "taskflow/core/task.hpp"
#include "taskflow/core/taskflow.hpp"
#include <future>
#include <taskflow/taskflow.hpp>

// 线程不安全，可能会出现打印异常
void put(const std::string& str) {
    std::cout<<str<<std::endl;
}


int main() {
    tf::Executor executor; 
    tf::Taskflow taskflow;
    
    // 第一种方式，通过async的方式注册，并将解决使用future传递
    std::future<int> future = executor.async([](){
        put("async task returns 1");
        return 1;
    }); 

    // 第二种方式，丢弃返回值的异步
    executor.silent_async([](){
        put("async task does not return");
    });

    // 第三种方式，创建异步，并进行动态依赖
    tf::AsyncTask A = executor.silent_dependent_async([](){put("A");});
    tf::AsyncTask B = executor.silent_dependent_async([](){put("B");}, A); // B依赖A
    tf::AsyncTask C = executor.silent_dependent_async([](){put("C");}, A); // C依赖A
    tf::AsyncTask D = executor.silent_dependent_async([](){put("D");}, B, C); // D依赖B和C

    executor.wait_for_all(); // 等待所有异步任务结束
    return 0;
}

执行一个 Taskflow

executor提供了几种线程安全的方法来运行任务流。

// runs the taskflow once
tf::Future<void> run_once = executor.run(taskflow); 

// wait on this run to finish
run_once.get();

// run the taskflow four times
executor.run_n(taskflow, 4);

// runs the taskflow five times
executor.run_until(taskflow, [counter=5](){ return --counter == 0; });

// block the executor until all submitted taskflows complete
executor.wait_for_all();

可视化图结构

使用dump，生成定义好的结构图，生成的内容复制到下面的网站：
GraphViz Online

#include <taskflow/taskflow.hpp>
int main() {
    tf::Taskflow taskflow;

    tf::Task A = taskflow.emplace([] () {}).name("A");
    tf::Task B = taskflow.emplace([] () {}).name("B");
    tf::Task C = taskflow.emplace([] () {}).name("C");
    tf::Task D = taskflow.emplace([] () {}).name("D");
    tf::Task E = taskflow.emplace([] () {}).name("E");
    A.precede(B, C, E);
    C.precede(D);
    B.precede(D, E);

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

    return 0;
}

digraph Taskflow {
subgraph cluster_p0xffffd542ef38 {
label="Taskflow: p0xffffd542eee0";
p0xffffa6914830[label="A" ];
p0xffffa6914830 -> p0xffffa6914928;
p0xffffa6914830 -> p0xffffa6914a20;
p0xffffa6914830 -> p0xffffa6914c10;
p0xffffa6914928[label="B" ];
p0xffffa6914928 -> p0xffffa6914b18;
p0xffffa6914928 -> p0xffffa6914c10;
p0xffffa6914a20[label="C" ];
p0xffffa6914a20 -> p0xffffa6914b18;
p0xffffa6914b18[label="D" ];
p0xffffa6914c10[label="E" ];
}
}