mold 项目内置 oneAPI TBB concurrent_queue 并发队列规范详解：API 全解、并发安全边界与无锁实现原理

mold 项目内置 oneAPI TBB concurrent_queue 并发队列规范详解：API 全解、并发安全边界与无锁实现原理

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

导读

本文以 mold 仓库内置的 oneAPI Threading Building Blocks（TBB）规范文档为主线，系统讲解oneapi::tbb::concurrent_queue这一无界 FIFO 并发容器的完整接口：类模板定义、构造/复制/移动语义、并发安全与非安全成员函数的分界线、迭代器约束、非成员函数与 C++17 推导指引。结合 concurrent_queue.h 与 _concurrent_queue_base.h 的源码实现，你会理解其"ticket 票据 + 原子计数器"的无锁架构，并能在多线程生产-消费场景中正确、安全地使用该容器，避开"unsafe_*"接口带来的未定义行为陷阱。

说明：mold（README.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_allocator<T>> class concurrent_queue { public: using value_type = T; using reference = T&; using const_reference = const T&; using pointer = typename std::allocator_traits<Allocator>::pointer; using const_pointer = typename std::allocator_traits<Allocator>::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_list<value_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_list<value_type> init ); template <typename InputIterator> void assign( InputIterator first, InputIterator last ); void assign( std::initializer_list<value_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_t，iterator为concurrent_queue_iterator<concurrent_queue, T, Allocator>，const_iterator为其const T特化——规范中的"implementation-defined"在此仓库中有明确落点。

1.3 类型要求（Requirements）

规范明确了两条硬性要求：

• 类型T必须满足 ISO C++ 标准 [container.requirements] 中的Erasable要求；不同成员函数可能按操作类型施加更严格的要求（如push(const T&)要求CopyInsertable，push(T&&)要求MoveInsertable，emplace要求EmplaceConstructible，try_pop要求MoveAssignable）。
• 类型Allocator必须满足 [allocator.requirements] 中的Allocator要求。

默认分配器为cache_aligned_allocator<T>，即 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要求。源码实现即逐元素push：

template <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_list<value_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_traits<allocator_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 赋值运算符与 assign

concurrent_queue& operator=( const concurrent_queue& other ); // 复制赋值 concurrent_queue& operator=( concurrent_queue&& other ); // 移动赋值 concurrent_queue& operator=( std::initializer_list<value_type> init ); // 列表赋值 template <typename InputIterator> void assign( InputIterator first, InputIterator last ); void assign( std::initializer_list<value_type> init );

语义要点：

• 复制赋值：用other中元素副本替换*this中全部元素；若std::allocator_traits<allocator_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 与 emplace

void push( const value_type& value ); // 要求 T 满足 CopyInsertable void push( value_type&& value ); // 要求 T 满足 MoveInsertable；value 被留在有效但未指定状态 template <typename... Args> void emplace( Args&&... args ); // 要求 T 满足 EmplaceConstructible；原地构造新元素

三个接口均为并发安全。emplace的优势是避免临时对象拷贝：直接在页槽位上以args构造元素。

3.2 出队：try_pop

bool 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_allocator

allocator_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，否则 false

unsafe_size()本身 O(1) 读取计数，但在并发读写下不可靠（读取瞬间的计数可能立即过期），因此被明确归入 unsafe 组。empty()同理：即使名字没有unsafe_前缀，规范仍将其列为并发不安全函数，仅适合串行判断。

4.2 clear

void clear();

移除容器中全部元素。并发操作（如与另一线程的 push/try_pop 同时执行）为 UB。注意即使在串行场景，clear也需要逐个销毁元素，代价与元素个数成正比。

4.3 swap

void swap( concurrent_queue& other );

交换*this与other的内容。若std::allocator_traits<allocator_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_queue<T, Allocator>& lhs, concurrent_queue<T, Allocator>& rhs ); // 等价于 lhs.swap(rhs) template <typename T, typename Allocator> bool operator==( const concurrent_queue<T, Allocator>& lhs, const concurrent_queue<T, Allocator>& rhs ); template <typename T, typename Allocator> bool operator!=( const concurrent_queue<T, Allocator>& lhs, const concurrent_queue<T, Allocator>& rhs );

语义细节：

• 非成员swap等价于lhs.swap(rhs)；
• operator==：检查lhs与rhs是否相等，即元素个数相同且lhs包含rhs的全部元素（顺序一致）；operator!=等价于!(lhs == rhs)；
• 规范特别说明：这些非成员函数定义的确切命名空间未指定，只要能在对应操作中被使用即可。例如实现可以将类与函数定义在同一个内部命名空间中，并把oneapi::tbb::concurrent_queue定义为类型别名，使这些非成员函数**仅能通过实参依赖查找（ADL）**被找到。这也提醒使用者：不要依赖显式的限定名去调用它们，依赖 ADL 即可。

六、C++17 推导指引（Deduction Guides）

本部分对应分节文档 deduction_guides.rst。

