现代 C++ 任务并行编程系统：Taskflow 详解

七、理论 vs 实际

理想线性加速： T(N) = T(1) / N

现实情况： T(N) = T(1) / (N * η) 其中： 0 < η < 1 η 是效率因子。 当通信开销大时： η → 0.5 或更低

八、一个简单 C++ 并行示例（OpenMP）

#include <vector>
#include <omp.h>
#include <iostream>

// 模拟一个可并行计算任务
int main() {
    const int N = 100000000;
    std::vector<double> data(N, 1.0);
    double sum = 0.0;

    // 并行循环
    #pragma omp parallel for reduction(+:sum)
    for (int i = 0; i < N; ++i) {
        sum += data[i];
    }

    std::cout << "Sum = " << sum << std::endl;
}

https://godbolt.org/z/G53TqTqPP

说明

#pragma omp parallel for

表示：

• 把 for 循环分配给多个线程执行

如果使用 8 核： 理论时间： T ≈ T_single / 8

九、并行计算的核心价值

并行计算的目标是：

• 更快训练 AI
• 更快物理仿真
• 更快金融建模
• 更快基因计算
• 更快科学计算

很多问题的时间复杂度是： O(N^2) 或 O(N^3) 如果不并行，计算时间会爆炸。

十、总结

并行计算的重要性在于：

1. 单核性能提升停滞
2. 数据规模指数增长
3. GPU 提供数量级并行能力
4. 机器学习天然可并行
5. 10–100x 加速是现实而不是理论

一、Why Task-parallel Programming?

Modern parallel workloads are very complex and irregular 现代并行负载非常复杂且不规则

核心问题

现实应用已经不再是：

• 单一大规模矩阵乘法
• 规则数据并行

而是：

• 图计算
• 电路仿真
• AI + 物理模拟混合
• 自适应网格
• 异构调度

这些计算具有：

• 不规则依赖
• 不同计算粒度
• 不同硬件需求
• 动态生成任务

这就需要：

任务并行模型

二、异构架构背景

现代系统包括：

• CPU
• GPU
• TPU
• FPGA
• 专用加速器

称为：

Heterogeneous Architecture（异构架构）

例如： 电路仿真算法可能包含：

• CPU 负责控制逻辑
• GPU 负责大规模矩阵计算
• FPGA 负责特定加速模块

三、什么是 Task Parallelism？

数据并行

典型形式： for i = 1...N: y_i = f(x_i) 每个元素独立。 适用于：

• 矩阵运算
• 向量运算

任务并行

把算法分解成任务： T = {t_1, t_2, ..., t_n} 并定义依赖关系： t_i → t_j 表示： t_j 必须在 t_i 完成后执行 形成：

有向无环图（DAG）

四、任务图模型

设任务图为： G = (V, E) 其中：

• V 是任务集合
• E 是依赖边

关键指标

1⃣ 总工作量（Work）

W = ∑_{i=1}^{n} w_i 每个任务的计算量之和。

2⃣ 关键路径长度（Span）

S = max_path ∑ w_i 最长依赖路径。

3⃣ 理论并行上限

根据 Brent 定理： T_p ≥ max(W/P, S) 其中： P = 处理器数量

说明：

性能受限于关键路径

五、为什么任务并行更可扩展？

未来架构趋势：

• 核心数量爆炸
• 硬件类型多样
• 计算模式混合

数据并行适合：

• 规则问题

任务并行适合：

• 不规则问题
• 动态问题
• 异构问题

因此：

Task parallelism is most scalable for future heterogeneous architectures

六、任务分解的思想

Capture your intention in decomposing an algorithm into a top-down task graph 意思是： 从算法逻辑结构出发，构建任务图。

例如：

1. 读取数据
2. 预处理
3. GPU 计算
4. 后处理
5. 输出

形成： t_1 → t_2 → t_3 → t_4 → t_5 但中间可能有分叉： t_2 → {t_3, t_4}

七、实际系统示例

下面列出常见任务并行框架（简要说明）

OpenMP

支持任务构造（task directive）

Kokkos

面向性能可移植性

TBB

Intel Threading Building Blocks

StarPU

异构任务调度

DASK

Python 分布式任务图

PaRSEC

分布式任务图调度

RAY

分布式 AI 任务系统

八、OpenMP 任务并行示例（带详细注释）

#include <iostream>
#include <omp.h>

// 模拟三个有依赖的任务
int main() {
    #pragma omp parallel {
        #pragma omp single {
            // 任务 A
            #pragma omp task {
                std::cout << "Task A running\n";
            }
            // 任务 B
            #pragma omp task {
                std::cout << "Task B running\n";
            }
            // 等待 A 和 B 完成
            #pragma omp taskwait
            // 任务 C 依赖 A 和 B
            #pragma omp task {
                std::cout << "Task C running after A and B\n";
            }
        }
    }
    return 0;
}

解释

#pragma omp task

表示创建一个独立任务。

#pragma omp taskwait

表示： 必须等待之前任务完成。 这构成一个简单 DAG：

 A
/ \
B C
\ /
 C

数学表达： A → C B → C

九、GPU + CPU 混合任务示意

假设：

• 任务 1：CPU 预处理
• 任务 2：GPU 计算
• 任务 3：CPU 后处理

