mold 第三方组件 oneTBBflow_graph 中 async_node 节点参考与实现解析【免费下载链接】moldmold: A Modern Linker 项目地址: https://gitcode.com/GitHub_Trending/mo/mold本文以 mold 仓库内嵌的 oneTBBIntel Threading Building Blocks规格文档flow_graph.async_node为主体系统讲解async_node的定位、类接口、类型要求、成员函数语义与并发/策略约束并结合 flow_graph.h 中的实际实现剖析 gateway 机制、消息广播路径与 body 复制语义。读完本文你能够掌握如何在 TBB 流图flow graph中把消息桥接到图外部的异步活动外部运行时、硬件回调、事件循环等并理解节点内部try_put/try_get与gateway的调用链。async_node 是什么流图与外部活动之间的桥梁async_node的规格定义见 async_node 规格文档。其核心定位是启用流图flow graph与外部活动由用户代码或其他运行时管理的活动之间通信的节点。典型使用模式是消息从流图上游流入async_node节点在收到消息时调用用户提供的 bodybody 的职责是把消息提交给图之外的外部活动例如提交到独立的任务队列、I/O 线程或另一套运行时而不是在流图内部同步处理外部活动处理完成后通过gateway_type接口把结果消息送回流图广播给该节点的下游后继。也就是说async_node同时承担两个方向的角色对上游而言它是receiverInput对下游而言它是senderOutput而 body 与 gateway 分别负责出图和入图两条通道。类声明与模板参数规格文档给出的类声明为// Defined in header oneapi/tbb/flow_graph.h namespace oneapi { namespace tbb { namespace flow { template typename Input, typename Output, typename Policy /*implementation-defined*/ class async_node : public graph_node, public receiverInput, public senderOutput { public: templatetypename Body async_node( graph g, size_t concurrency, Body body, Policy /*unspecified*/ Policy(), node_priority_t priority no_priority ); templatetypename Body async_node( graph g, size_t concurrency, Body body, node_priority_t priority no_priority ); async_node( const async_node src ); ~async_node(); using gateway_type /*implementation-defined*/; gateway_type gateway(); bool try_put( const input_type v ); bool try_get( output_type v ); }; } // namespace flow } // namespace tbb } // namespace oneapi结合 flow_graph.h 的实际实现可以确认几个规格中标注实现定义的部分template typename Input, typename Output, typename Policy queueing_lightweight __TBB_requires(std::default_initializableInput std::copy_constructibleInput) class async_node : public multifunction_node Input, std::tuple Output , Policy , public sender Output { typedef multifunction_node Input, std::tuple Output , Policy base_type;默认策略为queueing_lightweight轻量排队策略即并发达到上限时默认把消息排队而非拒绝基类multifunction_node是内部封装对外呈现的仍是规格所声明的graph_nodereceiverInputsenderOutput三重身份multifunction_node本身继承自graph_node并实现receiverInput的try_put__TBB_requires(std::default_initializableInput std::copy_constructibleInput)这一约束表达式与规格文档的 Requirements 一一对应见下节。类型要求Requirements规格文档明确列出三条类型要求实现中均可找到对应证据Input类型必须满足 ISO C 标准的 DefaultConstructible 与 CopyConstructible 要求——对应源码中的__TBB_requires(...)约束flow_graph.hPolicy类型可以指定为 功能节点策略文档 中的轻量/排队/拒绝lightweight、queueing、rejecting组合也可以采用默认值。策略类型别名在 detail/_flow_graph_body_impl.h 中定义typedef Policyqueueing, lightweight queueing_lightweight; typedef Policyrejecting, lightweight rejecting_lightweight;Body类型必须满足 AsyncNodeBody 要求。该要求规定Body必须可复制构造、可析构并提供void operator()(const Input v, GatewayType gateway)成员函数其中Input与GatewayType必须分别匹配async_node的模板参数和gateway_type成员类型。此外自 C17 起Body还可以是一个指向Input类中接受gateway_type参数的 const 成员函数的指针——即可以把消息类型本身当作 body 的宿主。源码中以__TBB_requires(async_node_bodyBody, input_type, gateway_type)对两个带 body 的构造函数施加该约束flow_graph.h而 flow_graph_abstractions.h 提供了对应的抽象与检查设施。转发与缓冲属性、并发上限规格文档对async_node的图论属性给出了明确结论转发属性为 discarding丢弃型节点不会把输入消息原样转发给下游而是交由 body 处理缓冲属性为 broadcast-push广播推送型body 通过 gateway 提交的输出消息会推送push给所有后继并广播broadcast到每一个后继。并发上限方面async_node支持用户设置的 concurrency 参数既可以取 预定义并发上限 中的值也可以传入std::size_t数值把并发限制在 1 到tbb::flow::unlimited之间。它限制的是该节点 body 被同时调用simultaneous invocations的次数超出上限时try_put的行为取决于策略——排队策略会把消息暂存queued拒绝策略则直接拒绝该消息。成员函数逐一解析构造函数无策略重载templatetypename Body async_node( graph g, size_t concurrency, Body body, node_priority_t priority no_priority );构造一个调用body副本的async_nodeconcurrency值限制该节点 body 同时调用的数量。priority参数指定 节点优先级。实现侧flow_graph.h这个重载是委托构造直接转发到带 Policy 的重载并使用默认策略template typename Body __TBB_requires(async_node_bodyBody, input_type, gateway_type) __TBB_NOINLINE_SYM async_node(graph g, size_t concurrency, Body body, node_priority_t a_priority) : async_node(g, concurrency, body, Policy(), a_priority) {}构造函数带策略重载templatetypename Body async_node( graph g, size_t concurrency, Body body, Policy /*unspecified*/ Policy(), node_priority_t priority no_priority );同样构造一个调用body副本的async_node但额外允许显式指定策略从而控制并发上限触达后排队还是拒绝。主构造函数实现flow_graph.h值得注意的两点templatetypename Body __TBB_requires(async_node_bodyBody, input_type, gateway_type) __TBB_NOINLINE_SYM async_node( graph g, size_t concurrency, Body body, Policy Policy(), node_priority_t a_priority no_priority ) : base_type( g, concurrency, async_bodyInput, typename base_type::output_ports_type, gateway_type, Body (body, my_gateway), a_priority ), my_gateway(self()) { fgt_multioutput_node_with_body1( ... ); }body 被包装成async_bodyInput, Ports, gateway_type, Body并在构造时就持有了指向本节点my_gateway的指针——这就是body 通过 gateway 与流图通信的底层绑定方式fgt_multioutput_node_with_body1(CODEPTR(), FLOW_ASYNC_NODE, ...)是面向性能剖析/跟踪profiling的节点注册点说明框架把 async_node 作为一类独立的多输出带 body 节点纳入观测体系。拷贝构造函数async_node( const async_node src )语义要点与规格文档一致新节点具有src构造时的初始状态引用与src相同的graph对象持有src初始 body 的一份拷贝并使用与src相同的并发阈值src的前驱和后继不会被复制新 body 是从传给src的原始 body 的拷贝再拷贝构造而来因此src构造之后对其 body 成员变量的修改不会影响新节点的 body。实现flow_graph.h中有一处容易被忽视但很关键的操作__TBB_NOINLINE_SYM async_node( const async_node other ) : base_type(other), senderOutput(), my_gateway(self()) { static_castasync_body_base_type*(this-my_body-get_body_ptr())-set_gateway(my_gateway); static_castasync_body_base_type*(this-my_init_body-get_body_ptr())-set_gateway(my_gateway); ... }由于每个async_node实例拥有自己独立的my_gateway拷贝构造后必须把内部 body以及保存的初始 body中的 gateway 指针重定向到新节点自己的 gateway否则外部活动会把结果错误地送回源节点。gateway() 与 gateway_typegateway_type gateway()返回gateway_type接口的引用。规格规定gateway_type满足 GatewayType 要求该接口包含三个操作方法语义bool try_put(const Output v)把v广播给对应async_node的所有后继Output必须与节点模板参数一致void reserve_wait()通知流图已有工作被提交到外部活动使graph::wait_for_completion能够感知图外未完成的工作void release_wait()通知流图提交到外部活动的工作已完成配对释放reserve_wait()持有的等待引用实现上gateway_type的具体类型是receiver_gatewayOutputflow_graph.h节点内部的receiver_gateway_impl完成了三个操作的落地flow_graph.hclass receiver_gateway_impl: public receiver_gatewayOutput { public: receiver_gateway_impl(async_node* node): my_node(node) {} void reserve_wait() override { fgt_async_reserve(static_casttypename async_node::receiver_type *(my_node), my_node-my_graph); my_node-my_graph.reserve_wait(); } void release_wait() override { async_node* n my_node; graph* g n-my_graph; g-release_wait(); fgt_async_commit(static_casttypename async_node::receiver_type *(n), g); } //! Implements gateway_type::try_put for an external activity to submit a message to FG bool try_put(const Output i) override { return my_node-try_put_impl(i); } ... };可以看到reserve_wait/release_wait成对地驱动graph的等待引用计数——这正是流图能够等待外部活动完成的机制基础而 gateway 的try_put转发到节点的try_put_impl。try_putbool try_put( const input_type v )若并发上限允许对传入消息v执行用户提供的 body否则按节点策略要么把消息v排队要么拒绝它。返回值输入被接受返回true否则返回false例如拒绝策略下并发已满。try_getbool try_get( output_type v )返回值恒为false。这符合 broadcast-push 属性async_node是纯推送型 sender不缓冲输出因此拉取方永远取不到东西下游若需要缓冲应使用 queue_node、buffer_node 等带缓冲能力的节点承接。输出消息的广播路径try_put_impl外部活动通过 gateway 提交结果时最终走到节点的私有try_put_implflow_graph.h//! Implements gateway_type::try_put for an external activity to submit a message to FG bool try_put_impl(const Output i) { multifunction_outputOutput port_0 output_port0(*this); broadcast_cacheoutput_type port_successors port_0.successors(); fgt_async_try_put_begin(this, port_0); graph_task_list tasks; bool is_at_least_one_put_successful port_successors.gather_successful_try_puts(i, tasks); __TBB_ASSERT( is_at_least_one_put_successful || tasks.empty(), Return status is inconsistent with the method operation. ); while( !tasks.empty() ) { enqueue_in_graph_arena(this-my_graph, tasks.pop_front()); } fgt_async_try_put_end(this, port_0); return is_at_least_one_put_successful; }从源码结构看其流程是取第 0 号输出端口及其后继缓存gather_successful_try_puts(i, tasks)把消息推给所有后继并收集由此派生的待执行任务派生任务被逐个提交回图所属的 arenaenqueue_in_graph_arena保证下游 body 的调度发生在正确的调度域内返回至少有一个后继接受的结果与receiver_gateway::try_put的契约一致。这也解释了规格中register_successor/remove_successor在实现里被断言为不可直接调用的原因——async_node的后继注册只能经由其输出端口完成flow_graph.h。body 的复制语义与 copy_body规格文档特别强调传给async_node的 body 对象会被拷贝。节点持有的是 body 的副本因此节点内部 body 成员变量的更新不会反映到构造时使用的原始对象。若需要在节点外部检查 body 内部状态应使用 copy_body 函数 获取一份更新后的拷贝。实现层提供了配套机制copy_function_objectflow_graph.htemplatetypename Body Body copy_function_object() { typedef multifunction_bodyinput_type, typename base_type::output_ports_type mfn_body_type; typedef async_bodyInput, typename base_type::output_ports_type, gateway_type, Body async_body_type; mfn_body_type body_ref *this-my_body; async_body_type ab *static_castasync_body_type*(dynamic_cast multifunction_body_leafinput_type, output_ports_type, async_body_type (body_ref).get_body_ptr()); return ab.get_body(); }它从节点内部解包出async_body并返回get_body()的结果即当前节点内真实在用的 body 副本——这正是拷贝语义下仍能取到最新状态的实现基础。同时async_body的operator()flow_graph.h直接以tbb::detail::invoke(my_body, v, *this-my_gateway)调用用户 body 并传入 gateway 实参与 AsyncNodeBody 要求 的签名operator()(const Input v, GatewayType gateway)完全对应其noexcept规格也是由noexcept(tbb::detail::invoke(...))透传的说明 body 抛异常的行为约束会直接传导到流图的异常安全边界。与 TBB 流图生态的关系在 flow_graph 规格目录 中async_node与 input_node、overwrite_node、func_node、multifunc_node 等节点并列是流图中唯一一类显式面向图外世界的桥接节点input_node面向图内手动注入消息而async_node面向由外部并发活动驱动的消息注入与提交通过reserve_wait/release_wait的一对语义graph::wait_for_completion一类的等待操作能够覆盖图外工作的生命周期借助策略Policy与并发上限可以在同一接口形态下表达排队缓冲与背压拒绝两种流控行为。小结async_node是 oneTBB 流图中连接图内消息流与图外活动的核心桥接节点它以Input/Output/Policy三模板参数定义了输入类型、输出类型与流控策略默认策略为queueing_lightweightbody 满足 AsyncNodeBody 要求并在收到消息时把消息提交给外部活动gateway()返回满足 GatewayType 要求的接口供外部活动广播结果并配对地通知等待计数节点自身具有 discarding 与 broadcast-push 属性try_put按并发上限与策略接受、排队或拒绝输入而try_get恒返回false。理解上述规格与 flow_graph.h 实现的对应关系是正确构建流图 外部异步活动混合并发系统的关键。【免费下载链接】moldmold: A Modern Linker 项目地址: https://gitcode.com/GitHub_Trending/mo/mold创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考