qidao123.com技术社区-IT企服评测·应用市场

标题: 【中心件】brpc之工作窃取队列 [打印本页]

---

作者: 欢乐狗    时间: 前天 14:33
标题: 【中心件】brpc之工作窃取队列
BRPC Work Stealing Queue

1 核心功能

无锁工作窃取队列（Lock-Free Work Stealing Queue），用于在 BRPC 的 bthread 用户态线程库中支持高效的任务调理。
设计目的：

• 高并发性能：允许线程本地快速存取任务，淘汰锁竞争。
  
• 负载均衡：空闲线程可“窃取”其他线程队列中的任务，提升 CPU 利用率。
  

---

2 关键数据布局

2.1 队列布局

1. template <typename T>
2. class WorkStealingQueue {
3. private:
4.     std::atomic<size_t> _bottom;  // 本地线程操作的队尾
5.     std::atomic<size_t> _top;     // 其他线程窃取的队头
6.     T* _array;                    // 环形缓冲区
7.     size_t _capacity;             // 队列容量
8. };

复制代码

• 环形缓冲区：通过 _array 存储任务，容量固定（通常为 2 的幂次，便于位运算优化）。
  
• 原子指针：_bottom 和 _top 使用原子操作包管线程安全。
  

2.2 内存布局优化

• 缓存行对齐：_bottom 和 _top 大概被放置在差别缓存行，避免伪共享（False Sharing）。
  1. alignas(64) std::atomic<size_t> _bottom;
  2. alignas(64) std::atomic<size_t> _top;

  复制代码

---

3 核心操作

3.1 本地线程操作（非线程安全）

• push(T item)：向队尾插入任务。
  1. bool push(T item) {
  2.     size_t b = _bottom.load(std::memory_order_relaxed);
  3.     size_t t = _top.load(std::memory_order_acquire);
  4.     if (b - t >= _capacity) return false; // 队列满
  5.     _array[b % _capacity] = item;
  6.     _bottom.store(b + 1, std::memory_order_release);
  7.     return true;
  8. }

  复制代码
• pop(T* item)：从队尾取出任务（本地线程独占）。
  1. bool pop(T* item) {
  2.     size_t b = _bottom.load(std::memory_order_relaxed) - 1;
  3.     _bottom.store(b, std::memory_order_relaxed);
  4.     std::atomic_thread_fence(std::memory_order_seq_cst);
  5.     size_t t = _top.load(std::memory_order_relaxed);
  6.     if (t <= b) { // 队列非空
  7.         *item = _array[b % _capacity];
  8.         if (t != b) return true; // 队列仍有元素
  9.         // 最后一个元素，检查竞争
  10.         if (_top.compare_exchange_strong(t, t + 1,
  11.                 std::memory_order_seq_cst,
  12.                 std::memory_order_relaxed)) {
  13.             _bottom.store(b + 1, std::memory_order_relaxed);
  14.             return true;
  15.         }
  16.         _bottom.store(b + 1, std::memory_order_relaxed);
  17.         return false;
  18.     }
  19.     _bottom.store(b + 1, std::memory_order_relaxed);
  20.     return false;
  21. }

  复制代码

3.2 窃取操作（线程安全）

• steal(T* item)：从队头窃取任务（多线程并发安全）。
  1. bool steal(T* item) {
  2.     size_t t = _top.load(std::memory_order_acquire);
  3.     std::atomic_thread_fence(std::memory_order_seq_cst);
  4.     size_t b = _bottom.load(std::memory_order_acquire);
  5.     if (t >= b) return false; // 队列空
  6.     *item = _array[t % _capacity];
  7.     if (!_top.compare_exchange_strong(t, t + 1,
  8.             std::memory_order_seq_cst,
  9.             std::memory_order_relaxed)) {
  10.         return false; // 竞争失败
  11.     }
  12.     return true;
  13. }

  复制代码

---

4 设计亮点

4.1 无锁原子操作

• compare_exchange_strong：用于更新 _top 指针，确保窃取操作的原子性。
  
• 内存序控制：
  
  
  
  • push 使用 release 包管写入可见性。
    
  • steal 使用 acquire 和 seq_cst 确保读取次序。
    
  
  

4.2 环形缓冲区优化

• 位运算代替取模：若容量为 2 的幂次，b % _capacity 可简化为 b & (_capacity - 1)，提升性能。
  
• 快速失败查抄：队列满/空时直接返回，避免无效操作。
  

4.3 线程角色分离

• 本地线程（Owner）：仅操作 _bottom，无需原子同步（relaxed 内存序）。
  
• 窃取线程（Thief）：操作 _top，需严格同步（acquire/release）。
  

---

5 性能优化

• 低竞争场景高效：本地线程的 push/pop 几乎无原子开销。
  
• 批量窃取：可扩展为一次窃取多个任务，淘汰竞争频率。
  
• 动态扩容：当前实现容量固定，实际工程中可结合 std::vector 实现动态扩容。
  

---

6 应用场景

• bthread 调理器：每个 Worker 线程维护一个任务队列，空闲线程窃取任务。
  
• 并行计算：如 Fork-Join 模型中的任务分发。
  
• 事件驱动框架：处置惩罚异步 I/O 任务的负载均衡。
  

---

7 潜在问题与改进

7.1 ABA 问题

• 风险：若 _top 指针被多次修改后回到原值，compare_exchange_strong 大概错误成功。
  
• 解决方案：使用带版本号的指针（如 (top << 16) | version）。
  

7.2 容量限定

• 固定容量：队列满时需由调用方处置惩罚（如丢弃任务或动态扩容）。
  
• 改进方案：实现动态扩容（需原子地替换 _array 和 _capacity）。
  

7.3 伪共享

• 缓存行竞争：_bottom 和 _top 需对齐到差别缓存行（代码中大概已通过 alignas 实现）。
  

---

8 代码示例（简化版）

1. template <typename T>
2. class WorkStealingQueue {
3. public:
4.     WorkStealingQueue(size_t capacity) :
5.         _bottom(0), _top(0), _capacity(next_pow2(capacity)) {
6.         _array = new T[_capacity];
7.     }
9.     bool push(T item) {
10.         size_t b = _bottom.load(std::memory_order_relaxed);
11.         if (b - _top.load(std::memory_order_acquire) >= _capacity)
12.             return false;
13.         _array[b & (_capacity - 1)] = item;
14.         _bottom.store(b + 1, std::memory_order_release);
15.         return true;
16.     }
18.     bool steal(T* item) {
19.         size_t t = _top.load(std::memory_order_acquire);
20.         size_t b = _bottom.load(std::memory_order_acquire);
21.         if (t >= b) return false;
22.         *item = _array[t & (_capacity - 1)];
23.         return _top.compare_exchange_strong(t, t + 1,
24.                    std::memory_order_seq_cst,
25.                    std::memory_order_relaxed);
26.     }
27. };

复制代码

---

9 总结

work_stealing_queue.h 是 BRPC 高性能任务调理的核心组件，通过无锁环形缓冲区和原子操作实现高效的任务存取与窃取。其设计充分优化了本地线程操作的性能，同时通过精致的内存序控制包管窃取操作的线程安全。实用于高并发场景下的负载均衡，但需注意容量限定和 ABA 问题的潜在风险。

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。

---

欢迎光临 qidao123.com技术社区-IT企服评测·应用市场 (https://dis.qidao123.com/)

Powered by Discuz! X3.4