时间模型： T = T_cpu1 + T_gpu + T_cpu2 如果 GPU 并行度高： T_gpu ≪ T_cpu1 总时间显著下降。

十、任务并行 vs 数据并行对比

十一、为什么说'最可扩展'？

未来趋势： cores → 10^3 ~ 10^6 而： 规则问题越来越少 复杂系统越来越多 只有任务图可以表达：

• 动态依赖
• 不同粒度
• 不同硬件映射

十二、总结

任务并行的核心优势：

1. 能表达复杂依赖
2. 能适配异构架构
3. 可扩展到未来超大规模系统
4. 能处理不规则问题
5. 以 DAG 形式表达算法意图

关键数学模型： G = (V, E) T_p ≥ max(W/P, S)

Taskflow 静态任务图并行（Static Task Graph Parallelism）

一、什么是 Taskflow？

Taskflow 是一个：

• 现代 C++ 任务并行库
• header-only（只需包含头文件）
• 基于 DAG（有向无环图）的任务调度框架
• 支持静态和动态任务图

核心思想：

用任务图表达算法结构，而不是写线程代码。

二、什么是 Static Task Graph？

静态任务图的意思是：

所有任务和依赖关系在运行前就构建完成。

数学模型： 定义任务图： G = (V, E) 其中：

• V = 任务集合
• E = 依赖边集合

执行时间满足： T_p ≥ max(W/P, S) 其中：

• W = 总工作量
• P = 线程数
• S = 关键路径长度（critical path）

#include <iostream>
#include <taskflow/taskflow.hpp>

// 引入 Taskflow 头文件（header-only 库）
int main() {
    // 执行器：负责调度线程池执行任务
    tf::Executor executor;
    // 任务图对象：用来构建 DAG
    tf::Taskflow taskflow;

    // 创建四个任务（lambda 表达式）
    // emplace 会返回 task 句柄
    auto [A, B, C, D] = taskflow.emplace(
        // 任务 A
        [](){ std::cout << "TaskA\n"; },
        // 任务 B
        [](){ std::cout << "TaskB\n"; },
        // 任务 C
        [](){ std::cout << "TaskC\n"; },
        // 任务 D
        [](){ std::cout << "TaskD\n"; });

    // 定义依赖关系：
    // A 必须在 B 和 C 之前执行
    A.precede(B, C);
    // D 必须在 B 和 C 之后执行
    D.succeed(B, C);

    // 提交任务图到执行器，并等待执行完成
    executor.run(taskflow).wait();
    return 0;
}

五、任务图结构解析

根据依赖：

A.precede(B, C); D.succeed(B, C);

等价关系： A → B A → C B → D C → D

DAG 图结构

 A
/ \
B C
\ /
 D

六、执行顺序分析

合法执行顺序：

1. A
2. B 和 C（可并行）
3. D

并行度分析： 若每个任务耗时为 t： 总工作量： W = 4t 关键路径： S = 3t （A → B → D 或 A → C → D）

理论最优时间： T_p ≥ max(4t/P, 3t) 如果 P ≥ 2： T_p = 3t

七、为什么 Taskflow 是'静态任务图'？

因为：

• 所有任务在 run() 之前已经定义好
• DAG 不会在执行过程中改变

对比： 动态任务图（Dynamic Task Graph）：

• 任务内部再生成新任务

八、Taskflow 设计优势

1⃣ 表达能力强

可以表达任意 DAG。

2⃣ 自动调度

内部使用：

Work-stealing 调度算法

线程空闲时：

• 会从其他线程窃取任务

九、和 OpenMP 对比

OpenMP：

#pragma omp task

缺点：

• 不够灵活
• 难表达复杂 DAG

Taskflow：

• 显式 DAG
• 可视化
• 更适合复杂依赖

十、线程池模型

Executor 内部维护： P 个 worker threads

执行规则：

• 当一个任务的入度为 0（无依赖）即可执行
• 依赖完成后触发后续任务

数学表达： 若任务 v 的入度： deg^-(v) = 0 则可调度。

十一、关键路径决定性能

对于 DAG： 总时间近似： T ≈ S + (W - S)/P 当： P → ∞ 有： T → S

说明：

关键路径是性能上限

十二、扩展示例（加入耗时）

auto [A,B,C,D] = taskflow.emplace(
    [](){ std::this_thread::sleep_for(std::chrono::milliseconds(100)); std::cout<<"A\n";},
    [](){ std::this_thread::sleep_for(std::chrono::milliseconds(200)); std::cout<<"B\n";},
    [](){ std::this_thread::sleep_for(std::chrono::milliseconds(200)); std::cout<<"C\n";},
    [](){ std::cout<<"D\n";});

你会看到：

• B 和 C 同时运行
• 总时间接近： 100 + 200 = 300 ms 而不是 500 ms。

十三、总结

Taskflow 核心思想：

1. 用 DAG 表达算法
2. 静态任务图提前构建
3. 自动并行调度
4. 关键路径决定性能
5. 非常适合复杂依赖结构

https://godbolt.org/z/1cGrajrz9

Taskflow 动态任务图并行（Dynamic Task Graph Parallelism）

一、什么是动态任务图（Dynamic Task Graph）？

和静态任务图不同：

