oneTBB 并行化复杂循环：parallel_for_each 与 parallel

并发编程
高性能计算

【免费下载链接】oneTBB

oneAPI Threading Building Blocks (oneTBB)

项目地址：https://gitcode.com/gh_mirrors/on/oneTBB

复杂循环的并行化并不总是像parallel_for那样简单直接。当迭代空间在运行前无法预知、循环体可能动态增加新任务，或数据需要像流水线一样经过多级处理时，oneTBB 提供了两个专门的高层模板：oneapi::tbb::parallel_for_each和oneapi::tbb::parallel_pipeline。本文以官方用户指南 Parallelizing_Complex_Loops.rst 为骨架，结合仓库中的头文件实现与可运行示例，完整讲解这两种模式的使用方法、函数对象约束、管线过滤器模式与性能调优原则，帮助读者在简单的parallel_for模式之外，掌握处理链表遍历、树形递归任务和分阶段数据流等场景的工程化方案。

阅读本文后，你将能够：用parallel_for_each并行遍历未知长度的迭代空间并通过 feeder 动态补充任务；用parallel_pipeline搭建"读入—变换—写出"的多级并行管线，并正确选择 filter mode 与 token 数量；理解管线的吞吐瓶颈、窗口大小与缓冲区复用的取舍。

为什么需要"复杂循环"模式

oneTBB 的 Parallelizing_Simple_Loops 一节介绍了最典型的可扩展并行形态——迭代之间彼此独立、可同时运行。但实际应用中，有些循环并不满足这一前提：

迭代空间未知：例如遍历一个链表，循环结束时才知道总共要处理多少个元素；
循环体会追加工作：例如处理树结构时，处理完一个节点后才知道需要继续处理哪些后代节点；
数据需要按级流动：例如读文件 → 变换 → 写文件的流水线，不同阶段对数据的不同处理方式适合不同的并行度。

上述三类场景分别由parallel_for_each和parallel_pipeline覆盖。官方文档对二者的定位是：在简单的parallel_for之外，为"其他并行模式"提供的备选方案。这一节的 toctree 还列出了配套的 Summary_of_Loops_and_Pipelines，用于汇总高层模板的定位：它们让开发者以任务模式（task-pattern）级别设计软件，而无需关心底层线程的调度细节。

编译提示：本节文档沿用了旧版文档的编译命令表（Windows 下icl /MD example.cpp tbb_debug.lib、Linux 下icc example.cpp -ltbb_debug，去掉_debug即为生产版链接）。在实际使用现代工具链时，更推荐通过仓库根目录的 CMakeLists.txt 配置 CMake 构建，并在链接阶段加入 oneTBB 库，以避免未定义引用。

模式一：parallel_for_each——"煮到熟为止"的并行遍历

适用场景与约束

parallel_for_each解决的问题非常具体：循环结束时迭代空间仍未完全确定，或者循环体会不断往任务列表里添加新工作。官方文档给出的典型例子是链表：

链表的长度在遍历前不可知，访问链表元素天然是串行的；
但如果每个元素的处理（Foo）至少需要几千条指令，那么并行处理这些元素仍然值得。

文档强调：在并行编程中，动态数组通常优于链表，因为链表访问本质上串行；仅当被迫使用链表且单元素处理开销足够大时，才用parallel_for_each换取并行收益。这个"每个 item 至少几千条指令"的门槛，是为了让并行调度的开销能够被实际计算摊薄。

从串行到并行的三步改造

文档以对链表每个元素执行Foo为例，给出串行版本：

void SerialApplyFooToList( const std::list<Item>& list ) { for( std::list<Item>::const_iterator i=list.begin(); i!=list.end(); ++i ) Foo(*i); }

第一步：定义函数对象。parallel_for_each要求函数对象具备const限定的operator()，这一点与标准库<functional>中的函数对象类似，但约束更严格：

class ApplyFoo { public: void operator()( Item& item ) const { Foo(item); } };

第二步：调用算法。把循环体替换为对parallel_for_each的一次调用：

void ParallelApplyFooToList( const std::list<Item>& list ) { parallel_for_each( list.begin(), list.end(), ApplyFoo() ); }

第三步（可选）：理解可扩展性。文档明确点出一个关键事实：parallel_for_each的一次调用绝不会让两个线程同时对同一个输入迭代器取值。这意味着顺序程序中的输入迭代器可以原样使用，但工作的获取（fetching of work）本身是串行的，因此算法并不天然可扩展。尽管如此，在多数场景下仍能获得比纯串行更有用的加速。

