并发编程高性能计算【免费下载链接】oneTBBoneAPI Threading Building Blocks (oneTBB)项目地址https://gitcode.com/gh_mirrors/on/oneTBB点击查看免费下载导读oneapi::tbb::task_group是 oneTBBoneAPI Threading Building Blocks提供的任务并行原语它允许你在程序运行过程中动态地向任务组添加任务并在合适的时机统一等待全部任务完成。与parallel_invoke这类编译期已知任务个数、每层递归必须汇合的 fork-join 结构不同task_group更适合事前不知道会产生多少任务、需要在运行期不断派生新任务的场景。读完本文你将掌握task_group的完整 APIrun/wait/run_and_wait/cancel/defer、理解其与task_group_context、task_group_status的协作机制并能用它在二叉树搜索这类递归任务中实现可提前终止的并行化。本文以官方用户指南 creating_tasks_with_task_group.rst 为主体结合 task_examples.cpp、task_group.h 源码与 task_group_cls.rst 参考文档、test_task_group.cpp 测试用例进行纵深展开。一、为什么需要 task_group从 fork-join 到无结构并行oneTBB 支持直接用任务进行并行化。最直观的入口是oneapi::tbb::parallel_invoke它一次性把若干互相独立的函数体作为任务并行执行参见 creating_tasks_with_parallel_invoke.rst 与 parallel_invoke_func.rst。两者的本质区别在于结构性与任务数量的确定性维度parallel_invoketask_group任务数量编译期固定函数参数包中的个数运行期动态增长事先无需知道数量等待语义fork-join每层调用必须等两侧子树全部完成才返回只有显式调用wait()才阻塞典型场景固定数量的独立子任务递归展开、任务数不可预知的搜索/分治算法并发添加任务不支持支持run()可从多个工作线程安全调用官方用户指南明确说明task_group用于以比parallel_invoke**更不结构化less structured**的方式运行和等待任务当你想创建一组可以并行运行的任务、但事先不知道会有多少个时它就派上了用场。在parallel_invoke版本中由于它是 fork-join 算法递归的每一层都必须等左右子树都完成后才能继续而task_group版本里递归调用并不会在每一层等待——任务被run派发出去后立即返回只在最外层调用一次tg.wait()统一汇合。二、二叉搜索树并行搜索完整示例官方指南给出的示例是在一棵二叉树中并行查找目标值。示例复用了creating_tasks_with_parallel_invoke.rst中定义的struct TreeNode与sequential_tree_search函数完整代码位于 task_examples.cpp。TreeNode定义与串行基线struct TreeNode { int value; TreeNode* left nullptr; TreeNode* right nullptr; ~TreeNode() { delete left; delete right; } }; void sequential_tree_search(TreeNode* node, int target, std::atomicTreeNode* result) { if (node !result.load()) { if (node-value target) { result.store(node); // overwrite is ok since any result is valid } else { sequential_tree_search(node-left, target, result); sequential_tree_search(node-right, target, result); } } }注意因为同一棵树上可能多个线程并发进入串行基线分支result必须用std::atomicTreeNode*持有即使从串行基线读取也一样。使用task_group的并行版本void parallel_tree_search_impl(tbb::task_group tg, TreeNode* node, int target, std::atomicTreeNode* result, size_t depth_threshold initial_depth_threshold) { if (node !result.load()) { if (node-value target) { result.store(node); // overwrite is ok since any result is valid } else if (depth_threshold 0) { sequential_tree_search(node, target, result); } else { // Run on left and right subtrees in parallel tg.run([node, target, result, tg, depth_threshold] { parallel_tree_search_impl(tg, node-left, target, result, depth_threshold - 1); }); tg.run([node, target, result, tg, depth_threshold] { parallel_tree_search_impl(tg, node-right, target, result, depth_threshold - 1); }); } } } TreeNode* parallel_tree_search(TreeNode* root, int target) { if (!root) return nullptr; std::atomicTreeNode* result{nullptr}; tbb::task_group tg; // Start the divide and conquer search with a single task group parallel_tree_search_impl(tg, root, target, result); // Wait for all tasks to complete at the outermost level tg.wait(); return result.load(); }运行入口main函数见 task_examples.cpp会生成一棵约 1000 万节点的随机二叉树依次运行串行搜索、parallel_invoke搜索、基础task_group搜索与带取消的task_group搜索并输出各自耗时与相对串行的加速比用于实测验证。三、示例背后的四个关键设计1. 用run()递归派生任务不在每层等待在parallel_tree_search_impl中task_group::run用于为左右子树各创建新任务。递归调用不会在每一层等待task_grouprun()只是把任务投递给调度器并立即返回调用栈随即展开让当前工作线程可以继续参与其它任务的执行。真正意义上的汇合只发生在最外层parallel_tree_search中对tg.wait()的一次调用。2.depth_threshold限制并行递归深度示例使用递归创建了大量任务。并行递归的深度由depth_threshold参数限制默认值initial_depth_threshold 10见 task_examples.cpp每次递归派发时depth_threshold - 1当阈值减到 0 时不再创建新任务而是退回sequential_tree_search串行搜索整棵子树。这避免了在树的浅层就产生海量微小任务、调度开销超过并行收益的问题。3. 原子变量result与提前终止示例中result会被周期性检查如果值已经被其它并发任务找到当前任务的搜索立即终止if (node !result.load())直接短路。由于多个线程可能同时访问result即便在串行基线里也必须使用std::atomicTreeNode*。注释 overwrite is ok since any result is valid 表明只要找到任意一个匹配节点即可后写覆盖前写是安全的。4. 线程安全的run与逻辑嵌套示例中在task_group tg内执行的任务会通过调用同一个task_group对象的run来创建更多任务。这些run调用是线程安全的tg.wait()会一直阻塞直到所有这些任务包括运行期新添加的全部完成。虽然任务可能由不同的工作线程添加但这些添加逻辑上嵌套在顶层tg.run调用之内因此工作线程中添加任务与主线程中等待任务之间不存在数据竞争。这正是task_group与朴素每层 fork-join在并发安全模型上的核心差异。四、task_group 核心 API 与底层实现4.1 API 总览task_group的公共接口在参考文档 task_group_cls.rst 中有完整定义头文件为include/oneapi/tbb/task_group.h对外通过namespace oneapi::tbb暴露using detail::d2::task_group成员函数语义task_group()构造一个空任务组。所有任务与其关联的task_group_context绑定默认构造时启用concurrent_wait特质task_group(task_group_context context)构造空任务组并让所有加入的任务与指定context关联void run(Func f)添加一个计算f()的任务并立即返回Func须满足 ISO C [function.objects] 要求void run(task_handle h)调度h指向的任务执行要求h非空且由*this创建task_group_status run_and_wait(const Func f)等价于{run(f); return wait();}task_group_status wait()等待组内所有任务完成或被取消void cancel()取消该task_group内的所有任务task_handle defer(Func f)创建延迟任务并返回task_handle任务直到被显式调度如run才执行但wait()仍会等它被调度或销毁bool is_current_task_group_canceling()非成员函数返回当前线程上最内层task_group是否正在取消wait()/run_and_wait()返回task_group_status枚举见 task_group_status_enum.rstenum task_group_status { not_complete, // 未取消且组内任务尚未全部完成 complete, // 未取消且组内任务已全部完成 canceled // 任务组已收到取消请求 };4.2 源码视角run 与 wait 的实现从 task_group.h 源码可以确认task_group继承自task_group_base后者内含一个d1::wait_context_vertex m_wait_vertex等待计数顶点和一个d1::task_group_context m_context取消/异常传播上下文。默认构造函数task_group() : task_group_base(d1::task_group_context::concurrent_wait) {}显式传入concurrent_wait特质以支持等待线程同时参与其它无关任务的语义——这正是参考文档所述执行wait()的线程可能参与与该task_group无关的其它任务。run(F f)的实现是d1::spawn(*prepare_task(std::forwardF(f)), context());。prepare_task通过small_object_allocator分配一个function_task对象并把任务挂到m_wait_vertex上spawn把任务投递进调度器后立即返回。wait()调用d1::wait(m_wait_vertex.get_context(), context())其内部是一个等待计数每个派发任务完成时释放一次引用因此能正确覆盖运行期动态添加的全部任务等待结束后还会读取is_group_execution_cancelled()并调用context().reset()据此返回complete或canceled。4.3 关联的 task_group_contexttask_group底层依赖 task_group_context每个任务只关联一个task_group_context对象多个 context 构成树状森林取消请求会沿父 → 子方向传播到整个子树。要点kind_t::isolated构造的 context 没有父节点kind_t::bound默认当第一个关联任务进入调度器时成为当前最内层任务组的子 context若当前线程无内层任务则退化为 isolated。traits_typefp_settings捕获当前线程浮点设置、default_traits 0头文件 task_group.h 中还有concurrent_wait特质。cancel_group_execution()请求取消若已被取消返回false否则返回true多线程并发调用时恰好一次返回true。reset()将 context 重置为未取消状态但必须等所有关联任务完成后方可调用且非线程安全。五、任务组取消让搜索更早收场基础版本通过周期性检查result提前剪枝而 oneTBB 还提供了显式取消机制。同一个 task_examples.cpp 中给出了可取消版本一旦某任务找到目标值立即调用tg.cancel()并通过tbb::is_current_task_group_canceling()让其它仍在派发/执行的任务快速退出。void parallel_tree_search_cancellable_impl(tbb::task_group tg, TreeNode* node, int target, std::atomicTreeNode* result, size_t depth_threshold initial_depth_threshold) { // tbb::is_current_task_group_canceling() checks associated task_group_context if (node !tbb::is_current_task_group_canceling()) { if (node-value target) { result.store(node); // overwrite is ok since any result is valid // cancel the task_group_context associated with task_group tg.cancel(); // multiple cancellations are ok due to single wait } else if (depth_threshold 0) { sequential_tree_search(node, target, result); if (result.load() ! nullptr) { // cancel the task_group_context associated with task_group tg.cancel(); // multiple cancellations are ok due to single wait } } else { // Run on left and right subtrees in parallel tg.run([node, target, result, tg, depth_threshold] { parallel_tree_search_cancellable_impl(tg, node-left, target, result, depth_threshold - 1); }); tg.run([node, target, result, tg, depth_threshold] { parallel_tree_search_cancellable_impl(tg, node-right, target, result, depth_threshold - 1); }); } } }要点tg.cancel()内部调用context().cancel_group_execution()见 task_group.h多次调用是安全的注释multiple cancellations are ok due to single wait最终由唯一一次wait()统一汇合。tbb::is_current_task_group_canceling()在 task_group.h 中定义为取当前执行任务的task_group_context并查询其是否已收到取消请求。取消请求会沿 context 树传播因此找到结果后整个搜索可以快速收敛避免无谓的遍历。六、异常、析构与必须 wait约束task_group的析构函数有明确约束task_group_cls.rst销毁前必须调用wait()否则析构函数抛出异常。从源码看task_group.h 的析构逻辑是若仍有任务未完成m_wait_vertex.continue_execution()会先尝试cancel()再d1::wait(...)完成清理随后若当前不在栈展开stack unwinding过程中则抛出exception_id::missing_wait异常对外表现为tbb::missing_wait。异常处理方面任务体内抛出的未捕获异常会被 oneTBB 拦截并转为对关联task_group_context的隐式取消请求参考 task_group_context_cls.rst 的说明wait()返回canceled状态。test_task_group.cpp 中有完整的验证TestExceptionHandling1/2/3验证异常在父子任务组间的传播与取消约 L524-L568另有专门测试捕获const tbb::missing_wait约 L608-L611验证未调用wait()即析构时的异常行为L1131-L1140 附近还覆盖了外部线程调用outer.cancel()取消整个任务组的场景。七、线程安全与并发场景验证test_task_group.cpp 针对task_group的线程安全语义提供了系统测试可作为理解文档结论的实证SharedGroupBodyImpl/TestParallelSpawn多线程并发向同一task_group添加任务验证run()的线程安全。TestParallelWait/RunAndWaitSynchronizationTestBody多个线程同时执行run_and_wait与wait()验证等待线程可同时参与其它任务的concurrent_wait语义见 L201-L239。TestVagabondGroup验证流浪任务组在多个线程间迁移时的正确性。FibTaskSymmetricTreeWithFunctor/FibTaskAsymmetricTreeWithFunctor用斐波那契风格的对称/非对称递归任务树验证动态添加任务的完成计数L320-L385。这些测试印证了指南中的核心结论任务可能由不同工作线程添加但由于这些添加逻辑上嵌套在顶层run调用之内添加与等待之间没有竞争。八、选型建议与最佳实践任务数量编译期已知、结构规整→ 优先parallel_invoke如固定两路分治运行期动态派发、数量不可预知→ 使用task_group。务必调用wait()或run_and_wait再销毁task_group否则触发tbb::missing_wait异常若希望在栈展开时安全析构可配合异常处理机制参见 creating_tasks_with_task_group.rst 与 exceptions_and_cancellation 相关文档。控制并行递归深度像示例那样用depth_threshold做阈值深度归零后回退串行避免任务爆炸。共享结果用原子变量多任务并发写共享结果时使用std::atomic并配合周期性检查实现提前终止。需要提前终止时优先利用cancel()is_current_task_group_canceling()比单纯轮询原子标志更彻底能让未派发的任务直接跳过执行。以上全部结论均可在一处对照验证官方用户指南 creating_tasks_with_task_group.rst、配套可运行示例 task_examples.cpp、头文件 task_group.h、参考文档 task_group_cls.rst 与测试 test_task_group.cpp。赞分享并发编程高性能计算【免费下载链接】oneTBBoneAPI Threading Building Blocks (oneTBB)项目地址https://gitcode.com/gh_mirrors/on/oneTBB点击查看免费下载相关推荐oneTBB parallel_invoke 实战指南用 fork-join 模式并行搜索二叉树oneTBB parallel_invoke 实战指南用 fork join 模式并行搜索二叉树 导读 本文基于 oneAPI Threading Build并发编程高性能计算oneTBB 任务级并行编程指南使用 parallel_invoke 与 task_group 实现并行搜索与任务协调oneTBB 任务级并行编程指南使用 parallel_invoke 与 task_group 实现并行搜索与任务协调 导读 oneAPI Threading并发编程高性能计算mold 中的任务级并行oneTBB parallel_invoke 与 task_group 编程实战指南mold 中的任务级并行oneTBB parallel_invoke 与 task_group 编程实战指南 导读 本指南基于仓库内 oneTBB 用户指南开发工具构建工具系统编程上一篇Rye性能基准测试与其他包管理器的终极对比指南下一篇NVIDIA Warp核心教程基础入门与实战指南创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考