• 静态任务图：先构建完整 DAG，再执行
• 动态任务图：任务在运行过程中动态创建

也就是说：

DAG 在运行时逐步生成，而不是提前构建完成。

二、动态任务图的数学模型

设任务集合为： V = {t_1, t_2, ...} 依赖关系为： E = {(t_i → t_j)}

在动态模型中： V(t) 和 E(t) 会随着时间变化。 即： G(t) = (V(t), E(t))

三、代码示例（修正版 + 详细注释）

#include <iostream>
#include <taskflow/taskflow.hpp>

// Taskflow 是 header-only
int main() {
    // 创建执行器（内部是线程池）
    tf::Executor executor;
    // 创建任务 A（无依赖）
    auto A = executor.silent_dependent_async([](){ std::cout << "TaskA\n";});
    // 创建任务 B（依赖 A）
    auto B = executor.silent_dependent_async([](){ std::cout << "TaskB\n";}, A);
    // 创建任务 C（依赖 A）
    auto C = executor.silent_dependent_async([](){ std::cout << "TaskC\n";}, A);
    // 创建任务 D（依赖 B 和 C）
    auto D = executor.silent_dependent_async([](){ std::cout << "TaskD\n";}, B, C);
    // 等待所有任务完成
    executor.wait_for_all();
    return 0;
}

四、DAG 结构分析

依赖关系： A → B A → C B → D C → D

图结构：

 A
/ \
B C
\ /
 D

五、执行流程解析

1. A 无依赖 → 立即执行
2. A 完成后：
   • B 和 C 入度变为 0 → 可并行执行
3. B 和 C 完成后：
   • D 入度变为 0 → 执行

六、与静态任务图的区别

七、关键性能分析

设：

• 每个任务耗时 t

总工作量： W = 4t 关键路径： S = 3t

理论执行时间： T_p ≥ max(W/P, S)

若线程数 P ≥ 2： T_p = 3t

八、silent_dependent_async 的含义

函数原型逻辑：

silent_dependent_async(task, dependencies...) 

意思：

• 创建任务
• 指定它依赖哪些任务
• 不返回 future（silent）
• 自动加入 executor

内部机制： 每个任务维护： deg^-(v) 当： deg^-(v) = 0 任务进入 ready queue。

九、为什么叫'动态'？

因为你可以：

• 在任务内部再创建任务
• 根据运行时数据决定依赖

例如：

auto A = executor.silent_dependent_async([&executor](){ std::cout<<"TaskA\n";
    // 在 A 运行期间创建新任务
    executor.silent_dependent_async([](){ std::cout<<"Dynamic task\n";});
});

这时： V(t) 在运行时增加

十、调度模型（Work-Stealing）

Executor 内部：

• 维护多个 worker 线程
• 每个线程有本地任务队列
• 空闲线程会'窃取'其他线程任务

数学上： 若线程数为 P： 期望时间近似： T_p ≈ W/P + O(S)

Work-stealing 具有：

• 理论近似最优调度
• 良好负载均衡

十一、动态模型适用场景

适合：

• 递归算法
• 图搜索
• 不规则计算
• 自适应网格
• AI 推理流水线
• 动态负载均衡

例如： 分治算法： T(n) = 2T(n/2) + O(n) 可以在运行中不断创建子任务。

十二、与 std::async 对比

std::async：

• 不是 DAG
• 不支持依赖图
• 无调度优化

Taskflow：

• 显式依赖
• 高效线程池
• 可扩展到复杂 DAG

十三、总结

动态任务图的本质： G(t) = (V(t), E(t)) 随着运行动态扩展。

优势：

1. 支持运行时生成任务
2. 适合不规则问题
3. 更灵活
4. 与异构计算更兼容
5. 更接近未来并行编程模型

https://godbolt.org/z/vhEExTh4Y

Control Taskflow Graph (CTFG) 编程模型

一、什么是 CTFG（Control Taskflow Graph）？

CTFG = 控制流任务图 它的核心思想是：

不仅表达'计算任务依赖'，还表达'控制流逻辑'（if / while / loop / branch） 普通 DAG 只能表达： A → B 但 CTFG 可以表达：

• 条件分支（if-else）
• 循环（while）
• 迭代优化
• 收敛判断

即：

把程序控制结构也变成任务图的一部分

二、第一个例子：条件分支（if-else）

原始结构类似：

auto [init, cond, yes, no] = taskflow.emplace(...);

我们写成一个完整、正确、带注释的版本：

完整示例：CTFG 条件任务

#include <iostream>
#include <taskflow/taskflow.hpp>

int main() {
    tf::Executor executor;
    tf::Taskflow taskflow;

    // 创建 4 个任务
    auto [init, cond, yes, no] = taskflow.emplace(
        // 初始化任务
        [](){ std::cout << "initialize\n"; },
        // 条件任务（Condition Task）
        // 返回值决定走哪条分支
        []()->int{ std::cout << "checking condition\n"; return 0; /* 返回 0 走 yes，返回 1 走 no */ },
        // yes 分支
        [](){ std::cout << "yes branch\n"; },
        // no 分支
        [](){ std::cout << "no branch\n"; });

    // 依赖关系定义
    // cond 必须在 init 后执行
    cond.succeed(init);
    // cond 决定执行 yes 或 no
    cond.precede(yes, no);

    executor.run(taskflow).wait();
    return 0;
}

