mold 第三方组件 oneTBB：flow_graph 中 async_node 节点参考与实现解析

mold 第三方组件 oneTBB：flow_graph 中 async_node 节点参考与实现解析

【免费下载链接】moldmold: A Modern Linker 🦠项目地址: https://gitcode.com/GitHub_Trending/mo/mold

本文以 mold 仓库内嵌的 oneTBB（Intel 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，节点在收到消息时调用用户提供的 body；
• body 的职责是把消息提交给图之外的外部活动（例如提交到独立的任务队列、I/O 线程或另一套运行时），而不是在流图内部同步处理；
• 外部活动处理完成后，通过gateway_type接口把结果消息送回流图，广播给该节点的下游后继。

也就是说，async_node同时承担两个方向的角色：对上游而言它是receiver<Input>，对下游而言它是sender<Output>，而 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 receiver<Input>, public sender<Output> { public: template<typename Body> async_node( graph &g, size_t concurrency, Body body, Policy /*unspecified*/ = Policy(), node_priority_t priority = no_priority ); template<typename 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_initializable<Input> && std::copy_constructible<Input>) 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_node+receiver<Input>+sender<Output>三重身份（multifunction_node本身继承自graph_node并实现receiver<Input>的try_put）；
• __TBB_requires(std::default_initializable<Input> && std::copy_constructible<Input>)这一约束表达式与规格文档的 Requirements 一一对应（见下节）。

类型要求（Requirements）

规格文档明确列出三条类型要求，实现中均可找到对应证据：

1. Input类型必须满足 ISO C++ 标准的 DefaultConstructible 与 CopyConstructible 要求——对应源码中的__TBB_requires(...)约束（flow_graph.h）；

2. Policy类型可以指定为 功能节点策略文档 中的轻量/排队/拒绝（lightweight、queueing、rejecting）组合，也可以采用默认值。策略类型别名在 detail/_flow_graph_body_impl.h 中定义：

   typedef Policy<queueing, lightweight> queueing_lightweight; typedef Policy<rejecting, lightweight> rejecting_lightweight;

3. Body类型必须满足 AsyncNodeBody 要求。该要求规定Body必须可复制构造、可析构，并提供void operator()(const Input &v, GatewayType &gateway)成员函数，其中Input与GatewayType必须分别匹配async_node的模板参数和gateway_type成员类型。此外，自 C++17 起，Body还可以是一个指向Input类中接受gateway_type&参数的 const 成员函数的指针——即可以把消息类型本身当作 body 的宿主。源码中以__TBB_requires(async_node_body<Body, 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），拒绝策略则直接拒绝该消息。

成员函数逐一解析

构造函数（无策略重载）

template<typename Body> async_node( graph &g, size_t concurrency, Body body, node_priority_t priority = no_priority );

构造一个调用body副本的async_node，concurrency值限制该节点 body 同时调用的数量。priority参数指定 节点优先级。

实现侧（flow_graph.h）这个重载是委托构造，直接转发到带 Policy 的重载并使用默认策略：

template <typename Body> __TBB_requires(async_node_body<Body, 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) {}

构造函数（带策略重载）

template<typename 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）值得注意的两点：

template<typename Body> __TBB_requires(async_node_body<Body, 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_body<Input, typename base_type::output_ports_type, gateway_type, Body> (body, &my_gateway), a_priority ), my_gateway(self()) { fgt_multioutput_node_with_body<1>( ... ); }

• body 被包装成async_body<Input, Ports, gateway_type, Body>，并在构造时就持有了指向本节点my_gateway的指针——这就是"body 通过 gateway 与流图通信"的底层绑定方式；
• fgt_multioutput_node_with_body<1>(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), sender<Output>(), my_gateway(self()) { static_cast<async_body_base_type*>(this->my_body->get_body_ptr())->set_gateway(&my_gateway); static_cast<async_body_base_type*>(this->my_init_body->get_body_ptr())->set_gateway(&my_gateway); ... }

由于每个async_node实例拥有自己独立的my_gateway，拷贝构造后必须把内部 body（以及保存的初始 body）中的 gateway 指针重定向到新节点自己的 gateway，否则外部活动会把结果错误地送回源节点。

gateway() 与 gateway_type

gateway_type& gateway()

返回gateway_type接口的引用。规格规定gateway_type满足 GatewayType 要求，该接口包含三个操作：

实现上，gateway_type的具体类型是receiver_gateway<Output>（flow_graph.h），节点内部的receiver_gateway_impl完成了三个操作的落地（flow_graph.h）：

class receiver_gateway_impl: public receiver_gateway<Output> { public: receiver_gateway_impl(async_node* node): my_node(node) {} void reserve_wait() override { fgt_async_reserve(static_cast<typename 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_cast<typename 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_put

bool try_put( const input_type& v )

• 若并发上限允许，对传入消息v执行用户提供的 body；
• 否则按节点策略，要么把消息v排队，要么拒绝它。

返回值：输入被接受返回true，否则返回false（例如拒绝策略下并发已满）。

try_get

bool try_get( output_type& v )

返回值：恒为false。这符合 broadcast-push 属性：async_node是纯推送型 sender，不缓冲输出，因此拉取方永远取不到东西；下游若需要缓冲，应使用 queue_node、buffer_node 等带缓冲能力的节点承接。

输出消息的广播路径：try_put_impl

外部活动通过 gateway 提交结果时，最终走到节点的私有try_put_impl（flow_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_output<Output> &port_0 = output_port<0>(*this); broadcast_cache<output_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; }

从源码结构看，其流程是：

1. 取第 0 号输出端口及其后继缓存；
2. gather_successful_try_puts(i, tasks)把消息推给所有后继，并收集由此派生的待执行任务；
3. 派生任务被逐个提交回图所属的 arena（enqueue_in_graph_arena），保证下游 body 的调度发生在正确的调度域内；
4. 返回"至少有一个后继接受"的结果，与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_object（flow_graph.h）：

template<typename Body> Body copy_function_object() { typedef multifunction_body<input_type, typename base_type::output_ports_type> mfn_body_type; typedef async_body<Input, 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_cast<async_body_type*>(dynamic_cast< multifunction_body_leaf<input_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_lightweight；body 满足 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