两种可扩展的取工作方式

文档进一步给出两条让parallel_for_each获得可扩展性的途径：

随机访问迭代器：当传入的迭代器是 random-access 类型时，oneTBB 可以直接把区间切分给多个线程。这一行为在源码中有直接对应：for_each_root_task针对 std::random_access_iterator_tag 有专门特化，其实现是把区间委托给parallel_for配合parallel_for_body_wrapper处理；而 forward/input 迭代器则走 input_block_handling_task 与 forward_block_handling_task，按每块最多 4 个迭代任务（max_block_size = 4）分批派发。从源码结构看，这就是"random-access 可扩展、其余迭代器逐块串行取值"的实现依据。
feeder 动态补充工作：parallel_for_each的 body 如果接受第二个参数feeder（类型为parallel_for_each<Item>&），就可以在处理过程中调用feeder.add(item)追加新任务。文档以树遍历为例：处理某个节点是处理其所有后代节点的前提，那么处理完节点后，用feeder.add把后代节点加进去即可。这次parallel_for_each调用不会提前结束，直到所有已添加的 item 都被处理完。

feeder 的实现细节同样可以在仓库中验证：

feeder 类提供void add(const Item&)与void add(Item&&)两个重载，分别走internal_add_copy与internal_add_move；
底层feeder_impl会把新 item 包装成feeder_item_task并通过small_object_allocator分配、spawn到当前执行上下文（见 feeder_impl）；
是否创建 feeder 由编译期探测决定：feeder_holder仅在 body 的operator()能接受feeder&参数时才真正构造 feeder 对象（见 feeder_holder），因此带 feeder 与不带 feeder 的 body 可以混用同一算法接口；
函数对象的签名探测通过 parallel_for_each_operator_selector 的重载解析完成，tbb::detail::invoke负责最终调用。

仓库中的一致性测试 test_parallel_for_each.cpp 用static_assert验证了这些约束：带 feeder 的 body（WithFeeder<int>）合法，而operator()非const的 body（OperatorRoundBracketsNonConst）会被编译期拒绝。

使用注意点

body 的operator()必须是const，因为它可能被并发调用；
body 可以接受Item&（可修改元素）或按值/常量引用处理元素，具体取决于迭代器类型与 Item 的可移动性——源码中input_iteration_task_iterator_helper会对支持右值调用的 body 自动采用std::move_iterator传递元素（见 include/oneapi/tbb/parallel_for_each.h#L240-L256）；
若 Item 既不可拷贝也不可移动，feeder.add会在断言中失败，编译期特性探测会尽量提前发现该问题。

模式二：parallel_pipeline——装配线上的并行处理

模式思想与三种过滤器模式

流水线（pipelining）模仿传统制造业的装配线：数据以 token 形式流经一系列过滤器（filter），每个过滤器对数据做某种处理。由于流式数据到达的顺序约束，有些过滤器可以并行，有些不能。例如视频处理中，部分帧操作彼此独立可以同时进行，而另一些帧操作必须先处理前面的帧。

oneTBB 的parallel_pipeline与filter类实现了这一模式。filter_mode在 include/oneapi/tbb/parallel_pipeline.h#L38-L46 中定义，共有三种取值，其含义可结合 base_filter 的位标志理解：

filter_mode	位标志组合	语义
`parallel`	`filter_is_out_of_order`	可并发处理多个 item，且不保证顺序
`serial_in_order`	`filter_is_serial`	一次处理一个 item，所有该模式的过滤器按同一顺序处理
`serial_out_of_order`	`filter_is_serial \\| filter_is_out_of_order`	一次处理一个 item，但不保证顺序

官方示例：文本格式化管线

文档用"读文本文件 → 把每个十进制数字平方 → 写回新文件"的示例演示完整用法。原始文件 I/O 是顺序的，因此输入输出过滤器必须是serial_in_order；中间的平方变换只依赖局部数据，可以指定为parallel，让多个 chunk 同时变换，再按顺序写出。

为了摊薄并行调度的开销，过滤器处理的是大约4000 字符的文本块（chunk）。每个 chunk 由TextSlice对象表示——注意该类在文档和示例 square.cpp 中完全一致：TextSlice的 C++ 声明只是内存中更大对象的头部，实例必须通过类内allocate/free方法分配与释放，即通过oneapi::tbb::tbb_allocator<char>申请sizeof(TextSlice)+max_size+1字节（多出的 1 字节用于存放终止符）。