https://godbolt.org/z/1oTW6jEPq

三、条件任务的数学模型

定义： t_cond : N → {0,1,...,k} 若： t_cond() = i 则调度： t_i 这是一个动态边选择机制。

四、CTFG 循环优化示例

第二段代码表示：

迭代优化直到收敛

等价伪代码：

initialize();
while(!converged()){
    optimize();
}
print("done");

完整 CTFG 实现版本

#include <iostream>
#include <taskflow/taskflow.hpp>

bool converged(){
    static int count = 0;
    return ++count > 3; // 模拟 3 次后收敛
}

int main() {
    tf::Executor executor;
    tf::Taskflow taskflow;

    auto [init, opt, cond, stop] = taskflow.emplace(
        // 初始化数据结构
        [](){ std::cout << "initialize data\n"; },
        // 优化步骤
        [](){ std::cout << "optimize step\n"; },
        // 条件任务（返回分支编号）
        []()->int{return converged()?1:0; },
        // 收敛后停止
        [](){ std::cout << "done!\n"; });

    // 执行顺序：opt.succeed(init).precede(cond)
    // cond:
    // 返回 0 → 回到 opt
    // 返回 1 → 进入 stop
    cond.precede(opt, stop);

    executor.run(taskflow).wait();
    return 0;
}

https://godbolt.org/z/5c4KTxTrf

五、循环的任务图结构

图结构：

 init
   |
   v
 opt
   |
   v
 cond
 / \ 
opt stop

六、数学表达（循环模型）

设：

• 每次优化时间为 t
• 收敛需要 k 次

总时间： T = k * t

任务图表示： opt_i → cond_i → opt_{i+1} 直到： cond_i = 1

七、CTFG 的核心价值

普通任务图：

• 只能表达静态依赖

CTFG：

• 表达控制流
• 表达循环
• 表达分支
• 表达递归

本质： Program → Control DAG

八、为什么说'CTFG enables end-to-end parallelism expression'？

因为： 整个算法流程可以完全用任务图表达：

1. 初始化
2. 计算
3. 分支
4. 循环
5. 收敛
6. 结束

没有隐藏的串行控制逻辑。 这意味着： 整个程序 = G(V,E) 调度器可以：

• 分析关键路径
• 优化执行顺序
• 自动并行

九、性能模型

设：

• 总工作量 W
• 关键路径 S

理论执行时间： T_p ≥ max(W/P, S)

CTFG 优势： 减少人为串行代码，缩短 S。

十、对比传统 while 循环

传统：

while(!converged()){
    optimize();
}

问题：

• 隐式串行
• 不可分析 DAG

CTFG：

• 显式控制结构
• 可并行分析
• 可跨设备调度

十一、CTFG vs 数据并行

十二、总结

CTFG 是：

把程序的控制流结构变成可调度的任务图

它可以表达：

• if
• while
• for
• 递归
• 收敛循环

数学本质： Program = G(V,E) 执行时间受： critical path 控制。

Taskflow 中的 Work-stealing Runtime（工作窃取运行时）

一、什么是 Work-stealing？

Work-stealing（工作窃取）是一种并行调度算法：

空闲线程主动从其他线程'窃取'任务执行。 其目标是：

• 动态负载均衡
• 最大化 CPU 利用率
• 减少调度开销
• 保证高吞吐与低延迟

二、整体运行逻辑解析（结合流程图）

我们按照流程图逐步解释。

1⃣ Worker 主循环

Start → CheckEmpty

每个 worker 线程执行：

while(true){
    if(queue.empty()){ wait_or_steal(); }
    else{ execute_task(); }
}

2⃣ 队列是否为空？

情况 A：空

CheckEmpty -- Y --> Wait

线程会：

• 尝试从其他线程窃取任务
• 若无任务 → 等待（sleep）

情况 B：非空

CheckEmpty -- N --> Dequeue

从本地双端队列取任务：

t = queue.pop_front();

三、执行任务

判断：

IsCondition{Condition task?}

在 CTFG 中，任务可能是：

• 普通任务
• 条件任务（控制流节点）

四、条件任务执行路径

流程：

r = invoke(t)
enqueue r-th child

代码逻辑类似：

int r = t(); // 执行条件函数
enqueue(t.child(r));

数学表达： 设： t_cond : N → {0,1,...,k} 若： t_cond() = r 则只激活： child_r 这实现了：

• if
• switch
• loop 分支控制

五、普通任务执行路径

流程：

invoke(t)
↓
DecDeps
↓
EnqueueSuccessors

对应逻辑：

invoke(t);
for(auto s : successors(t)){
    s.remaining_deps--;
    if(s.remaining_deps == 0) enqueue(s);
}

六、依赖计数模型（Strong Dependencies）

每个任务维护： deg^-(v) 即：

剩余强依赖数量

当： deg^-(v) = 0 任务进入 ready queue。

七、完整调度数学模型

设：

• 任务图为 G = (V, E)
• 总工作量 W
• 关键路径 S
• 线程数 P

根据理论： T_P ≤ W/P + O(S)

Work-stealing 被证明：

期望时间接近理论最优。

