【中心件】brpc之工作窃取队列

打印 上一主题 下一主题

主题 1953|帖子 1953|积分 5859

马上注册,结交更多好友,享用更多功能,让你轻松玩转社区。

您需要 登录 才可以下载或查看,没有账号?立即注册

x
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.     }
  8.     bool push(T item) {
  9.         size_t b = _bottom.load(std::memory_order_relaxed);
  10.         if (b - _top.load(std::memory_order_acquire) >= _capacity)
  11.             return false;
  12.         _array[b & (_capacity - 1)] = item;
  13.         _bottom.store(b + 1, std::memory_order_release);
  14.         return true;
  15.     }
  16.     bool steal(T* item) {
  17.         size_t t = _top.load(std::memory_order_acquire);
  18.         size_t b = _bottom.load(std::memory_order_acquire);
  19.         if (t >= b) return false;
  20.         *item = _array[t & (_capacity - 1)];
  21.         return _top.compare_exchange_strong(t, t + 1,
  22.                    std::memory_order_seq_cst,
  23.                    std::memory_order_relaxed);
  24.     }
  25. };
复制代码

9 总结

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

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

使用道具 举报

0 个回复

倒序浏览

快速回复

您需要登录后才可以回帖 登录 or 立即注册

本版积分规则

欢乐狗

论坛元老
这个人很懒什么都没写!
快速回复 返回顶部 返回列表