oneTBB 流图 AsyncNodeBody 命名要求解析为 async_node 编写异步体与网关协作【免费下载链接】moldmold: A Modern Linker 项目地址: https://gitcode.com/GitHub_Trending/mo/mold导读本文聚焦 oneAPI Threading Building BlocksoneTBB本仓库内位于 third-party/tbb流图flow graph中的async_node节点深入解析其函数对象Body必须满足的AsyncNodeBody命名要求named requirement。文章以规范文档 async_node_body.rst 为骨架结合 flow_graph.h 中的真实实现与 test_async_node.cpp 测试用例说明如何编写一个能把手头消息提交给外部活动如线程池、异步 I/O、另一套运行时处理并借助 Gateway 与流图双向通信的异步节点体。读完本文你将掌握AsyncNodeBody的完整接口契约、gateway_type的协作语义并能照此写出可编译、可运行的async_node自定义 Body。背景为什么需要 AsyncNodeBody 这一命名要求在 oneTBB 流图中大多数节点如function_node的 Body 是同步执行的消息进入节点后在图的工作线程上就地运行。而async_node的定位是让流图与一个由用户或其他运行时管理的外部活动通信见 async_node_cls.rst。典型场景包括把消息投递给线程池、协程调度器或 GPU/网络异步引擎处理消息在图内到达后立即返回实际计算在别处发生结果再通过网关送回图内。这要求传入async_node的 Body 具备一套精确、可静态检查的接口契约这就是AsyncNodeBody命名要求规范编号[req.async_node_body]。它在 C 标准库命名要求的意义上规定了 Body 类型必须提供的成员函数签名与语义源码中则通过__TBB_requires(async_node_bodyBody, input_type, gateway_type)以概念concept形式强制约束见 flow_graph.h。AsyncNodeBody 的接口契约规范 async_node_body.rst 规定类型Body满足AsyncNodeBody需提供以下三个伪签名pseudo-signature成员。1. 拷贝构造函数Body::Body( const Body )Body 类型必须可拷贝构造。原因在于async_node构造时保存的是用户传入 Body 的拷贝而不是原对象本身。规范在 async_node_cls.rst 中明确指出The body object passed to anasync_nodeis copied. Updates to member variables do not affect the original object used to construct the node.即构造节点后对原 Body 成员变量的修改不会影响节点内部已拷贝的副本同理节点内部运行期间对副本状态的更新外部也无从直接感知。若需在外部检查节点内部 Body 的最新状态应使用tbb::flow::copy_body函数取回一份更新后的拷贝规范见 copy_body_func.rst。测试 test_async_node.cpp 中反复使用tbb::flow::copy_bodycounting_async_serial_body(a)取回节点内部 Body 的计数并与全局计数比对正是这一语义的验证。2. 析构函数Body::~Body()Body 必须可析构用于节点销毁或重置时释放内部状态。3. 调用运算符void Body::operator()( const Input v, GatewayType gateway )这是 AsyncNodeBody 的核心。规范给出了两条硬性约束Input类型必须与构造该async_node时传入的Input模板类型实参完全一致GatewayType类型必须与对应async_node实例的gateway_type成员类型完全一致。这两条约束在编译期由概念async_node_bodyBody, input_type, gateway_type落实见 flow_graph.h 的__TBB_requires断言并在 test/tbb/test_tbb_header.cpp 与 test/conformance/conformance_async_node.cpp 等测试中通过概念校验覆盖。语义上规范只有一句话但分量极重The input valuevis submitted by the flow graph to an external activity. The gateway interface allows the external activity to communicate with the enclosing flow graph.即当async_node收到输入消息v经try_put进入节点后会以v与网关引用为参数调用 BodyBody 的职责是把v提交给某个外部活动去处理投递线程池、写入队列、发起异步调用等处理完毕后由外部活动通过gateway把结果送回流图。Body 自身应立即返回不阻塞图的工作线程——这正是async_node与同步节点的本质区别。与 GatewayType 的协作外部活动的回程通道Body 的第二个参数GatewayType 就是规范 gateway_type.rst 所定义的GatewayType命名要求。async_node的gateway_type成员类型在实现中定义为receiver_gatewayoutput_type见 flow_graph.h它向外暴露三个方法构成外部活动与流图通信的完整接口方法语义bool try_put( const Output v )把结果v广播给对应async_node的全部后继节点Output必须与该async_node的Output模板实参一致返回是否至少有一个后继成功接收void reserve_wait()通知流图已有工作被提交给外部活动图在等待外部完成void release_wait()通知流图此前提交给外部活动的工作已经完成reserve_wait/release_wait与graph::wait_for_all协同外部活动在提交任务前后调用它们保证wait_for_all会一直阻塞到异步工作真正结束。实现位于async_node私有的receiver_gateway_impl中见 flow_graph.hreserve_wait()调用my_graph.reserve_wait()release_wait()调用my_graph.release_wait()try_put则委托给try_put_impl后者通过输出端口 0 的broadcast_cache向后继执行广播式投递并把产生的图任务入队到图的任务竞技场arena中执行flow_graph.h。用户代码通过async_node::gateway()成员函数拿到该接口的引用flow_graph.h。源码实现async_body 与 async_body_base从源码结构看AsyncNodeBody的落地由两层类完成见 flow_graph.hasync_body_baseGateway持有gateway_type *my_gateway指针提供set_gateway以便节点拷贝构造时重新绑定网关async_bodyInput, Ports, Gateway, Body组合用户 Body重载operator()(const Input, Ports)内部用tbb::detail::invoke(my_body, v, *this-my_gateway)把输入值与网关一起转发给用户 Bodynoexcept说明符同样由invoke的noexcept性质推导。async_node构造函数把my_gateway传给async_body见 flow_graph.h实现网关与 Body 的绑定拷贝构造时则通过set_gateway(my_gateway)让新节点的 Body 指向新节点的网关flow_graph.h从而保证每个节点实例的 Body 与自己的网关配对。async_node本身继承自multifunction_nodeInput, std::tupleOutput, Policy并实现senderOutput因此它同时是graph_node、receiverInput与senderOutput具有 discarding 与 broadcast-push 属性并发度上限由构造参数concurrency控制serial、unlimited或 1 到 unlimited 之间的std::size_t值缓冲/拒绝行为由Policylightweight / queueing / rejecting决定——详见 async_node_cls.rst 与 functional_node_policies.rst。实践编写一个满足 AsyncNodeBody 的 Body下面参照 test_async_node.cpp 中的counting_async_unlimited_body给出一个完整可用的示例Body 把输入提交给外部线程模拟外部活动由外部线程通过网关回传结果。#include tbb/flow_graph.h #include thread using node_type tbb::flow::async_nodeint, int; using gateway_ref node_type::gateway_type; // 即 AsyncNodeBody 中的 GatewayType struct async_worker_body { // 1) 拷贝构造编译器默认生成即可 // 2) 析构默认即可 // 3) 核心调用运算符签名必须与 async_nodeint,int 的 Input 和 gateway_type 完全一致 void operator()(const int v, gateway_ref gateway) { // 把消息提交给外部活动这里用独立线程模拟并通知图正在等待外部工作 gateway.reserve_wait(); std::thread([v, gateway] { int result v * 2; // 在外部完成计算 gateway.try_put(result); // 结果广播给所有后继节点 gateway.release_wait(); // 通知图外部工作已完成 }).detach(); } }; int main() { tbb::flow::graph g; node_type a(g, tbb::flow::unlimited, async_worker_body{}); tbb::flow::function_nodeint, int sink(g, tbb::flow::unlimited, [](int x) { /* 消费异步结果 */ }); tbb::flow::make_edge(a, sink); a.try_put(21); g.wait_for_all(); // 会一直等待直到外部线程调用 release_wait() return 0; }要点回顾Body必须可拷贝构造async_node存的是拷贝成员变量改动不回流原对象查内部状态请用copy_bodyoperator()第一参数类型 async_node的Input第二参数类型 async_node::gateway_type即receiver_gatewayOutput偏差会在编译期被概念检查拦截每次提交外部工作前调用gateway.reserve_wait()完成无论成功失败后调用gateway.release_wait()否则wait_for_all可能在异步结果落地前就返回结果通过gateway.try_put(output)广播给全部后继成功与否由返回值指示。测试验证AsyncNodeBody 语义如何被守护仓库中的 test_async_node.cpp对应规范编号[flow_graph.async_node]对上述契约做了系统验证可归纳为四类接口形状定义counting_async_serial_body继承自counting_async_unlimited_body其operator()(const int, counting_gateway_type)与async_nodeint, int, queueing的类型严格对应test_async_node.cpp提交与回传Body 内gateway.try_put(input)把输入原样回传配合多个harness_counting_receiver验证广播语义R个后继都收到消息见 test_async_node.cpp取消与恢复输入-1时调用my_tgc.cancel_group_execution()取消任务组验证取消后剩余消息滞留节点队列、rf_reset_protocol重置后队列被清空且计数保留test_async_node.cppBody 拷贝语义连续使用rf_reset_bodies、rf_clear_edges等重置标志配合copy_body取回的本地计数与全局原子计数比对精确验证节点存副本、外部取新拷贝的语义test_async_node.cpp。此外test/conformance/conformance_async_node.cpp 与 test/conformance/conformance_flowgraph.h 提供面向规范逐条校验的符合性测试是排查自定义 Body 是否符合AsyncNodeBody要求时可参考的行为基准。小结AsyncNodeBody是 oneTBB 流图中桥接图内同步世界与图外异步世界的关键契约它要求 Body 可拷贝构造、可析构并以operator()(const Input, GatewayType)的形式接收图内消息、把消息交给外部活动外部活动再借gateway_type的try_put/reserve_wait/release_wait三方法把结果广播回图并正确参与图的完成等待。理解这套接口是正确编写async_node自定义 Body、避免wait_for_all提前返回或消息丢失的前提。相关规范、实现与测试分别位于 async_node_body.rst、gateway_type.rst、flow_graph.h 与 test_async_node.cpp可对照研读。【免费下载链接】moldmold: A Modern Linker 项目地址: https://gitcode.com/GitHub_Trending/mo/mold创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
oneTBB 流图 AsyncNodeBody 命名要求解析：为 async_node 编写异步体与网关协作
oneTBB 流图 AsyncNodeBody 命名要求解析为 async_node 编写异步体与网关协作【免费下载链接】moldmold: A Modern Linker 项目地址: https://gitcode.com/GitHub_Trending/mo/mold导读本文聚焦 oneAPI Threading Building BlocksoneTBB本仓库内位于 third-party/tbb流图flow graph中的async_node节点深入解析其函数对象Body必须满足的AsyncNodeBody命名要求named requirement。文章以规范文档 async_node_body.rst 为骨架结合 flow_graph.h 中的真实实现与 test_async_node.cpp 测试用例说明如何编写一个能把手头消息提交给外部活动如线程池、异步 I/O、另一套运行时处理并借助 Gateway 与流图双向通信的异步节点体。读完本文你将掌握AsyncNodeBody的完整接口契约、gateway_type的协作语义并能照此写出可编译、可运行的async_node自定义 Body。背景为什么需要 AsyncNodeBody 这一命名要求在 oneTBB 流图中大多数节点如function_node的 Body 是同步执行的消息进入节点后在图的工作线程上就地运行。而async_node的定位是让流图与一个由用户或其他运行时管理的外部活动通信见 async_node_cls.rst。典型场景包括把消息投递给线程池、协程调度器或 GPU/网络异步引擎处理消息在图内到达后立即返回实际计算在别处发生结果再通过网关送回图内。这要求传入async_node的 Body 具备一套精确、可静态检查的接口契约这就是AsyncNodeBody命名要求规范编号[req.async_node_body]。它在 C 标准库命名要求的意义上规定了 Body 类型必须提供的成员函数签名与语义源码中则通过__TBB_requires(async_node_bodyBody, input_type, gateway_type)以概念concept形式强制约束见 flow_graph.h。AsyncNodeBody 的接口契约规范 async_node_body.rst 规定类型Body满足AsyncNodeBody需提供以下三个伪签名pseudo-signature成员。1. 拷贝构造函数Body::Body( const Body )Body 类型必须可拷贝构造。原因在于async_node构造时保存的是用户传入 Body 的拷贝而不是原对象本身。规范在 async_node_cls.rst 中明确指出The body object passed to anasync_nodeis copied. Updates to member variables do not affect the original object used to construct the node.即构造节点后对原 Body 成员变量的修改不会影响节点内部已拷贝的副本同理节点内部运行期间对副本状态的更新外部也无从直接感知。若需在外部检查节点内部 Body 的最新状态应使用tbb::flow::copy_body函数取回一份更新后的拷贝规范见 copy_body_func.rst。测试 test_async_node.cpp 中反复使用tbb::flow::copy_bodycounting_async_serial_body(a)取回节点内部 Body 的计数并与全局计数比对正是这一语义的验证。2. 析构函数Body::~Body()Body 必须可析构用于节点销毁或重置时释放内部状态。3. 调用运算符void Body::operator()( const Input v, GatewayType gateway )这是 AsyncNodeBody 的核心。规范给出了两条硬性约束Input类型必须与构造该async_node时传入的Input模板类型实参完全一致GatewayType类型必须与对应async_node实例的gateway_type成员类型完全一致。这两条约束在编译期由概念async_node_bodyBody, input_type, gateway_type落实见 flow_graph.h 的__TBB_requires断言并在 test/tbb/test_tbb_header.cpp 与 test/conformance/conformance_async_node.cpp 等测试中通过概念校验覆盖。语义上规范只有一句话但分量极重The input valuevis submitted by the flow graph to an external activity. The gateway interface allows the external activity to communicate with the enclosing flow graph.即当async_node收到输入消息v经try_put进入节点后会以v与网关引用为参数调用 BodyBody 的职责是把v提交给某个外部活动去处理投递线程池、写入队列、发起异步调用等处理完毕后由外部活动通过gateway把结果送回流图。Body 自身应立即返回不阻塞图的工作线程——这正是async_node与同步节点的本质区别。与 GatewayType 的协作外部活动的回程通道Body 的第二个参数GatewayType 就是规范 gateway_type.rst 所定义的GatewayType命名要求。async_node的gateway_type成员类型在实现中定义为receiver_gatewayoutput_type见 flow_graph.h它向外暴露三个方法构成外部活动与流图通信的完整接口方法语义bool try_put( const Output v )把结果v广播给对应async_node的全部后继节点Output必须与该async_node的Output模板实参一致返回是否至少有一个后继成功接收void reserve_wait()通知流图已有工作被提交给外部活动图在等待外部完成void release_wait()通知流图此前提交给外部活动的工作已经完成reserve_wait/release_wait与graph::wait_for_all协同外部活动在提交任务前后调用它们保证wait_for_all会一直阻塞到异步工作真正结束。实现位于async_node私有的receiver_gateway_impl中见 flow_graph.hreserve_wait()调用my_graph.reserve_wait()release_wait()调用my_graph.release_wait()try_put则委托给try_put_impl后者通过输出端口 0 的broadcast_cache向后继执行广播式投递并把产生的图任务入队到图的任务竞技场arena中执行flow_graph.h。用户代码通过async_node::gateway()成员函数拿到该接口的引用flow_graph.h。源码实现async_body 与 async_body_base从源码结构看AsyncNodeBody的落地由两层类完成见 flow_graph.hasync_body_baseGateway持有gateway_type *my_gateway指针提供set_gateway以便节点拷贝构造时重新绑定网关async_bodyInput, Ports, Gateway, Body组合用户 Body重载operator()(const Input, Ports)内部用tbb::detail::invoke(my_body, v, *this-my_gateway)把输入值与网关一起转发给用户 Bodynoexcept说明符同样由invoke的noexcept性质推导。async_node构造函数把my_gateway传给async_body见 flow_graph.h实现网关与 Body 的绑定拷贝构造时则通过set_gateway(my_gateway)让新节点的 Body 指向新节点的网关flow_graph.h从而保证每个节点实例的 Body 与自己的网关配对。async_node本身继承自multifunction_nodeInput, std::tupleOutput, Policy并实现senderOutput因此它同时是graph_node、receiverInput与senderOutput具有 discarding 与 broadcast-push 属性并发度上限由构造参数concurrency控制serial、unlimited或 1 到 unlimited 之间的std::size_t值缓冲/拒绝行为由Policylightweight / queueing / rejecting决定——详见 async_node_cls.rst 与 functional_node_policies.rst。实践编写一个满足 AsyncNodeBody 的 Body下面参照 test_async_node.cpp 中的counting_async_unlimited_body给出一个完整可用的示例Body 把输入提交给外部线程模拟外部活动由外部线程通过网关回传结果。#include tbb/flow_graph.h #include thread using node_type tbb::flow::async_nodeint, int; using gateway_ref node_type::gateway_type; // 即 AsyncNodeBody 中的 GatewayType struct async_worker_body { // 1) 拷贝构造编译器默认生成即可 // 2) 析构默认即可 // 3) 核心调用运算符签名必须与 async_nodeint,int 的 Input 和 gateway_type 完全一致 void operator()(const int v, gateway_ref gateway) { // 把消息提交给外部活动这里用独立线程模拟并通知图正在等待外部工作 gateway.reserve_wait(); std::thread([v, gateway] { int result v * 2; // 在外部完成计算 gateway.try_put(result); // 结果广播给所有后继节点 gateway.release_wait(); // 通知图外部工作已完成 }).detach(); } }; int main() { tbb::flow::graph g; node_type a(g, tbb::flow::unlimited, async_worker_body{}); tbb::flow::function_nodeint, int sink(g, tbb::flow::unlimited, [](int x) { /* 消费异步结果 */ }); tbb::flow::make_edge(a, sink); a.try_put(21); g.wait_for_all(); // 会一直等待直到外部线程调用 release_wait() return 0; }要点回顾Body必须可拷贝构造async_node存的是拷贝成员变量改动不回流原对象查内部状态请用copy_bodyoperator()第一参数类型 async_node的Input第二参数类型 async_node::gateway_type即receiver_gatewayOutput偏差会在编译期被概念检查拦截每次提交外部工作前调用gateway.reserve_wait()完成无论成功失败后调用gateway.release_wait()否则wait_for_all可能在异步结果落地前就返回结果通过gateway.try_put(output)广播给全部后继成功与否由返回值指示。测试验证AsyncNodeBody 语义如何被守护仓库中的 test_async_node.cpp对应规范编号[flow_graph.async_node]对上述契约做了系统验证可归纳为四类接口形状定义counting_async_serial_body继承自counting_async_unlimited_body其operator()(const int, counting_gateway_type)与async_nodeint, int, queueing的类型严格对应test_async_node.cpp提交与回传Body 内gateway.try_put(input)把输入原样回传配合多个harness_counting_receiver验证广播语义R个后继都收到消息见 test_async_node.cpp取消与恢复输入-1时调用my_tgc.cancel_group_execution()取消任务组验证取消后剩余消息滞留节点队列、rf_reset_protocol重置后队列被清空且计数保留test_async_node.cppBody 拷贝语义连续使用rf_reset_bodies、rf_clear_edges等重置标志配合copy_body取回的本地计数与全局原子计数比对精确验证节点存副本、外部取新拷贝的语义test_async_node.cpp。此外test/conformance/conformance_async_node.cpp 与 test/conformance/conformance_flowgraph.h 提供面向规范逐条校验的符合性测试是排查自定义 Body 是否符合AsyncNodeBody要求时可参考的行为基准。小结AsyncNodeBody是 oneTBB 流图中桥接图内同步世界与图外异步世界的关键契约它要求 Body 可拷贝构造、可析构并以operator()(const Input, GatewayType)的形式接收图内消息、把消息交给外部活动外部活动再借gateway_type的try_put/reserve_wait/release_wait三方法把结果广播回图并正确参与图的完成等待。理解这套接口是正确编写async_node自定义 Body、避免wait_for_all提前返回或消息丢失的前提。相关规范、实现与测试分别位于 async_node_body.rst、gateway_type.rst、flow_graph.h 与 test_async_node.cpp可对照研读。【免费下载链接】moldmold: A Modern Linker 项目地址: https://gitcode.com/GitHub_Trending/mo/mold创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考