八、为什么 Work-stealing 高效？

1⃣ 局部性好

• 线程优先执行本地任务
• 减少缓存失效

2⃣ 负载均衡

• 空闲线程自动窃取
• 避免手动负载分配

3⃣ 低调度开销

使用：

• 无锁双端队列
• 原子操作减少锁竞争

九、与集中调度对比

集中式调度：

• 所有线程访问同一个全局队列
• 锁竞争严重

Work-stealing：

• 每线程一个队列
• 只在窃取时竞争

复杂度对比： 集中式： O(P) 窃取式： O(1) 平均

十、动态图 + Work-stealing 的优势

CTFG 中：

• 条件任务动态生成任务
• 循环任务动态回边

Work-stealing 具有：

强平衡能力（Strong Balance）

即： Active workers ≈ Available parallelism 保持：

• 低延迟
• 高吞吐
• 能源效率高

十一、关键性能指标解释

Low Latency（低延迟）

任务一旦依赖满足： deg^-(v) = 0 立即入队。

High Throughput（高吞吐）

单位时间执行任务数量： Throughput = Tasks completed / Time

Work-stealing 减少 idle time。

Energy Efficient（节能）

• 减少 busy-wait
• 减少锁竞争
• 减少 cache miss

十二、代码伪实现（简化）

void worker_loop(){
    while(true){
        Task* t = pop_local();
        if(!t){
            t = steal_from_others();
            if(!t){ sleep(); continue; }
        }
        if(t->is_condition()){
            int r = t->invoke();
            enqueue(t->child(r));
        } else {
            t->invoke();
            for(auto s : t->successors){
                if(--s->remaining_deps == 0) enqueue(s);
            }
        }
    }
}

十三、为什么说'强平衡'？

任务生成速度： λ_g 任务执行速度： λ_e

理想情况： λ_g ≈ λ_e

Work-stealing 能自动调节：

• 若生成过快 → 更多线程参与
• 若生成过慢 → 线程空闲休眠

十四、关键路径影响

即使 Work-stealing 非常优秀： 总时间仍受限于： S = critical path 当： P → ∞ 有： T → S

十五、总结

Taskflow 的 Work-stealing Runtime：

1. 每线程独立队列
2. 条件任务支持控制流
3. 强依赖计数机制
4. 动态任务图支持
5. 理论接近最优调度

数学核心： T_P ≤ W/P + O(S)

工程效果：

• 低延迟
• 高吞吐
• 高能效
• 支持动态图
• 适合未来异构架构

完整示例代码

#include <iostream>
#include <taskflow/taskflow.hpp>

// Taskflow 是 header-only 库
int main() {
    // 创建执行器（内部维护线程池）
    tf::Executor executor;
    // 创建主任务图
    tf::Taskflow taskflow;

    // 创建主图中的任务 A
    tf::Task A = taskflow.emplace([](){ std::cout << "Task A\n"; }).name("A");
    // 创建主图中的任务 C
    tf::Task C = taskflow.emplace([](){ std::cout << "Task C\n"; }).name("C");
    // 创建主图中的任务 D
    tf::Task D = taskflow.emplace([](){ std::cout << "Task D\n"; }).name("D");

    // 创建任务 B（包含一个子任务图 Subflow）
    tf::Task B = taskflow.emplace([](tf::Subflow& subflow){
        std::cout << "Task B (start)\n";
        // 在 B 内部创建子任务 B1
        tf::Task B1 = subflow.emplace([](){ std::cout << "Task B1\n"; }).name("B1");
        // 子任务 B2
        tf::Task B2 = subflow.emplace([](){ std::cout << "Task B2\n"; }).name("B2");
        // 子任务 B3
        tf::Task B3 = subflow.emplace([](){ std::cout << "Task B3\n"; }).name("B3");

        // 依赖关系：
        // B3 必须在 B1 和 B2 执行完成后执行
        B3.succeed(B1, B2);
        std::cout << "Task B (end)\n";
    }).name("B");

    // 主任务图依赖关系：
    // A 必须在 B 和 C 之前执行
    A.precede(B, C);
    // D 必须在 B 和 C 之后执行
    D.succeed(B, C);

    // 执行任务图并等待完成
    executor.run(taskflow).wait();
    return 0;
}

https://godbolt.org/z/3c3b7YzsP

任务结构解析

主任务图结构

依赖关系： A → B A → C B → D C → D

图结构：

 A
/ \
B C
\ /
 D

子任务图（B 内部）

 B1 B2
 \ /
  B3

数学表示： B1 → B3 B2 → B3

执行顺序分析

可能的执行顺序：

1. A
2. B 和 C 并行
3. B 内部：
   • B1 和 B2 并行
   • B3 在 B1 和 B2 之后
4. D 在 B 和 C 完成后执行

并行度分析

设：

• 每个任务耗时 t

总工作量： W = 7t 关键路径（最长路径）： A → B → B3 → D 长度： S = 4t

理论并行时间： T_P ≥ max(7t/P, 4t)

当线程数足够多： T ≈ 4t

Subflow 的重要特性

• 子任务图在 B 运行时动态创建
• 属于动态任务图（Dynamic Task Graph）
• 继承父任务的调度上下文
• 可嵌套

Subflow 执行语义

默认行为：

