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

一、Why Parallel Computing?

Advance performance to a new level previously out of reach 将性能提升到过去无法达到的水平。核心思想：随着单核 CPU 频率提升变慢（Dennard scaling 失效）， 性能提升的主要方式已经从'提高频率'转向'增加并行度'。也就是说：

以前：靠更快的单核
现在：靠更多核心 + GPU + 分布式

二、图表含义分析

图中表示：

Time (minutes) to speed up a machine learning algorithm 横轴：

1 CPU
8 CPUs
16 CPUs
24 CPUs
32 CPUs
40 CPUs
1 GPU

纵轴：

执行时间（分钟）

趋势：

随着 CPU 数量增加，时间下降
但 GPU 下降幅度更大
GPU 可以达到 10–100x 加速

三、为什么 GPU 可以 10–100 倍加速？

1⃣ 并行度数量差异

典型 CPU：

8–64 核
每核适合复杂逻辑

典型 GPU：

几千甚至上万个小核心
适合大规模数据并行

例如： CPU 并行度 ≈ 32 GPU 并行度 ≈ 5000

在高度可并行任务中（如矩阵乘法）：理论加速比 ≈ 5000 / 32 ≈ 156 当然实际不会达到理论值，但 10–100x 是常见的。

四、Amdahl 定律解释加速极限

并行加速并不是无限的。设：

P = 可并行部分比例
N = 并行处理器数量

Amdahl 定律： Speedup(N) = 1 / ((1-P) + P/N)

举例

假设： P = 0.95 (95% 可并行) 若使用 1000 个 GPU 核： Speedup = 1 / (0.05 + 0.95/1000) = 1 / (0.05 + 0.00095) = 1 / 0.05095 ≈ 19.6

即使有 1000 核，最多也只能约 20 倍加速。这说明：

串行部分是性能瓶颈

五、为什么机器学习特别适合 GPU？

机器学习核心是：

矩阵乘法
张量运算
向量运算
卷积
批量数据处理

这些都属于：

数据并行（Data Parallelism）

例如矩阵乘法： C_ij = ∑_k A_ik * B_kj 每个 C_ij 都可以独立计算：这意味着： O(n^2) 个元素可以并行这正是 GPU 擅长的结构。

六、为什么多 CPU 扩展不如 GPU？

1⃣ CPU 核心少 2⃣ 同步开销大 3⃣ 内存带宽限制 4⃣ NUMA 访问延迟 5⃣ 上下文切换开销

而 GPU：

大带宽显存
SIMT 架构
硬件级线程调度

七、理论 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) 如果不并行，计算时间会爆炸。

十、总结

并行计算的重要性在于：

单核性能提升停滞
数据规模指数增长
GPU 提供数量级并行能力
机器学习天然可并行
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 意思是：从算法逻辑结构出发，构建任务图。

例如：

读取数据
预处理
GPU 计算
后处理
输出

形成： 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 而：规则问题越来越少复杂系统越来越多只有任务图可以表达：

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

十二、总结

任务并行的核心优势：

能表达复杂依赖
能适配异构架构
可扩展到未来超大规模系统
能处理不规则问题
以 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

六、执行顺序分析

合法执行顺序：

A
B 和 C（可并行）
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 核心思想：

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

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

五、执行流程解析

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

六、与静态任务图的区别

静态 Taskflow	动态 async
taskflow.emplace	silent_dependent_async
先构建 DAG	边创建边执行
run(taskflow)	直接提交到 executor
适合结构固定	适合运行时动态生成

七、关键性能分析

设：

每个任务耗时 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)) 随着运行动态扩展。

优势：

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

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'？

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

初始化
计算
分支
循环
收敛
结束

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

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

九、性能模型

设：

总工作量 W
关键路径 S

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

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

十、对比传统 while 循环

传统：

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

问题：

隐式串行
不可分析 DAG

CTFG：

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

十一、CTFG vs 数据并行

特性	数据并行	CTFG
表达能力	低	高
控制流	无	有
适合复杂算法	否	是
异构友好	一般	极好

十二、总结

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：

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

数学核心： 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

执行顺序分析

可能的执行顺序：