构建并运行管线的顶层代码（来自文档，与示例实现一致）：

void RunPipeline( int ntoken, FILE* input_file, FILE* output_file ) { oneapi::tbb::parallel_pipeline( ntoken, oneapi::tbb::make_filter<void,TextSlice*>( oneapi::tbb::filter_mode::serial_in_order, MyInputFunc(input_file) ) & oneapi::tbb::make_filter<TextSlice*,TextSlice*>( oneapi::tbb::filter_mode::parallel, MyTransformFunc() ) & oneapi::tbb::make_filter<TextSlice*,void>( oneapi::tbb::filter_mode::serial_in_order, MyOutputFunc(output_file) ) ); }

参数与接口约定（继承自文档并补充实现细节）：

ntoken（第一个参数）控制并行度。概念上 token 流经整个管线：serial_in_order过滤器必须按序串行处理每个 token，parallel过滤器可以让多个 token 并行处理。如果 token 无上限，中间的无序过滤器可能因为输出过滤器跟不上而不断积累 token，导致资源被中间过滤器过度消耗。ntoken指定同时在途（in flight）的 token 上限：一旦达到该上限，输入过滤器不再产生新 token，直到输出过滤器销毁一个 token。该参数在 parallel_pipeline 声明中对应max_number_of_live_tokens。
第二个参数是过滤器链。每个过滤器由make_filter<inputType, outputType>(mode, functor)构造，其中：
- inputType：过滤器输入值的类型，输入过滤器（源）为void；
- outputType：过滤器输出值的类型，输出过滤器（汇）为void；
- mode：上述三种过滤器模式之一；
- functor：定义如何从输入值产生输出值的函数对象。
过滤器用operator&串联，串联时前一过滤器的outputType必须等于后一过滤器的inputType。operator&在 include/oneapi/tbb/parallel_pipeline.h#L104-L109 中实现，它把两个过滤器节点组合成一个新的filter_node，形成一棵解析树；parallel_pipeline最终把这棵树扁平化为底层运行时的过滤器链（见 include/oneapi/tbb/detail/_pipeline_filters.h#L335-L378 的filter_node与引用计数机制）。

先构造再运行的形式（文档同样给出，便于复用过滤器链）：

void RunPipeline( int ntoken, FILE* input_file, FILE* output_file ) { oneapi::tbb::filter<void,TextSlice*> f1( oneapi::tbb::filter_mode::serial_in_order, MyInputFunc(input_file) ); oneapi::tbb::filter<TextSlice*,TextSlice*> f2(oneapi::tbb::filter_mode::parallel, MyTransformFunc() ); oneapi::tbb::filter<TextSlice*,void> f3(oneapi::tbb::filter_mode::serial_in_order, MyOutputFunc(output_file) ); oneapi::tbb::filter<void,void> f = f1 & f2 & f3; oneapi::tbb::parallel_pipeline(ntoken,f); }

注意：filter类还有移动/拷贝构造与赋值、clear()方法（见 filter 类定义），filter 节点通过引用计数（add_ref/remove_ref）管理生命周期，多个filter对象可以共享同一棵过滤器树，因此"先构造再运行"是安全的。

三个 functor 的职责与实现

输出 functor（最简单）：把TextSlice写到文件并释放。由于它位于管线末端，operator()返回void：

void MyOutputFunc::operator()( TextSlice* out ) const { size_t n = fwrite( out->begin(), 1, out->size(), my_output_file ); if( n!=out->size() ) { fprintf(stderr,"Can't write into file '%s'\n", OutputFileName); exit(1); } out->free(); }

变换 functor（中间过滤器）：返回它产出的新TextSlice*。实现要点是先追加终止符（让strtol在数字恰好在 slice 末尾时也能正确工作），再把非数字字符原样复制、数字用strtol解析并平方后以sprintf写回；输出 buffer 分配两倍输入长度，文档与示例注释指出：非负整数 n 的平方位数不可能超过 n 位数的两倍，因此无需溢出检查：

TextSlice* MyTransformFunc::operator()( TextSlice* input ) const { *input->end() = '\0'; char* p = input->begin(); TextSlice* out = TextSlice::allocate( 2*MAX_CHAR_PER_INPUT_SLICE ); char* q = out->begin(); for(;;) { while( p<input->end() && !isdigit(*p) ) *q++ = *p++; if( p==input->end() ) break; long x = strtol( p, &p, 10 ); long y = x*x; sprintf(q,"%ld",y); q = strchr(q,0); } out->set_end(q); input->free(); return out; }