• B 会等待其子任务全部完成后才结束

如果调用：

subflow.detach();

则：

• 子任务会独立执行
• B 不会等待

总结

这个例子展示了：

1. 静态主任务图
2. 动态子任务图（Subflow）
3. 条件依赖
4. 嵌套并行
5. Work-stealing 调度

数学模型仍为： T_P ≤ W/P + O(S) 其中：

• W 是总工作量
• S 是关键路径

B 子任务图

任务 A

任务 B

任务 C

任务 D

任务 B1

任务 B3

任务 B2

graph TD %% 主任务图
A["任务 A"] --> B["任务 B"]
A --> C["任务 C"]
B --> D["任务 D"]
C --> D
%% 子任务图 B 内部
subgraph B 子任务图
B1["任务 B1"] --> B3["任务 B3"]
B2["任务 B2"] --> B3
B3 --> B
end
%% 样式美化
style A fill:#ffcc00,stroke:#333,stroke-width:2px
style B fill:#66ccff,stroke:#333,stroke-width:2px
style C fill:#ff99cc,stroke:#333,stroke-width:2px
style D fill:#99ff99,stroke:#333,stroke-width:2px
style B1 fill:#ccff99,stroke:#333
style B2 fill:#ccff99,stroke:#333
style B3 fill:#ccff99,stroke:#333

#include <taskflow/taskflow.hpp>
// Taskflow 是头文件库
#include <iostream>
#include <cstdlib> // std::rand()
#include <ctime> // std::time()

int main() {
    // 初始化随机数种子
    std::srand(static_cast<unsigned>(std::time(nullptr)));

    // 创建 Taskflow 执行器和任务流
    tf::Executor executor;
    tf::Taskflow taskflow;

    // -------------------------
    // Step 1: 初始化任务
    // -------------------------
    tf::Task init = taskflow.emplace([](){ std::cout << "初始化任务 init\n"; }).name("init");

    // -------------------------
    // Step 2: 停止任务
    // -------------------------
    tf::Task stop = taskflow.emplace([](){ std::cout << "结束任务 stop\n"; }).name("stop");

    // -------------------------
    // Step 3: 条件任务 (Condition Task)
    // 返回随机 0 或 1，用于控制流程
    // 0 -> 反馈回 cond（循环）
    // 1 -> 结束 stop
    // -------------------------
    tf::Task cond = taskflow.emplace([](){
        int r = std::rand()%2;
        std::cout << "条件任务 cond 返回：" << r << "\n";
        return r;
    }).name("cond");

    // -------------------------
    // Step 4: 构建任务依赖关系
    // -------------------------
    init.precede(cond); // init 任务在 cond 前执行
    // cond 返回 0 或 1 决定下一步：
    // 0 -> 再次执行 cond（形成循环）
    // 1 -> 执行 stop
    cond.precede(cond, stop);

    // -------------------------
    // Step 5: 运行任务流
    // -------------------------
    executor.run(taskflow).wait();
    return 0;
}

https://godbolt.org/z/Yzeo4x79Y

说明

1. init 任务
   • 初始化数据结构或准备工作
   • 必须在 cond 之前执行
2. cond 任务
   • 条件任务，返回 0 或 1
   • Taskflow 会根据返回值选择下一步任务
     • 返回 0 → 再次执行 cond（形成循环）
     • 返回 1 → 执行 stop，结束任务流
3. stop 任务
   • 程序结束前的清理任务
   • 只有当 cond 返回 1 时执行
4. 依赖关系
   • init.precede(cond)：init 在 cond 前执行
   • cond.precede(cond, stop)：cond 根据返回值决定下一步

返回 0

返回 1

初始化任务 init

条件任务 cond?

再次执行 cond

结束任务 stop

#include <taskflow/taskflow.hpp>
// Taskflow header-only
#include <iostream>

int main() {
    // ---------------------------
    // 创建子任务流 f1
    // ---------------------------
    tf::Taskflow f1;
    tf::Task f1A = f1.emplace([](){ std::cout << "Task f1A\n"; }).name("f1A");
    tf::Task f1B = f1.emplace([](){ std::cout << "Task f1B\n"; }).name("f1B");
    // f1A 先于 f1B
    f1A.precede(f1B);

    // ---------------------------
    // 创建主任务流 f2
    // ---------------------------
    tf::Taskflow f2;
    // 普通任务
    tf::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");

    // 模块任务，将子任务流 f1 嵌入 f2
    tf::Task f1_module_task = f2.composed_of(f1).name("module");

    // 模块任务与其他任务的依赖关系
    f1_module_task.succeed(f2A, f2B); // module 之后执行 f2A 和 f2B
    f1_module_task.precede(f2C);      // module 之前执行 f2C

    // ---------------------------
    // 创建执行器并运行 f2
    // ---------------------------
    tf::Executor executor;
    executor.run(f2).wait();
    return 0;
}

https://godbolt.org/z/MjTYn3Mrv

理解与注释：

1. 子任务流 f1
   • 包含任务 f1A 和 f1B，并且 f1A 先于 f1B 执行。
2. 主任务流 f2
   • 包含普通任务 f2A, f2B, f2C。
   • 使用 composed_of(f1) 将子任务流 f1 嵌入为模块任务 module。