A
B 和 C 并行
B 内部：
- B1 和 B2 并行
- B3 在 B1 和 B2 之后
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 不会等待

总结

这个例子展示了：

静态主任务图
动态子任务图（Subflow）
条件依赖
嵌套并行
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

说明

init 任务
- 初始化数据结构或准备工作
- 必须在 cond 之前执行
cond 任务
- 条件任务，返回 0 或 1
- Taskflow 会根据返回值选择下一步任务
  - 返回 0 → 再次执行 cond（形成循环）
  - 返回 1 → 执行 stop，结束任务流
stop 任务
- 程序结束前的清理任务
- 只有当 cond 返回 1 时执行
依赖关系
- 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

理解与注释：

子任务流 f1
- 包含任务 f1A 和 f1B，并且 f1A 先于 f1B 执行。
主任务流 f2
- 包含普通任务 f2A, f2B, f2C。
- 使用 composed_of(f1) 将子任务流 f1 嵌入为模块任务 module。
依赖关系
- f1_module_task.succeed(f2A, f2B) → 模块执行后再执行 f2A 和 f2B。
- f1_module_task.precede(f2C) → 模块任务必须在 f2C 执行前完成。
执行器 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

理解：

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

注释

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

注释

tf::Executor executor
- 创建执行器，用于管理线程池和任务调度。
tf::Taskflow taskflow
- 创建任务流，用于组合任务和依赖关系。
taskflow.for_each
- 并行遍历数组 data，将每个元素赋值为 100。
taskflow.reduce
- 并行归约数组 data，计算总和 init_sum。
taskflow.sort
- 并行排序数组 data。
task1.precede(task2) 和 task2.precede(task3)
- 设置依赖关系，确保顺序：先赋值 → 再归约 → 再排序。
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

注释

tf::Executor executor
- 创建执行器，用于调度和执行 pipeline。
tf::Taskflow taskflow
- 创建任务流容器，用于组合 pipeline。
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 结果。
taskflow.composed_of(pl)
- 将 pipeline 封装为任务流的一部分。
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 之间可以并行流动，实现高吞吐量。

一、Why Parallel Computing?

Advance performance to a new level previously out of reach 将性能提升到过去无法达到的水平。核心思想：随着单核 CPU 频率提升变慢（Dennard scaling 失效）， 性能提升的主要方式已经从'提高频率'转向'增加并行度'。也就是说：

以前：靠更快的单核
现在：靠更多核心 + GPU + 分布式

二、图表含义分析

图中表示：

Time (minutes) to speed up a machine learning algorithm 横轴：

1 CPU
8 CPUs
16 CPUs
24 CPUs
32 CPUs
40 CPUs
1 GPU

纵轴：

执行时间（分钟）

趋势：

随着 CPU 数量增加，时间下降
但 GPU 下降幅度更大
GPU 可以达到 10–100x 加速

三、为什么 GPU 可以 10–100 倍加速？

1⃣ 并行度数量差异

典型 CPU：

8–64 核
每核适合复杂逻辑

典型 GPU：

几千甚至上万个小核心
适合大规模数据并行

例如： CPU 并行度 ≈ 32 GPU 并行度 ≈ 5000

在高度可并行任务中（如矩阵乘法）：理论加速比 ≈ 5000 / 32 ≈ 156 当然实际不会达到理论值，但 10–100x 是常见的。

四、Amdahl 定律解释加速极限

并行加速并不是无限的。设：

P = 可并行部分比例
N = 并行处理器数量

Amdahl 定律： Speedup(N) = 1 / ((1-P) + P/N)

举例

假设： P = 0.95 (95% 可并行) 若使用 1000 个 GPU 核： Speedup = 1 / (0.05 + 0.95/1000) = 1 / (0.05 + 0.00095) = 1 / 0.05095 ≈ 19.6

即使有 1000 核，最多也只能约 20 倍加速。这说明：

串行部分是性能瓶颈

五、为什么机器学习特别适合 GPU？

机器学习核心是：

矩阵乘法
张量运算
向量运算
卷积
批量数据处理

这些都属于：

数据并行（Data Parallelism）