自 C++17 起，concurrent_queue的构造器支持类模板实参推导（CTAD）。复制/移动构造器（含带显式allocator_type参数的版本）提供隐式生成的推导指引；此外规范还显式提供以下指引：

template <typename InputIterator, typename Allocator = tbb::cache_aligned_allocator<iterator_value_t<InputIterator>> concurrent_queue( InputIterator, InputIterator, Allocator = Allocator() ) -> concurrent_queue<iterator_value_t<InputIterator>, Allocator>; template <typename InputIterator> using iterator_value_t = typename std::iterator_traits<InputIterator>::value_type;

该指引参与重载决议需同时满足：

• InputIterator满足 [input.iterators] 的InputIterator要求；
• Allocator满足 [allocator.requirements] 的Allocator要求。

规范附带的完整示例：

#include <oneapi/tbb/concurrent_queue.h> #include <vector> #include <memory> int main() { std::vector<int> vec; // Deduces cq1 as oneapi::tbb::concurrent_queue<int> oneapi::tbb::concurrent_queue cq1(vec.begin(), vec.end()); // Deduces cq2 as oneapi::tbb::concurrent_queue<int, std::allocator<int>> oneapi::tbb::concurrent_queue cq2(vec.begin(), vec.end(), std::allocator<int>{}) }

从示例可以看到 CTAD 的实用价值：cq1自动推导为concurrent_queue<int>（默认cache_aligned_allocator<int>），cq2自动推导为concurrent_queue<int, std::allocator<int>>，无需手写模板实参。

七、源码级原理：ticket 票据机制与无锁实现

规范文档描述的是"是什么"，而 _concurrent_queue_base.h 与 concurrent_queue.h 揭示了"如何做到"。

7.1 internal_try_pop_impl：读序与 CAS 重试

template <typename QueueRep, typename Allocator> std::pair<bool, 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_cast<std::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_counter（acquire）再读 tail_counter（relaxed），通过原子内存序在计数器上建立 happens-before，避免读到"虚高"的队尾计数；
• 用CAS（compare_exchange_strong）抢占下一个 head ticket，若竞争失败则重试内层循环；
• 外层循环保证：即使成功抢占 ticket，但底层pop（需要等待元素真正就位）失败，也会重新取票重试；
• 队列为空时立即返回{false, ticket}，这正是try_pop非阻塞语义的实现根基。

7.2 push：票据递增与异常安全

template<typename... 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::forward<Args>(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的无锁设计骨架：

1. ticket 序：head/tail 两个原子计数器维护全局单调递增票据，消费者按票号顺序取元素，生产者按票号顺序放元素，天然保证 FIFO；
2. 页式存储：元素存于按需分配的padded_page链表中，页内槽位用mask位图标记就绪状态，避免半构造元素被消费者读到；
3. 自旋等待 + CAS：消费者通过 CAS 抢占票据并在必要时自旋等待生产者推进计数，全程无锁（无互斥量），只在页分配/回收等内存管理点使用spin_mutex保护页链表；
4. 异常安全：RAII 守卫 +n_invalid_entries无效票计数，确保构造异常不会破坏队列进度；
5. 内存序：head 用 acquire、tail 用 relaxed/释放，在计数器上构建 happens-before，兼顾正确性与性能。

这一设计正是规范中"多线程可同时 push/pop"这一核心承诺的底层支撑。

八、典型使用模式与注意事项

8.1 生产-消费基本用法（推荐模式）

#include <oneapi/tbb/concurrent_queue.h> oneapi::tbb::concurrent_queue<int> 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 并发安全与不安全接口对照表

这张表直接对应规范中 safe_member_functions.rst 与 unsafe_member_functions.rst 的划分，是排查并发 bug 的第一张检查清单。

8.3 易错点清单

1. 不要对unsafe_size()/empty()抱并发期望：即使名字没有unsafe_前缀，empty()也是并发不安全函数（见规范 unsafe_member_functions.rst）。
2. 不要在并发期间遍历：迭代器全部为unsafe_*，并发遍历行为未定义。
3. 复制/移动构造或赋值时，源/目标对象不得有并发操作（规范 construct_destroy_copy.rst 明确标注 UB）。
4. 移动构造带分配器版本的行为依赖分配器比较：分配器不同时退化为逐元素移动，这是 O(n) 而非 O(1)（concurrent_queue.h）。
5. try_pop要求T可移动赋值：元素类型需要满足MoveAssignable，否则编译期即不满足要求。
6. 依赖 ADL 使用非成员swap/比较运算符：这些函数所在命名空间未指定，显式限定调用不可移植。

九、结语

oneapi::tbb::concurrent_queue是一个接口完整、语义清晰的无界 FIFO 并发容器：push/emplace/try_pop/get_allocator属于并发安全面，可在多线程中自由调用；unsafe_size/empty/clear/swap/迭代器/构造复制面则严格限定串行；非成员swap与operator==/!=提供容器级比较；C++17 推导指引让concurrent_queue<int> 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