3. 依赖关系
   • f1_module_task.succeed(f2A, f2B) → 模块执行后再执行 f2A 和 f2B。
   • f1_module_task.precede(f2C) → 模块任务必须在 f2C 执行前完成。
4. 执行器 tf::Executor
   • 用来调度和执行整个任务流 f2。

f1A

f1B

module (f1)

f2A

f2B

f2C

Taskflow Executor 异步任务示例

#include <taskflow/taskflow.hpp>
// Taskflow header-only
#include <iostream>
#include <future>

int main() {
    // 创建一个 Executor 对象，用于运行异步任务
    tf::Executor executor;

    // -------------------------------
    // 1. 创建普通异步任务，返回值类型为 int
    // executor.async 会返回 std::future<int>
    std::future<int> future = executor.async([](){
        std::cout << "async task returns 1\n";
        return 1; // 返回值
    });

    // -------------------------------
    // 2. 创建普通异步任务，不返回值
    executor.silent_async([](){
        std::cout << "async task does not return\n";
    });

    // -------------------------------
    // 3. 创建带依赖关系的异步任务（动态任务图）
    // silent_dependent_async 可以让任务依赖于其他任务
    tf::AsyncTask A = executor.silent_dependent_async([](){ std::cout << "Task A\n"; });
    tf::AsyncTask B = executor.silent_dependent_async([](){ std::cout << "Task B\n"; }, A); // B 在 A 完成后执行
    tf::AsyncTask C = executor.silent_dependent_async([](){ std::cout << "Task C\n"; }, A); // C 在 A 完成后执行
    tf::AsyncTask D = executor.silent_dependent_async([](){ std::cout << "Task D\n"; }, B, C); // D 在 B 和 C 完成后执行

    // -------------------------------
    // 等待所有异步任务完成
    executor.wait_for_all();

    // -------------------------------
    // 获取 future 的返回值
    int result = future.get();
    std::cout << "Future result: " << result << "\n";
    return 0;
}

https://godbolt.org/z/55KKq6Pjc

理解：

1. Executor
   • tf::Executor 是 Taskflow 的运行时，管理线程池与任务调度。
2. 普通异步任务
   • executor.async 返回 std::future，可以获取返回值。
   • executor.silent_async 不返回值，也不阻塞。
3. 动态依赖任务
   • silent_dependent_async 可以让任务依赖其他任务完成后执行。
   • 例如：D 依赖于 B 和 C，只有当 B 和 C 都完成后，D 才会运行。
4. 等待任务完成
   • executor.wait_for_all() 阻塞当前线程直到所有异步任务完成。
5. 获取 future
   • 对于返回值任务，可以通过 future.get() 获取执行结果。

async task returns 1 (future)

async task does not return

Task A

Task B

Task C

Task D

#include <taskflow/taskflow.hpp>
// Taskflow header-only
#include <iostream>

int main() {
    tf::Executor executor;
    tf::Taskflow taskflow;

    // 创建三个简单任务
    auto A = taskflow.emplace([](){ std::cout << "Task A\n"; }).name("A");
    auto B = taskflow.emplace([](){ std::cout << "Task B\n"; }).name("B");
    auto C = taskflow.emplace([](){ std::cout << "Task C\n"; }).name("C");

    // 设置依赖关系
    A.precede(B, C); // A 先于 B 和 C 执行

    // 1⃣ 运行一次 taskflow
    tf::Future<void> run_once = executor.run(taskflow);
    run_once.get(); // 等待完成

    // 2⃣ 运行 4 次
    executor.run_n(taskflow, 4);

    // 3⃣ 运行直到计数器为 0
    executor.run_until(taskflow, [counter=5]() mutable{return --counter == 0;});

    // 阻塞等待所有提交的 taskflow 完成
    executor.wait_for_all();
    return 0;
}

https://godbolt.org/z/hhvGdabzh

注释

1. tf::Executor executor;
   • 创建一个执行器，用于管理线程池和任务调度。
2. tf::Taskflow taskflow;
   • 创建一个任务流容器，可以添加任务并定义依赖关系。
3. taskflow.emplace(...)
   • 创建任务，并用 .name("A") 给任务命名。
4. A.precede(B, C)
   • 表示任务 A 先于 B 和 C 执行，即 B 和 C 都依赖 A。
5. executor.run(taskflow)
   • 运行 taskflow 一次，返回 tf::Future<void> 可用于查询状态。
6. executor.run_n(taskflow, 4)
   • 运行 taskflow 四次。
7. executor.run_until(taskflow, [counter=5]{...})
   • 运行 taskflow 直到 lambda 返回 true，这里用计数器控制执行 5 次。
8. executor.wait_for_all()
   • 阻塞，直到 executor 中所有 taskflow 完成。

对应 Mermaid DAG 图

Task A

Task B

Task C

说明

• A: 起始任务
• B 和 C: 依赖 A 执行的任务
• 任务流可以被执行器多次执行（run_once, run_n, run_until），DAG 表示任务间的依赖关系，而非执行次数。

Taskflow 并行算法

#include <taskflow/taskflow.hpp>
#include <taskflow/algorithm/reduce.hpp>
#include <taskflow/algorithm/sort.hpp>
#include <taskflow/algorithm/for_each.hpp>
#include <vector>
#include <iostream>