例如矩阵乘法： C_ij = ∑_k A_ik * B_kj 每个 C_ij 都可以独立计算：这意味着： O(n^2) 个元素可以并行这正是 GPU 擅长的结构。

六、为什么多 CPU 扩展不如 GPU？

1⃣ CPU 核心少 2⃣ 同步开销大 3⃣ 内存带宽限制 4⃣ NUMA 访问延迟 5⃣ 上下文切换开销

而 GPU：

大带宽显存
SIMT 架构
硬件级线程调度

七、理论 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) 如果不并行，计算时间会爆炸。

十、总结

并行计算的重要性在于：

单核性能提升停滞
数据规模指数增长
GPU 提供数量级并行能力
机器学习天然可并行
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 意思是：从算法逻辑结构出发，构建任务图。

例如：

读取数据
预处理
GPU 计算
后处理
输出

形成： 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 而：规则问题越来越少复杂系统越来越多只有任务图可以表达：

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

十二、总结

任务并行的核心优势：

能表达复杂依赖
能适配异构架构
可扩展到未来超大规模系统
能处理不规则问题
以 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

六、执行顺序分析

合法执行顺序：

A
B 和 C（可并行）
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 核心思想：

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

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

五、执行流程解析

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

六、与静态任务图的区别

静态 Taskflow	动态 async
taskflow.emplace	silent_dependent_async
先构建 DAG	边创建边执行
run(taskflow)	直接提交到 executor
适合结构固定	适合运行时动态生成

七、关键性能分析

设：

每个任务耗时 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)) 随着运行动态扩展。

优势：

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

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'？

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

初始化
计算
分支
循环
收敛
结束

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

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

九、性能模型

设：

总工作量 W
关键路径 S

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

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

十、对比传统 while 循环

传统：

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

问题：

隐式串行
不可分析 DAG

CTFG：

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

十一、CTFG vs 数据并行

特性	数据并行	CTFG
表达能力	低	高
控制流	无	有
适合复杂算法	否	是
异构友好	一般	极好

十二、总结

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：

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

数学核心： 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

执行顺序分析

可能的执行顺序：

A
B 和 C 并行
B 内部：
- B1 和 B2 并行
- B3 在 B1 和 B2 之后
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 不会等待

总结

这个例子展示了：

静态主任务图
动态子任务图（Subflow）
条件依赖
嵌套并行
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

说明

init 任务
- 初始化数据结构或准备工作
- 必须在 cond 之前执行
cond 任务
- 条件任务，返回 0 或 1
- Taskflow 会根据返回值选择下一步任务
  - 返回 0 → 再次执行 cond（形成循环）
  - 返回 1 → 执行 stop，结束任务流
stop 任务
- 程序结束前的清理任务
- 只有当 cond 返回 1 时执行
依赖关系
- 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

理解与注释：

子任务流 f1
- 包含任务 f1A 和 f1B，并且 f1A 先于 f1B 执行。
主任务流 f2
- 包含普通任务 f2A, f2B, f2C。
- 使用 composed_of(f1) 将子任务流 f1 嵌入为模块任务 module。
依赖关系
- f1_module_task.succeed(f2A, f2B) → 模块执行后再执行 f2A 和 f2B。
- f1_module_task.precede(f2C) → 模块任务必须在 f2C 执行前完成。
执行器 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

理解：

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

注释

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

注释

tf::Executor executor
- 创建执行器，用于管理线程池和任务调度。
tf::Taskflow taskflow
- 创建任务流，用于组合任务和依赖关系。
taskflow.for_each
- 并行遍历数组 data，将每个元素赋值为 100。
taskflow.reduce
- 并行归约数组 data，计算总和 init_sum。
taskflow.sort
- 并行排序数组 data。
task1.precede(task2) 和 task2.precede(task3)
- 设置依赖关系，确保顺序：先赋值 → 再归约 → 再排序。
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

注释

tf::Executor executor
- 创建执行器，用于调度和执行 pipeline。
tf::Taskflow taskflow
- 创建任务流容器，用于组合 pipeline。
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 结果。
taskflow.composed_of(pl)
- 将 pipeline 封装为任务流的一部分。
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 之间可以并行流动，实现高吞吐量。