并发编程高性能计算【免费下载链接】oneTBBoneAPI Threading Building Blocks (oneTBB)项目地址https://gitcode.com/gh_mirrors/on/oneTBB点击查看免费下载本指南围绕 oneTBBoneAPI Threading Building Blocks流图flow graph接口中的 Token令牌机制展开讲解如何借助reserving策略的join_node精确限制图中同时存活的消息数量从而控制资源消耗与并发度。读完本文你将掌握 token-based 系统的完整搭建流程、join_node三种缓冲策略的取舍、令牌回收环路的实现方式以及用multifunction_node在运行期动态增减令牌的高阶用法。为什么需要 Token先看parallel_pipeline的做法oneTBB 的parallel_pipeline算法本身就是基于 token-based 系统的经典案例。它的核心接口见 parallel_pipeline 参考文档要求调用者显式给出并发度上限// Defined in header oneapi/tbb/parallel_pipeline.h namespace oneapi { namespace tbb { void parallel_pipeline( size_t max_number_of_live_tokens, const filtervoid,void filter_chain ); void parallel_pipeline( size_t max_number_of_live_tokens, const filtervoid,void filter_chain, task_group_context context ); } // namespace tbb } // namespace oneapi其中max_number_of_live_tokens即为“同时存活正在被各 filter 处理的条目数量”的阈值参考文档中给出的示例使用parallel_pipeline( /*max_number_of_live_token*/16, ...)。管道不同阶段以流水线方式并行处理条目但任意时刻“在途”条目数被令牌数严格限制从而避免上游阶段无限产出、导致内存被大量中间结果撑爆。流图flow graph接口则没有parallel_pipeline那样内建的 token 支持但文档明确指出join_node可以用来构建一个与之同构的令牌系统。join_node与三种缓冲策略join_node的作用是把多个输入端口收到的消息组装成一个std::tuple并广播给所有后继节点。它的模板签名见 join_node 类参考包含两个模板参数——输出元组类型与缓冲策略templatetypename OutputTuple, graph_buffer_policy JP queueing class join_node;缓冲策略详见 join_node Policies共有三种策略行为queueing每个输入端口维护一个无界 FIFO 队列当所有端口都至少有一条消息时取各队列队首组装成元组并广播。若后继接受则从各端口队列移除队首否则消息保留在队列中tag_matching即key_matchingtag_value的特化对进入各端口的消息应用用户提供的键提取函数按“键”匹配当某个键在每个端口都有匹配消息时取出并组装广播。若后继拒绝元组被暂存等待后续try_get再转发reserving端口不做内部缓冲而是“先预定、后消费”当各端口都被标记为“可能有输入”后join_node尝试从每个端口的上游源**预定reserve**一条消息若所有端口都预定成功才真正取走这些消息并组装成元组广播任一端口预定失败则释放此前所有预定并取消该端口的标记关键点在于reserving的“两阶段”语义它先把消息“预定”下来而不真正消费只有确认所有端口都能凑齐一组输入时才落地消费。正是这种“先确认再取走”的机制让reserving成为构建 token-based 系统的核心构件。这一行为在仓库测试中也有直接印证例如 test_join_node.h 中专门有一条注释“join_node (reserving) does not consume inputs until an item is available at ...”并在测试中对reserving端口的双阶段行为逐项断言test_flow_graph_whitebox.cpp 等白盒测试也大量使用join_nodestd::tupleint,int, tbb::flow::reserving验证该策略的预定/释放/消费路径。核心示例令牌限流的四节点环路下面是本指南核心的完整可编译示例基于 create_token_based_system.rst 的原始代码补齐了头文件、token_t与tuple_t的类型定义。它的目标让input_node源源不断地产生M个大对象但图中任意时刻同时存活的大对象不超过3令牌数 1input_node 自身缓冲个#include iostream #include tuple #include oneapi/tbb/flow_graph.h #include oneapi/tbb/parallel_for.h using namespace oneapi::tbb; using namespace oneapi::tbb::flow; typedef int token_t; // 令牌可以是任意类型 typedef std::tuple big_object*, token_t tuple_t; // join_node 的输出元组 void spin_for( int n ) { /* 模拟工作负载 */ } int main() { graph g; int src_count 0; int M 100; // 源节点要生成的对象总数 int max_objects 3; // 初始令牌数量 // 1) 数据源仅在被拉动时才生成对象 input_node big_object* s( g, - big_object* { if ( src_count M ) { big_object* v new big_object(); src_count; return v; } else { fc.stop(); // 通知 input_node 停止生产 return nullptr; } } ); s.activate(); // input_node 默认不激活需显式启动 // 2) 令牌闸门reserving join_node 把对象与令牌配对 join_node tuple_t, reserving j(g); // 3) 令牌池预填充 3 个令牌 buffer_node token_t b(g); // 4) 消费者处理完一个对象后把令牌还给令牌池 function_node tuple_t, token_t f( g, unlimited, []( const tuple_t t ) - token_t { spin_for(1); std::cout get1(t) \n; // 打印令牌编号 delete get0(t); // 销毁大对象 return get1(t); // 回收令牌 } ); // 组装环路s - j, b - j, j - f, f - b make_edge( s, input_port0(j) ); make_edge( b, input_port1(j) ); make_edge( j, f ); make_edge( f, b ); // 预填充 3 个令牌 b.try_put( 1 ); b.try_put( 2 ); b.try_put( 3 ); g.wait_for_all(); // 等待图中所有工作完成 }环路如何工作把上面的代码按数据流拆开看input_nodebig_object*它内部只缓冲一个条目且其body只会在“被拉动”时执行reservingjoin_node 会调用try_get从上游拉取。当src_count M时返回新对象否则调用fc.stop()结束生产——这正是 input_node 参考文档 描述的flow_control语义。注意input_node创建后默认处于非激活状态必须调用s.activate()才会开始产出。join_nodetuple_t, reserving它是整个系统的“闸门”。只有当端口 0来自input_node和端口 1来自buffer_node都能预定到消息时它才会真正取走两者组装成tuple_t交给function_node。因此input_node只有在“有令牌可用”时才会被要求生成新对象。buffer_nodetoken_t一个无界缓冲见 buffer_node 参考文档这里作为令牌池使用开篇通过 3 次显式try_put预填充了令牌 1、2、3。function_nodetuple_t, token_t并发度设为unlimited收到配对成功的元组后处理对象、delete掉大对象并把令牌原样返回给buffer_node。注意它的输出类型与buffer_node的元素类型一致正是为了形成f - b这条回流边。于是图中形成了一条s - j - f - b - j的令牌回收环路令牌被消费后不会消失而是回到令牌池等待与下一个对象配对。文档明确指出由于这种循环的存在图中最多同时存在 4 个大对象——3 个可能正在function_nodeunlimited并发度下最多同时处理 3 个因为只有 3 个令牌在流通1 个缓冲在input_node中等待与令牌配对。整个M个对象的产生节奏完全由令牌供给驱动无需任何显式节流逻辑。扩展一让令牌本身携带业务语义流图接口并没有为“令牌”定义专门类型——文档强调token_t可以是任何类型包括对象或指向数组的指针。因此令牌不必是int这样的哑类型它完全可以就是计算所需的资源本身。一个很实用的变形直接用大对象自身作为令牌。把上面示例中的token_t换成big_object*让function_node处理完对象后不delete它而是把同一个对象指针作为令牌返还给buffer_node下一次join_node配对的“对象”其实就是这个被回收复用的对象指针。这样既省去了反复new/delete的开销又在图中形成了一个大对象的free list空闲对象链表——对象被循环利用内存分配次数大幅下降。扩展二动态控制令牌数量实现运行期并发调节前面示例用固定次数try_put预填充令牌池但这只是众多初始化方式之一用input_node生成令牌把一个input_node接到buffer_node的输入端由它按需产出令牌。令牌的产生同样受拉取驱动可以做到“用完一批、再生一批”。用multifunction_node替代function_nodemultifunction_node见 参考文档的 body 可以向多个输出端口各投递 0 条或多条消息。把它放在f的位置就可以在每次处理后选择性地回收令牌把令牌投回b维持并发度不变丢弃令牌不投回降低系统允许的并发度追加令牌多投几个令牌进b提高系统允许的并发度。由于流图中的边是动态可改的make_edge/remove_edge令牌的注入与移除可以在图执行期间随时进行从而对整条链路的并发度做运行期动态调节。何时使用 Token 机制token-based 系统之所以灵活核心在于三点均为原文档明确结论令牌类型自由任何类型都可用作令牌令牌可以携带真实资源语义并发度动态可调执行期间可自由注入/移除令牌动态控制系统允许的并发级别源头限流令牌在数据源处与输入配对因此资源消耗的限制作用覆盖整张图而非单个节点——这正是它与 use_limiter_node.rst 中limiter_node方案的差异所在limiter_node作用于其所在的边而 token 配对发生在源端能防止上游在源头就过度生产。相比固定的并发度设置如把function_node的并发度硬编码为 3token 方案把“限流”从“单个节点的内部约束”提升为“整个数据路径上的端到端约束”在内存受限、对象构造昂贵、或需要运行期动态调节吞吐的场景下尤其有价值。完整示例代码可参照 get_started 示例 中流图的基本骨架结合本文的环路结构即可落地为可运行程序join_node的更多构造细节如key_matching特化、input_ports()、try_get语义可继续查阅 join_node 类参考 与 join_node Policies。赞分享并发编程高性能计算【免费下载链接】oneTBBoneAPI Threading Building Blocks (oneTBB)项目地址https://gitcode.com/gh_mirrors/on/oneTBB点击查看免费下载相关推荐ChatGPT Shortcut 浏览器扩展使用指南侧边栏、显示模式与 AltShiftS 快捷键ChatGPT Shortcut 浏览器扩展使用指南侧边栏、显示模式与 AltShiftS 快捷键 导读 本文基于 ChatGPT ShortcutAAI 应用提示工程人工智能前端Symfony Rate Limiter 组件实战指南Token Bucket 限流与令牌消费机制解析Symfony Rate Limiter 组件实战指南Token Bucket 限流与令牌消费机制解析 本篇指南以 Symfony Rate Limiter后端Web框架JWT双令牌认证系统Access Token与Refresh Token的终极安全指南 JWT双令牌认证系统Access Token与Refresh Token的终极安全指南 在现代Web应用开发中 JWT双令牌认证系统 已经成为保护AP后端认证鉴权上一篇Unlock Music音乐解锁神器终极免费解决方案下一篇数据分析原理从统计指标到漏斗与留存模型构建产品决策的数据直觉easy-vibe 数据基础篇创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考