输入 functor（最复杂）：必须保证没有数字跨越 chunk 边界。当它发现某个可能跨入下一个 slice 的不完整数字时，会把数字的残余部分复制到下一个 slice；同时必须指示输入何时结束——通过调用特殊参数flow_control的stop()方法。所有用于管线第一个过滤器的 functor 都必须采用这一约定：

TextSlice* MyInputFunc::operator()( oneapi::tbb::flow_control& fc ) const { if( !next_slice ) next_slice = TextSlice::allocate( MAX_CHAR_PER_INPUT_SLICE ); size_t m = next_slice->avail(); size_t n = fread( next_slice->end(), 1, m, input_file ); if( !n && next_slice->size()==0 ) { // No more characters to process fc.stop(); return NULL; } else { TextSlice* t = next_slice; next_slice = TextSlice::allocate( MAX_CHAR_PER_INPUT_SLICE ); char* p = t->end()+n; if( n==m ) { // Might have read partial number. // If so, transfer characters of partial number to next slice. while( p>t->begin() && isdigit(p[-1]) ) --p; assert(p>t->begin(),"Number too large to fit in buffer.\n"); next_slice->append( p, t->end()+n ); } t->set_end(p); return t; } }

输入 functor 还必须提供拷贝构造函数，因为 functor 在从filter_t（即filter）构造时会被拷贝一次，管线运行时还会再拷贝一次。这一点可以从MyInputFunc的拷贝构造（仅复制input_file）看出。底层实现中，concrete_filter保存的是const Body& my_body引用（见 include/oneapi/tbb/detail/_pipeline_filters.h#L232-L254），而filter_node_leaf持有按值拷贝的const Body my_body（include/oneapi/tbb/detail/_pipeline_filters.h#L438-L447），这印证了"构造时拷贝、运行时再拷贝"的说法。

重要警示：body 的 operator() 必须是 const

文档对此专门给出 CAUTION：提供给管线过滤器的 body 对象可能被拷贝，因此其operator()不应修改 body 自身。否则，修改是否对调用parallel_pipeline的线程可见，取决于operator()作用于原件还是拷贝件，行为不确定。作为提醒，parallel_pipeline要求 body 的operator()声明为const——这一点在MyInputFunc、MyTransformFunc、MyOutputFunc中均有体现。

三种过滤器模式的顺序语义

输入过滤器在本例中必须为serial_in_order，因为它从顺序文件中读取 chunk；
输出过滤器必须按输入顺序写出 chunk，因此同为serial_in_order；
所有serial_in_order过滤器按同一顺序处理 item：如果某个 item 到达MyOutputFunc时晚于MyInputFunc建立起的顺序，管线会自动推迟对该 item 的operator()调用，直到其所有前驱处理完毕；
另一类串行过滤器serial_out_of_order不保持顺序；
中间过滤器只操作局部数据，其 functor 的任意次调用都可以并发运行，因此指定为parallel。

运行时实现上，base_filter用位标志区分串行/并行与有序/无序（include/oneapi/tbb/detail/_pipeline_filters.h#L58-L64），串行过滤器会配备输入缓冲区（my_input_buffer），用于按序暂存到达的 token。

深入：token 数量、吞吐瓶颈与窗口大小

ntoken 的选择

管线吞吐量受两个约束限制（详见 Throughput_of_pipeline）：

token 数量上限：用N个 token 运行时，并行执行的操作不可能超过N个。N选得太低限制并行度，选得太高则可能占用过多资源（例如更多缓冲区）。官方示例 square.cpp 给出的经验值是nthreads * 4，注释说明"每个线程需要多于一个在途 token 才能让所有线程保持忙碌；2–4 倍即可"。
最慢串行过滤器是瓶颈：即使管线没有任何并行过滤器，吞吐量也被最慢的串行过滤器限制——其他过滤器再快也无济于事。因此应尽量让串行过滤器保持轻快，并把工作尽可能移到并行过滤器上。

文本示例的现实速度上限

文档明确提醒：文本处理示例的加速比相对有限，因为串行过滤器受限于系统 I/O 速度——即使文件在本地磁盘上，加速比也不太可能明显超过 2。要让管线真正受益，并行过滤器必须相对于串行过滤器承担较重的计算。

窗口（chunk）大小的权衡

窗口大小（每个 token 的子问题规模）同样限制吞吐：

窗口太小，调度与传递开销可能压过有效工作；
窗口太大，可能溢出 cache；
好的经验法则是：在仍能装进 cache 的前提下尽量取大的窗口大小，通常需要少量实验来确定。

环形缓冲区复用（进阶优化）

Using_Circular_Buffers 介绍了一种降低管线过滤器间分配/释放开销的技巧：当第一个创建 item 的过滤器与最后一个消费 item 的过滤器都是serial_in_order时，可以通过大小至少为ntoken的环形缓冲区来分配与释放 item。原因在于：最多只有ntoken个 item 同时在途，且 item 会按分配顺序被释放；因此环形缓冲区回绕到复用某个 item 时，该 item 必然已经结束上一次在管线中的使用。反之，如果首尾过滤器不是serial_in_order，则必须自己跟踪哪些缓冲区正在使用中，因为缓冲区不会按分配顺序退役。

线性管线的局限

Non-Linear_Pipelines 说明：parallel_pipeline只支持线性管线，不直接处理分支、汇合等复杂拓扑。若业务需要分叉/合并结构，可以把过滤器按拓扑排序压成线性顺序来近似。代价分析如下：

线性化损失的只是延迟（latency），而非吞吐量（throughput）。延迟是 token 从管线首端流到末端的时间；
例如一个原本 A、B 可并发、D、E 也可并发的五过滤器拓扑，原始延迟相当于 3 级过滤器，线性化后延迟变为 5 级；
吞吐量不变，因为无论拓扑如何，吞吐仍受最慢串行过滤器限制；
线性化后，那些"不需要处理、只需透传"的过滤器（A、B、D、E）需要修改其 functor 以正确透传对象。

文档的结论是：parallel_pipeline不支持非线性管线，是一个"收益对代价"的合理权衡——若支持非线性会显著增加编程复杂度，且并不改善吞吐。

从模式到实战：结合示例运行与验证

仓库中的 examples/parallel_pipeline/square 目录提供了与文档完全对应的可运行示例，其 README.md 说明了构建与运行方式：

# 构建 cmake <path_to_example> cmake --build . # 运行（也可使用预设 target） make run_square # 预定义参数 make perf_run_square # 性能测试参数 make light_test_square # 快速验证参数

命令行参数（与 square.cpp 的 main 对应）：

n-of-threads：使用的线程数；可用low[:high]区间或auto，示例内部通过oneapi::tbb::global_control的max_allowed_parallelism控制；
input-file/output-file：输入输出文件名；
max-slice-size：每个 slice 的最大字符数（对应文档中的MAX_CHAR_PER_INPUT_SLICE，默认 4000）；
silent：除耗时外不输出其他信息。

示例的main会先用 1 个线程做串行基准运行，再用自动选择的线程数做并行运行，并借助oneapi::tbb::tick_count计时，方便读者直观对比加速效果——这正是文档所讲"吞吐受串行过滤器限制"这一结论的实测入口。

小结：两种复杂循环模式的选用准则

场景	首选模板	关键注意点
迭代空间未知（链表等）	`parallel_for_each`	body 的`operator()`必须`const`；单元素处理需足够重（几千条指令量级）
循环体动态追加任务（树遍历等）	`parallel_for_each`+`feeder.add`	feeder 追加的 item 会在本次调用内处理完；迭代器为 random-access 时可扩展
数据分阶段流动、各阶段并行度不同	`parallel_pipeline`	输入过滤器必须用`flow_control::stop()`标识结束；body 的`operator()`必须`const`
管线吞吐调优	—	合理选择`ntoken`（经验值：线程数×2~4）；窗口大小在能装进 cache 的前提下取大值；把重活移入并行过滤器
首尾均为 serial_in_order 的管线	—	可用大小≥ntoken 的环形缓冲区消除分配/释放开销
需要分支/汇合拓扑	—	拓扑排序线性化，接受延迟增加、吞吐不变

这些高层模板的目标，正如 Summary_of_Loops_and_Pipelines 所总结：以任务模式级别设计软件、让 generic 模板按需定制，从而在不必从头管理线程的前提下，充分挖掘多核芯片的计算能力。需要进一步深入时，可以阅读本文引用的源码文件 include/oneapi/tbb/parallel_for_each.h、include/oneapi/tbb/parallel_pipeline.h、include/oneapi/tbb/detail/_pipeline_filters.h，以及示例 examples/parallel_pipeline/square/square.cpp 与测试 test/tbb/test_parallel_for_each.cpp。