并发编程高性能计算【免费下载链接】oneTBBoneAPI Threading Building Blocks (oneTBB)项目地址https://gitcode.com/gh_mirrors/on/oneTBB点击查看免费下载导读本文围绕 oneTBBoneAPI Threading Building BlocksFlow Graph 的嵌套并行技巧展开讲解两种核心做法在节点体内嵌套其他并行算法如parallel_for以提升可扩展性以及在节点体内构造并执行嵌套的流图。读完本文后你将掌握如何把 Flow Graph 当作一种协调语言——在图层面表达粗粒度并行在节点内部表达细粒度并行——并学会在嵌套图结构不变时通过复用持久化图来消除不必要的重建开销。原文出自仓库文档 Flow_Graph_nested_parallelism_tips.rst包含 use_nested_algorithms.rst 与 use_nested_flow_graphs.rst 两篇姊妹篇。为什么要嵌套并行把 Flow Graph 当作协调语言提高 Flow Graph 可扩展性的一条强大途径是在节点体node body内部嵌套其他并行算法。这样做的意义在于你可以把 Flow Graph 用作一种协调语言coordination language——在图的层面表达最粗粒度的并行在节点内部嵌套更细粒度的并行从而把任务如何组合与单个任务内部如何并行两个层次解耦。从实现上看这种设计之所以可行是因为 Flow Graph 的节点执行与底层 oneTBB 任务调度器深度融合节点体最终被封装为图任务graph task提交到与图关联的任务竞技场task arena中执行见 src/tbb/task_dispatcher.cpp 与 include/oneapi/tbb/detail/_flow_graph_impl.h 中的graph_task。因此当节点体内部再调用parallel_for等并行算法时这些算法生成的子任务会被同一调度器无缝接纳实现嵌套并行的自然展开。技巧一在节点体内嵌套并行算法场景与图结构文档给出的示例是一个典型的生产者–变换–消费者流水线由五个节点组成一个input_nodematrix_source从一个文件按顺序读取矩阵序列两个function_noden1、n2接收矩阵并对每个元素应用一个函数f1/f2生成两个新矩阵两个终结点function_noden1_sink、n2_sink分别消费n1和n2产出的结果矩阵。连接关系为matrix_source同时连接n1与n2广播n1连接n1_sinkn2连接n2_sink。其中read_next_matrix、f1、f2、consume_f1、consume_f2等函数在原文中未给出需由读者自行实现例如对矩阵逐元素应用f1后返回新矩阵consume_*负责释放或汇总结果。完整代码示例以下代码完整复刻原文示例其中的关键点是在n1和n2的 lambda 表达式中嵌套了parallel_for对矩阵的每个元素并行执行变换graph g; input_node double * matrix_source( g, - double* { double *a read_next_matrix(); if ( a ) { return a; } else { fc.stop(); return nullptr; } } ); function_node double *, double * n1( g, unlimited, - double * { double *b new double[N]; parallel_for( 0, N, { b[i] f1(a[i]); } ); return b; } ); function_node double *, double * n2( g, unlimited, - double * { double *b new double[N]; parallel_for( 0, N, { b[i] f2(a[i]); } ); return b; } ); function_node double *, double * n1_sink( g, unlimited, []( double *b ) - double * { return consume_f1(b); } ); function_node double *, double * n2_sink( g, unlimited, []( double *b ) - double * { return consume_f2(b); } ); make_edge( matrix_source, n1 ); make_edge( matrix_source, n2 ); make_edge( n1, n1_sink ); make_edge( n2, n2_sink ); matrix_source.activate(); g.wait_for_all();关键 API 的底层语义input_node与flow_controlinput_node是无前驱、只做数据源的可执行节点定义于 include/oneapi/tbb/flow_graph.h。它创建后默认处于未激活状态需要显式调用activate()才会开始向后继节点投递消息——这正是示例末尾matrix_source.activate()的含义。节点体接收一个oneapi::tbb::flow_control 参数其实现位于 include/oneapi/tbb/detail/_pipeline_filters.h内部维护bool is_pipeline_stopped调用fc.stop()即置位该标志input_node在调用完节点体后检查该标志若已停止则不再缓存/投递本次返回的nullptr见try_reserve_apply_body中对control.is_pipeline_stopped的判断flow_graph.h。因此示例中读到结尾返回nullptr并fc.stop()是标准的停止输入流的模式。unlimited并发度function_node的构造参数concurrency决定节点体可以被多少个并发调用。unlimited与serial定义于 include/oneapi/tbb/flow_graph.henum concurrency { unlimited 0, serial 1 };。使用unlimited意味着每个到达的消息都可以立即启动一个新的节点体执行节点之间、以及节点体内部的嵌套并行算法之间可以充分重叠——这是本例中两个矩阵分支能够并行推进、同时每个矩阵内部又能再拆成多路并行的前提。function_node的构造与内部function_input/function_output的实现参见 flow_graph.h。嵌套算法的并行度叠加示例里n1与n2是unlimited的两者可以并行各自内部的parallel_for(0, N, ...)又会把单矩阵的逐元素变换进一步切分给多个工作线程。整体效果是图层面并行 × 节点内部并行共同压满硬件线程。需要留意的是节点体内创建的新数组b由节点负责释放示例中由consume_f1/consume_f2处理这是开发者需要自行保证的内存生命周期约定。技巧二嵌套流图Flow Graph 内嵌 Flow Graph除了在节点体内嵌套算法还可以在节点体内嵌套整个流图外层图g的两个节点a、b各自在收到消息时构造并执行一个内层图。节点a收到消息后构造并执行一个内层依赖图dependence graph其节点以continue_msg为消息类型通过make_edge形成n1 → n2、n1 → n3、n2 → n4、n3 → n4的菱形依赖结构节点b收到消息后构造并执行一个内层数据流图data flow graph由四个function_nodem1m4以相同的菱形拓扑连接并注入整数消息。完整代码示例graph g; function_node int, int a( g, unlimited, []( int i ) - int { graph h; node_t n1( h, { cout n1: i \n; } ); node_t n2( h, { cout n2: i \n; } ); node_t n3( h, { cout n3: i \n; } ); node_t n4( h, { cout n4: i \n; } ); make_edge( n1, n2 ); make_edge( n1, n3 ); make_edge( n2, n4 ); make_edge( n3, n4 ); n1.try_put(continue_msg()); h.wait_for_all(); return i; } ); function_node int, int b( g, unlimited, []( int i ) - int { graph h; function_node int, int m1( h, unlimited, []( int j ) - int { cout m1: j \n; return j; } ); function_node int, int m2( h, unlimited, []( int j ) - int { cout m2: j \n; return j; } ); function_node int, int m3( h, unlimited, []( int j ) - int { cout m3: j \n; return j; } ); function_node int, int m4( h, unlimited, []( int j ) - int { cout m4: j \n; return j; } ); make_edge( m1, m2 ); make_edge( m1, m3 ); make_edge( m2, m4 ); make_edge( m3, m4 ); m1.try_put(i); h.wait_for_all(); return i; } ); make_edge( a, b ); for ( int i 0; i 3; i ) { a.try_put(i); } g.wait_for_all();为什么必须调用h.wait_for_all()在第一种实现中内层图h是节点体作用域内的局部变量每次调用都会在离开作用域时被析构。而 oneTBB 的graph析构函数本身会调用wait_for_all()等待图中所有未完成任务执行完毕见 include/oneapi/tbb/detail/_flow_graph_impl.h 中graph::~graph()的实现。因此若节点体结束时不显式调用h.wait_for_all()内层图的析构会隐式等待语义上仍然安全但显式调用h.wait_for_all()能让等待内层图空闲这一意图更加清晰并让b的节点体阻塞至内层图完成从而保证返回i时内层图已彻底结束便于测试与排错。graph::wait_for_all的实现依赖于图内置的等待计数机制reserve_wait/release_wait见 flow_graph_impl.h图不会从wait_for_all返回直到所有reserve_wait都有对应的release_wait与之匹配。依赖图 vs 数据流图两种内层图的差异上述示例有意展示了两种内层图依赖图节点以continue_msg驱动n1.try_put(continue_msg())。continue_msg在 flow_graph.h 中定义为一个空标记类型continue_receiver则负责维护前驱计数只有所有前驱都投递过continue_msg后节点才会执行flow_graph.h。它表达的语义是依赖满足即触发适合建模 DAG 式的任务依赖关系数据流图节点以真实的整数消息驱动m1.try_put(i)消息沿边逐级变换传递适合建模数据加工管线。两者都是嵌套图的有效形态选择哪一种取决于内层任务之间的耦合方式是依赖关系还是数据流动。技巧三复用持久化内层图消除重建开销如果嵌套图的结构在节点的多次调用之间保持不变那么每次调用都重新构造一遍图是冗余的——重建只会在执行上增加不必要的开销。文档对此给出了优化方案把内层图提升为外层作用域的持久对象节点b每次调用时直接向已存在的图投递消息。复用版完整代码示例graph h; function_node int, int m1( h, unlimited, []( int j ) - int { cout m1: j \n; return j; } ); function_node int, int m2( h, unlimited, []( int j ) - int { cout m2: j \n; return j; } ); function_node int, int m3( h, unlimited, []( int j ) - int { cout m3: j \n; return j; } ); function_node int, int m4( h, unlimited, []( int j ) - int { cout m4: j \n; return j; } ); make_edge( m1, m2 ); make_edge( m1, m3 ); make_edge( m2, m4 ); make_edge( m3, m4 ); graph g; function_node int, int a( g, unlimited, []( int i ) - int { graph h; node_t n1( h, { cout n1: i \n; } ); node_t n2( h, { cout n2: i \n; } ); node_t n3( h, { cout n3: i \n; } ); node_t n4( h, { cout n4: i \n; } ); make_edge( n1, n2 ); make_edge( n1, n3 ); make_edge( n2, n4 ); make_edge( n3, n4 ); n1.try_put(continue_msg()); h.wait_for_all(); return i; } ); function_node int, int b( g, unlimited, - int { m1.try_put(i); h.wait_for_all(); // 可选h 不会被析构 return i; } ); make_edge( a, b ); for ( int i 0; i 3; i ) { a.try_put(i); } g.wait_for_all();复用后wait_for_all变得可选文档特别指出在复用版中只有当你希望b的节点体阻塞等待内层图执行完毕时才需要在每次调用末尾调用h.wait_for_all()。因为在第一版实现中图h在离开作用域时被析构析构会隐式等待而复用版中h是持久对象不会在调用结束时被销毁所以即使只调用m1.try_put(i)后直接返回、不等待h变为空闲也是合法的——内层图的消息会在后台继续执行。这一优化有两个实践要点生命周期管理持久图h的存活时间必须覆盖外层图g的全部使用周期例如两者都定义为main作用域内的局部变量或具有合适的对象生命周期避免悬垂引用并发安全如果外层图允许b的多个副本并发执行unlimited并发度那么对同一个持久内层图h的并发try_put需要自行评估其线程安全性若需要串行化对内层图的访问应把b的并发度设为serial或在内层图访问外加锁。测试与验证依据仓库中与本文主题相关的验证素材包括流图节点基础行为测试test/tbb/test_flow_graph.cpp 与 test/tbb/test_function_node.cpp覆盖function_node、input_node、make_edge的消息投递与并发语义嵌套复合节点测试test/tbb/test_composite_node.cpp 中的test_nested_adderL296验证了复合节点内部再嵌套节点的场景可作为嵌套结构用法的补充参考嵌套并行与任务上下文测试test/tbb/test_eh_algorithms.cpp 讨论了嵌套parallel_for/parallel_reduce与任务组上下文的关系test/tbb/test_arena_priorities.cpp 则包含嵌套 arena相关场景的测试L355-L372印证了 oneTBB 对多层嵌套并行执行的支持。小结oneTBB Flow Graph 的嵌套并行提供了两种互补的扩展手段手段适用场景关键注意点节点体内嵌套并行算法parallel_for等单个节点内部存在可并行的细粒度计算节点并发度设为unlimited以充分重叠注意节点体内内存的分配与释放节点体内嵌套流图节点需要表达一组内部任务之间的依赖/数据关系内层图被析构时会隐式wait_for_all显式等待以阻塞节点体复用持久化内层图内层图结构在多次调用间不变仅在需要阻塞时调用h.wait_for_all()注意生命周期与并发访问把粗粒度并行交给图的拓扑把细粒度并行交给节点体内的算法再把结构固定的嵌套图持久化复用——这套组合拳能让你在保持图结构清晰的同时最大化硬件资源的利用率。更多流图与嵌套并行的背景知识可继续阅读仓库中的 Flow_Graph.rst、Nodes.rst 与 Guiding_Task_Scheduler_Execution.rst 等文档。赞分享并发编程高性能计算【免费下载链接】oneTBBoneAPI Threading Building Blocks (oneTBB)项目地址https://gitcode.com/gh_mirrors/on/oneTBB点击查看免费下载相关推荐FAIR Chemistry 生成模型全景ADiT、FlowMM、FlowLLM 与 Crystal-text-llm 的分子与材料生成技术指南FAIR Chemistry 生成模型全景ADiT、FlowMM、FlowLLM 与 Crystal text llm 的分子与材料生成技术指南 本文系统梳理并发编程高性能计算mold 内嵌 oneTBB 并行基石task_group_context 取消与嵌套并行深度解析mold 内嵌 oneTBB 并行基石task_group_context 取消与嵌套并行深度解析 mold 链接器的并行加速高度依赖内嵌的 oneTBB见开发工具构建工具系统编程oneTBB 嵌套并行取消机制详解task_group_context、隔离上下文与流图取消传播oneTBB 嵌套并行取消机制详解task_group_context、隔离上下文与流图取消传播 导读 本文围绕 oneTBBoneAPI Threadin并发编程高性能计算上一篇Changes架构设计原理分布式构建协调系统实现详解 下一篇DLSS Swapper 完整指南:游戏 DLSS 版本管理一键切换不折腾创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考