int main() {
    tf::Executor executor;
    tf::Taskflow taskflow;
    std::vector<int> data(10, 0); // 创建长度为 10 的数组，初始化为 0
    int init_sum = 0;

    // 1⃣ 并行赋值：将每个元素设置为 100
    tf::Task task1 = taskflow.for_each(
        data.begin(), data.end(),
        [](auto& i){ i = 100; }
    ).name("Parallel ForEach");

    // 2⃣ 并行归约：对所有元素求和
    tf::Task task2 = taskflow.reduce(
        data.begin(), data.end(), init_sum,
        [](auto a, auto b){ return a + b; }
    ).name("Parallel Reduce");

    // 3⃣ 并行排序：对数组排序
    tf::Task task3 = taskflow.sort(
        data.begin(), data.end(),
        [](auto a, auto b){ return a < b; }
    ).name("Parallel Sort");

    // 设置依赖关系
    task1.precede(task2); // 先并行赋值，再归约
    task2.precede(task3); // 再并行排序

    // 执行任务流
    executor.run(taskflow).wait();

    // 输出最终数组
    std::cout << "Sorted data: ";
    for(auto v : data) std::cout << v << " ";
    std::cout << std::endl;
    return 0;
}

https://godbolt.org/z/vWrqPh55c

注释

1. tf::Executor executor
   • 创建执行器，用于管理线程池和任务调度。
2. tf::Taskflow taskflow
   • 创建任务流，用于组合任务和依赖关系。
3. taskflow.for_each
   • 并行遍历数组 data，将每个元素赋值为 100。
4. taskflow.reduce
   • 并行归约数组 data，计算总和 init_sum。
5. taskflow.sort
   • 并行排序数组 data。
6. task1.precede(task2) 和 task2.precede(task3)
   • 设置依赖关系，确保顺序：先赋值 → 再归约 → 再排序。
7. executor.run(taskflow).wait()
   • 执行任务流，并等待完成。

对应 Mermaid DAG 图

Parallel ForEach

Parallel Reduce

Parallel Sort

说明

• Parallel ForEach (A)：并行赋值任务
• Parallel Reduce (B)：并行求和任务
• Parallel Sort (C)：并行排序任务
• DAG 展示了任务间的执行顺序依赖：A → B → C。

Taskflow Pipeline

完整 C++ 示例代码（Pipeline）

#include <taskflow/taskflow.hpp>
#include <taskflow/algorithm/pipeline.hpp>
#include <vector>
#include <cstdio>

int main() {
    tf::Executor executor;
    tf::Taskflow taskflow;
    const size_t num_parallel_lines = 5; // 并行 pipeline 的 token 数量
    std::vector<int> buffer(num_parallel_lines, 0);

    // 创建 pipeline
    tf::Pipeline pl(
        num_parallel_lines,
        // Stage 1: 初始化 token
        tf::Pipe{
            tf::PipeType::SERIAL,
            [&buffer](tf::Pipeflow& pf){
                buffer[pf.line()] = static_cast<int>(pf.token());
                printf("stage 1: token %zu stored in buffer[%zu]\n", pf.token(), pf.line());
                if(pf.token() == 5){
                    pf.stop(); // 当 token = 5 时停止 pipeline
                }
            }
        },
        // Stage 2: 处理 token
        tf::Pipe{
            tf::PipeType::SERIAL,
            [&buffer](tf::Pipeflow& pf){
                printf("stage 2: input buffer[%zu] = %d\n", pf.line(), buffer[pf.line()]);
                buffer[pf.line()] += 10; // 模拟处理
            }
        },
        // Stage 3: 输出 token
        tf::Pipe{
            tf::PipeType::SERIAL,
            [&buffer](tf::Pipeflow& pf){
                printf("stage 3: input buffer[%zu] = %d\n", pf.line(), buffer[pf.line()]);
            }
        }
    );

    // 将 pipeline 添加到任务流
    taskflow.composed_of(pl);

    // 执行任务流
    executor.run(taskflow).wait();
    return 0;
}

https://godbolt.org/z/PqneefM5x

注释

1. tf::Executor executor
   • 创建执行器，用于调度和执行 pipeline。
2. tf::Taskflow taskflow
   • 创建任务流容器，用于组合 pipeline。
3. tf::Pipeline pl
   • 创建一个 pipeline，有 num_parallel_lines 个 token 并行流。
   • 每个 token 会依次经过三个串行 stage（SERIAL Pipe）：
     • Stage 1：初始化 token 并存储到 buffer 中，如果 token == 5 停止 pipeline。
     • Stage 2：处理 token，修改 buffer 内容。
     • Stage 3：输出 token 结果。
4. taskflow.composed_of(pl)
   • 将 pipeline 封装为任务流的一部分。
5. executor.run(taskflow).wait()
   • 执行 pipeline 并等待完成。

对应 Mermaid DAG 图

Stage 1: 初始化 token

Stage 2: 处理 token

Stage 3: 输出 token

说明

• Stage 1：token 初始化
• Stage 2：处理 token（模拟业务逻辑）
• Stage 3：输出 token
• 依赖关系：每个 token 在三个 stage 中顺序执行，但不同 token 之间可以并行流动，实现高吞吐量。