mold 项目内置 oneAPI TBB concurrent_queue 并发队列规范详解API 全解、并发安全边界与无锁实现原理【免费下载链接】moldmold: A Modern Linker 项目地址: https://gitcode.com/GitHub_Trending/mo/mold导读本文以 mold 仓库内置的 oneAPI Threading Building BlocksTBB规范文档为主线系统讲解oneapi::tbb::concurrent_queue这一无界 FIFO 并发容器的完整接口类模板定义、构造/复制/移动语义、并发安全与非安全成员函数的分界线、迭代器约束、非成员函数与 C17 推导指引。结合 concurrent_queue.h 与 _concurrent_queue_base.h 的源码实现你会理解其ticket 票据 原子计数器的无锁架构并能在多线程生产-消费场景中正确、安全地使用该容器避开unsafe_*接口带来的未定义行为陷阱。说明moldREADME.md是一个以速度为设计目标的高性能链接器其第三方依赖目录third-party/tbb中捆绑了完整的 TBB 库及其规范文档third-party/tbb/doc/main/specification/source/containers/本文即基于其中concurrent_queue_cls.rst及同名子目录下的 7 个分节文档整理而成实现事实均可在 concurrent_queue.h 与 _concurrent_queue_base.h 中验证。一、concurrent_queue 是什么无界 FIFO 并发队列1.1 核心语义oneapi::tbb::concurrent_queue是一个无界unbounded先进先出FIFO数据结构允许多个线程同时执行 push 与 pop而无需调用方加锁。其头文件与命名空间为// Defined in header oneapi/tbb/concurrent_queue.h namespace oneapi { namespace tbb { ... } }在仓库中实际头文件位于 concurrent_queue.h并在 concurrent_priority_queue.h 等兄弟容器中共享底层的 _concurrent_queue_base.h 实现。TBB 为兼容历史版本还在 tbb/concurrent_queue.h 提供旧命名空间入口。1.2 类模板 Synopsis规范文档给出了完整类模板定义这是理解全部 API 的总纲// Defined in header oneapi/tbb/concurrent_queue.h namespace oneapi { namespace tbb { template typename T, typename Allocator cache_aligned_allocatorT class concurrent_queue { public: using value_type T; using reference T; using const_reference const T; using pointer typename std::allocator_traitsAllocator::pointer; using const_pointer typename std::allocator_traitsAllocator::const_pointer; using allocator_type Allocator; using size_type implementation-defined unsigned integer type; using difference-type implementation-defined signed integer type; using iterator implementation-defined ForwardIterator; using const_iterator implementation-defined constant ForwardIterator; // Construction, destruction, copying concurrent_queue(); explicit concurrent_queue( const allocator_type alloc ); template typename InputIterator concurrent_queue( InputIterator first, InputIterator last, const allocator_type alloc allocator_type() ); concurrent_queue( std::initializer_listvalue_type init, const allocator_type alloc allocator_type() ); concurrent_queue( const concurrent_queue other ); concurrent_queue( const concurrent_queue other, const allocator_type alloc ); concurrent_queue( concurrent_queue other ); concurrent_queue( concurrent_queue other, const allocator_type alloc ); ~concurrent_queue(); concurrent_queue operator( const concurrent_queue other ); concurrent_queue operator( concurrent_queue other ); concurrent_queue operator( std::initializer_listvalue_type init ); template typename InputIterator void assign( InputIterator first, InputIterator last ); void assign( std::initializer_listvalue_type init ); void swap( concurrent_queue other ); void push( const value_type value ); void push( value_type value ); template typename... Args void emplace( Args... args ); bool try_pop( value_type result ); allocator_type get_allocator() const; size_type unsafe_size() const; bool empty() const; void clear(); iterator unsafe_begin(); const_iterator unsafe_begin() const; const_iterator unsafe_cbegin() const; iterator unsafe_end(); const_iterator unsafe_end() const; const_iterator unsafe_cend() const; }; // class concurrent_queue } // namespace tbb } // namespace oneapi对照源码concurrent_queue.h实际实现中size_type即std::size_t、difference_type即std::ptrdiff_titerator为concurrent_queue_iteratorconcurrent_queue, T, Allocatorconst_iterator为其const T特化——规范中的implementation-defined在此仓库中有明确落点。1.3 类型要求Requirements规范明确了两条硬性要求类型T必须满足 ISO C 标准 [container.requirements] 中的Erasable要求不同成员函数可能按操作类型施加更严格的要求如push(const T)要求CopyInsertablepush(T)要求MoveInsertableemplace要求EmplaceConstructibletry_pop要求MoveAssignable。类型Allocator必须满足 [allocator.requirements] 中的Allocator要求。默认分配器为cache_aligned_allocatorT即 TBB 的缓存行对齐分配器——队列内部结构head/tail 计数器、页数组按max_nfs_size非共享缓存行大小对齐以避免伪共享false sharing这一点在 concurrent_queue.h 的__TBB_ASSERT(is_aligned(...))断言中可以直接看到。二、构造、析构与复制语义本部分对应分节文档 construct_destroy_copy.rst。2.1 空容器构造concurrent_queue(); explicit concurrent_queue( const allocator_type alloc );构造一个空concurrent_queue若提供alloc则使用它分配内存。源码中默认构造委托给concurrent_queue(allocator_type())随后通过r1::cache_aligned_allocate分配queue_representation_type表示对象并构造concurrent_queue.h。2.2 从元素序列构造template typename InputIterator concurrent_queue( InputIterator first, InputIterator last, const allocator_type alloc allocator_type() );构造包含半开区间[first, last)中全部元素的队列。要求InputIterator必须满足 [input.iterators] 的InputIterator要求。源码实现即逐元素pushtemplate typename InputIterator concurrent_queue(InputIterator begin, InputIterator end, const allocator_type a allocator_type()) : concurrent_queue(a) { for (; begin ! end; begin) push(*begin); }concurrent_queue( std::initializer_listvalue_type init, const allocator_type alloc allocator_type() );等价于concurrent_queue(init.begin(), init.end(), alloc)。2.3 复制构造concurrent_queue( const concurrent_queue other ); concurrent_queue( const concurrent_queue other, const allocator_type alloc );构造other的副本。若未提供分配器参数则通过std::allocator_traitsallocator_type::select_on_container_copy_construction(other.get_allocator())获取。注意与other并发操作时行为未定义UB。源码中复制构造最终调用my_queue_representation-assign(*src.my_queue_representation, my_allocator, copy_construct_item)即对队列表示按页深拷贝concurrent_queue.h。2.4 移动构造concurrent_queue( concurrent_queue other ); concurrent_queue( concurrent_queue other, const allocator_type alloc );以移动语义构造other被留在有效但未指定的状态未提供分配器时由std::move(other.get_allocator())取得。与other并发操作时行为未定义。源码揭示了移动构造的两种路径concurrent_queue.h无分配器版本直接internal_swap(src)O(1) 交换内部表示带分配器版本若my_allocator src.my_allocator同样走internal_swap否则由于一个分配器实例分配的内存不能由另一个实例释放退化为逐元素移动move_construct_item并src.clear()。2.5 析构~concurrent_queue();销毁队列调用存储元素的析构函数并释放存储。与*this并发操作时行为未定义。源码中析构依次执行clear()、清空队列表示、销毁并释放表示对象concurrent_queue.h。2.6 赋值运算符与 assignconcurrent_queue operator( const concurrent_queue other ); // 复制赋值 concurrent_queue operator( concurrent_queue other ); // 移动赋值 concurrent_queue operator( std::initializer_listvalue_type init ); // 列表赋值 template typename InputIterator void assign( InputIterator first, InputIterator last ); void assign( std::initializer_listvalue_type init );语义要点复制赋值用other中元素副本替换*this中全部元素若std::allocator_traitsallocator_type::propagate_on_container_copy_assignment::value为true则复制赋值分配器与*this或other并发操作时为 UB。源码中复制赋值会先clear()再整体assign(...)且当前实现中分配器传播以TODO注释标注concurrent_queue.h。移动赋值以移动语义替换other留在有效但未指定状态按propagate_on_container_move_assignment决定是否移动赋值分配器并发操作为 UB。列表赋值等价于用init的元素替换全部元素。assign(first, last)等价于assign(init.begin(), init.end())要求InputIterator满足InputIterator要求并发操作为 UB。三、并发安全成员函数Concurrently Safe Member Functions本部分对应分节文档 safe_member_functions.rst。规范给出的核心准则本节所有成员函数可以彼此并发执行。即任意数量的线程可以同时调用 push/emplace/try_pop/get_allocator而无须外部锁。3.1 入队push 与 emplacevoid push( const value_type value ); // 要求 T 满足 CopyInsertable void push( value_type value ); // 要求 T 满足 MoveInsertablevalue 被留在有效但未指定状态 template typename... Args void emplace( Args... args ); // 要求 T 满足 EmplaceConstructible原地构造新元素三个接口均为并发安全。emplace的优势是避免临时对象拷贝直接在页槽位上以args构造元素。3.2 出队try_popbool try_pop( value_type value );非阻塞弹出语义若容器为空什么都不做否则取出容器中最后一个元素即队列尾部、最旧的元素赋值给value弹出的元素被销毁要求T满足MoveAssignable返回true表示成功弹出false表示队列为空。注意try_pop是非阻塞的队列为空立即返回false不会自旋等待——这与底层pop实现的spin_wait_until_eq / spin_wait_while_eq等待逻辑配合internal_try_pop_impl的外层判定共同保证了空即返回的语义详见第五节。3.3 get_allocatorallocator_type get_allocator() const;返回与*this关联的分配器的副本。四、并发不安全成员函数Concurrently Unsafe Member Functions本部分对应分节文档 unsafe_member_functions.rst 与 iterators.rst。本节所有成员函数只能串行执行若与任何其他包括并发安全的方法并发执行行为未定义。这是 TBB 并发容器最容易被误用的一条红线unsafe_前缀不是可选优化而是串行专用的警告。4.1 元素个数与判空size_type unsafe_size() const; // 返回容器中元素个数 bool empty() const; // 容器为空返回 true否则 falseunsafe_size()本身 O(1) 读取计数但在并发读写下不可靠读取瞬间的计数可能立即过期因此被明确归入 unsafe 组。empty()同理即使名字没有unsafe_前缀规范仍将其列为并发不安全函数仅适合串行判断。4.2 clearvoid clear();移除容器中全部元素。并发操作如与另一线程的 push/try_pop 同时执行为 UB。注意即使在串行场景clear也需要逐个销毁元素代价与元素个数成正比。4.3 swapvoid swap( concurrent_queue other );交换*this与other的内容。若std::allocator_traitsallocator_type::propagate_on_container_swap::value为true则交换分配器否则若get_allocator() ! other.get_allocator()行为未定义。4.4 迭代器unsafe_begin / unsafe_end 系列iterator unsafe_begin(); const_iterator unsafe_begin() const; const_iterator unsafe_cbegin() const; iterator unsafe_end(); const_iterator unsafe_end() const; const_iterator unsafe_cend() const;concurrent_queue::iterator与const_iterator满足 [forward.iterators] 的ForwardIterator要求unsafe_begin()/unsafe_cbegin()返回指向首个元素的迭代器unsafe_end()/unsafe_cend()返回指向末尾后一位置的迭代器迭代器相关操作遍历、解引用同样只能串行执行与并发安全方法并发执行时行为未定义。正因为 FIFO 队列中元素会随时被弹出销毁迭代器在并发下必然失效所以 TBB 刻意把迭代器全部标记为unsafe_*。五、非成员函数swap 与二元比较本部分对应分节文档 non_member_swap.rst 与 non_member_binary_comparisons.rst。template typename T, typename Allocator void swap( concurrent_queueT, Allocator lhs, concurrent_queueT, Allocator rhs ); // 等价于 lhs.swap(rhs) template typename T, typename Allocator bool operator( const concurrent_queueT, Allocator lhs, const concurrent_queueT, Allocator rhs ); template typename T, typename Allocator bool operator!( const concurrent_queueT, Allocator lhs, const concurrent_queueT, Allocator rhs );语义细节非成员swap等价于lhs.swap(rhs)operator检查lhs与rhs是否相等即元素个数相同且lhs包含rhs的全部元素顺序一致operator!等价于!(lhs rhs)规范特别说明这些非成员函数定义的确切命名空间未指定只要能在对应操作中被使用即可。例如实现可以将类与函数定义在同一个内部命名空间中并把oneapi::tbb::concurrent_queue定义为类型别名使这些非成员函数**仅能通过实参依赖查找ADL**被找到。这也提醒使用者不要依赖显式的限定名去调用它们依赖 ADL 即可。六、C17 推导指引Deduction Guides本部分对应分节文档 deduction_guides.rst。自 C17 起concurrent_queue的构造器支持类模板实参推导CTAD。复制/移动构造器含带显式allocator_type参数的版本提供隐式生成的推导指引此外规范还显式提供以下指引template typename InputIterator, typename Allocator tbb::cache_aligned_allocatoriterator_value_tInputIterator concurrent_queue( InputIterator, InputIterator, Allocator Allocator() ) - concurrent_queueiterator_value_tInputIterator, Allocator; template typename InputIterator using iterator_value_t typename std::iterator_traitsInputIterator::value_type;该指引参与重载决议需同时满足InputIterator满足 [input.iterators] 的InputIterator要求Allocator满足 [allocator.requirements] 的Allocator要求。规范附带的完整示例#include oneapi/tbb/concurrent_queue.h #include vector #include memory int main() { std::vectorint vec; // Deduces cq1 as oneapi::tbb::concurrent_queueint oneapi::tbb::concurrent_queue cq1(vec.begin(), vec.end()); // Deduces cq2 as oneapi::tbb::concurrent_queueint, std::allocatorint oneapi::tbb::concurrent_queue cq2(vec.begin(), vec.end(), std::allocatorint{}) }从示例可以看到 CTAD 的实用价值cq1自动推导为concurrent_queueint默认cache_aligned_allocatorintcq2自动推导为concurrent_queueint, std::allocatorint无需手写模板实参。七、源码级原理ticket 票据机制与无锁实现规范文档描述的是是什么而 _concurrent_queue_base.h 与 concurrent_queue.h 揭示了如何做到。7.1 internal_try_pop_impl读序与 CAS 重试template typename QueueRep, typename Allocator std::pairbool, ticket_type internal_try_pop_impl(void* dst, QueueRep queue, Allocator alloc ) { ticket_type ticket{}; do { // 需要在读 tail_counter 之前读 head_counter从而在 head_counter 上建立 happens-before ticket queue.head_counter.load(std::memory_order_acquire); do { if (static_caststd::ptrdiff_t(queue.tail_counter.load(std::memory_order_relaxed) - ticket) 0) { // 队列为空 return { false, ticket }; } // 查看时队列中还有 ticket 为 k 的元素尝试取走它 // 若被其他线程抢先取走则重试 } while (!queue.head_counter.compare_exchange_strong(ticket, ticket 1)); } while (!queue.choose(ticket).pop(dst, ticket, queue, alloc)); return { true, ticket }; }关键点先读 head_counteracquire再读 tail_counterrelaxed通过原子内存序在计数器上建立 happens-before避免读到虚高的队尾计数用CAScompare_exchange_strong抢占下一个 head ticket若竞争失败则重试内层循环外层循环保证即使成功抢占 ticket但底层pop需要等待元素真正就位失败也会重新取票重试队列为空时立即返回{false, ticket}这正是try_pop非阻塞语义的实现根基。7.2 push票据递增与异常安全templatetypename... Args void push( ticket_type k, queue_rep_type base, queue_allocator_type allocator, Args... args ) { padded_page* p nullptr; page_allocator_type page_allocator(allocator); size_type index prepare_page(k, base, page_allocator, p); __TBB_ASSERT(p ! nullptr, Page was not prepared); // 用 RAII 守卫保证异常安全构造元素若抛异常则把该票标记为无效并推进 tail_counter auto value_guard make_raii_guard([] { base.n_invalid_entries; d1::call_itt_notify(d1::releasing, tail_counter); tail_counter.fetch_add(queue_rep_type::n_queue); }); page_allocator_traits::construct(page_allocator, (*p)[index], std::forwardArgs(args)...); // 元素构造成功置位页内 mask 位标记元素已就位 p-mask.store(p-mask.load(std::memory_order_relaxed) | uintptr_t(1) index, std::memory_order_relaxed); d1::call_itt_notify(d1::releasing, tail_counter); value_guard.dismiss(); tail_counter.fetch_add(queue_rep_type::n_queue); }实现要点每个 push 对应一个递增的ticket票据tail_counter.fetch_add(n_queue)发布该票已入队队列按page页组织内存prepare_page负责为新页分配缓存行对齐的padded_page并串起页链表元素构造成功后才置位 mask 位表示此槽位元素就绪随后推进tail_counter——这样消费者pop中通过p-mask判断槽位有效性时不会读到半构造的元素RAII 守卫保证若元素构造抛出异常该 ticket 被标记为无效n_invalid_entries并仍推进tail_counter避免队列卡死在未完成入队状态。7.3 pop自旋等待与页回收bool pop( void* dst, ticket_type k, queue_rep_type base, queue_allocator_type allocator ) { k -queue_rep_type::n_queue; spin_wait_until_eq(head_counter, k); d1::call_itt_notify(d1::acquired, head_counter); spin_wait_while_eq(tail_counter, k); d1::call_itt_notify(d1::acquired, tail_counter); padded_page *p head_page.load(std::memory_order_relaxed); __TBB_ASSERT( p, nullptr ); size_type index modulo_power_of_two( k/queue_rep_type::n_queue, items_per_page ); bool success false; { // finalizer 负责在必要时释放整页当 index 为页内最后一个槽位时 micro_queue_pop_finalizer... finalizer(*this, page_allocator, k queue_rep_type::n_queue, index items_per_page - 1 ? p : nullptr ); if (p-mask.load(std::memory_order_relaxed) (std::uintptr_t(1) index)) { success true; assign_and_destroy_item(dst, *p, index); } else { --base.n_invalid_entries; } } return success; }实现要点先自旋等待 head_counter 到达自己的 ticket再自旋等待 tail_counter 越过该 ticket保证元素已就位随后才真正取元素——这正是票号顺序保证 FIFO的体现通过mask位判断槽位是否有效处理入队失败被标记为无效票的情形页内最后一个槽位被消费时通过micro_queue_pop_finalizer回收整页内存实现无界队列的按需分配与释放。7.4 宏观架构小结从源码结构可以归纳出concurrent_queue的无锁设计骨架ticket 序head/tail 两个原子计数器维护全局单调递增票据消费者按票号顺序取元素生产者按票号顺序放元素天然保证 FIFO页式存储元素存于按需分配的padded_page链表中页内槽位用mask位图标记就绪状态避免半构造元素被消费者读到自旋等待 CAS消费者通过 CAS 抢占票据并在必要时自旋等待生产者推进计数全程无锁无互斥量只在页分配/回收等内存管理点使用spin_mutex保护页链表异常安全RAII 守卫 n_invalid_entries无效票计数确保构造异常不会破坏队列进度内存序head 用 acquire、tail 用 relaxed/释放在计数器上构建 happens-before兼顾正确性与性能。这一设计正是规范中多线程可同时 push/pop这一核心承诺的底层支撑。八、典型使用模式与注意事项8.1 生产-消费基本用法推荐模式#include oneapi/tbb/concurrent_queue.h oneapi::tbb::concurrent_queueint q; // 生产者线程并发 push / emplace q.push(42); q.emplace(43); // 消费者线程非阻塞 try_pop空队列立即返回 false int v; while (q.try_pop(v)) { // 处理 v }要点永远不要在try_pop返回false后假设队列永远空了——生产者在任意时刻都可能再次 push如需阻塞等待应在应用层配合条件变量或 TBB 流图flow graph等机制需要队列为空且不再有生产者才能退出的场景应通过独立的停止标志/毒丸元素协调而不是依赖empty()它属于并发不安全函数。8.2 并发安全与不安全接口对照表类别成员函数可否并发调用并发安全push(const T)/push(T)/emplace(...)可与其他安全函数任意并发并发安全try_pop(value)可与其他安全函数任意并发并发安全get_allocator()可与其他安全函数任意并发并发不安全unsafe_size()/empty()/clear()/swap()仅串行并发执行即 UB并发不安全unsafe_begin/end/cbegin/cend仅串行并发执行即 UB并发不安全全部构造/析构/赋值/assign仅串行与被操作对象并发即 UB这张表直接对应规范中 safe_member_functions.rst 与 unsafe_member_functions.rst 的划分是排查并发 bug 的第一张检查清单。8.3 易错点清单不要对unsafe_size()/empty()抱并发期望即使名字没有unsafe_前缀empty()也是并发不安全函数见规范 unsafe_member_functions.rst。不要在并发期间遍历迭代器全部为unsafe_*并发遍历行为未定义。复制/移动构造或赋值时源/目标对象不得有并发操作规范 construct_destroy_copy.rst 明确标注 UB。移动构造带分配器版本的行为依赖分配器比较分配器不同时退化为逐元素移动这是 O(n) 而非 O(1)concurrent_queue.h。try_pop要求T可移动赋值元素类型需要满足MoveAssignable否则编译期即不满足要求。依赖 ADL 使用非成员swap/比较运算符这些函数所在命名空间未指定显式限定调用不可移植。九、结语oneapi::tbb::concurrent_queue是一个接口完整、语义清晰的无界 FIFO 并发容器push/emplace/try_pop/get_allocator属于并发安全面可在多线程中自由调用unsafe_size/empty/clear/swap/迭代器/构造复制面则严格限定串行非成员swap与operator/!提供容器级比较C17 推导指引让concurrent_queueint q(vec.begin(), vec.end())这样的写法开箱即用。其底层以 ticket 票据 原子计数器 页式存储实现无锁 FIFO并通过内存序、mask 位图与 RAII 异常安全机制把多线程同时 push/pop从规范承诺落实为可验证的实现事实。相关规范原文位于 concurrent_queue_cls.rst实现可进一步阅读 concurrent_queue.h 与 _concurrent_queue_base.h。【免费下载链接】moldmold: A Modern Linker 项目地址: https://gitcode.com/GitHub_Trending/mo/mold创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
