brpc小课堂：有界队列BoundedQueue

分类：菜鸟笔记作者：果冻虾仁

开场白

什么是有界队列？

brpc实现了一个“有界队列”的类模板BoundedQueue。先说一下什么是有界队列。 所谓有界队列表示的就是一个队列其中的容量是有限的（固定的），不能动态扩容的队列。这种听起来没有vector那种自动扩容能力的容器，主要还是全面为了性能考虑的。一般也是用作生产者和消费者模式，当队列容量已满的时候，一般就表示超过了这个队列的最大吞吐能力，故而拒绝加入新的任务。

在实践中有界队列一般是基于ring buffer（环形缓冲区）来实现的。或者说有界队列就是ring buffer的别名也不为过。ring buffer也被称为circular buffer或circular queue。 整体思路就是用一段固定大小的线性连续空间来模拟循环的队列。详细的定义可以阅读维基百科：

https://en.wikipedia.org/wiki/Circular_buffer

图片来自于网络

ring buffer的定义很简单，实现细节上有多种不同的方案。比如存储读指针和写指针，标记下一次pop和push的位置。这种实现的麻烦之处是对于队列的满或者空是有歧义的。因为队列满和队列空的时候读写两个指针都是指向相同位置。要区分具体是满还是空，就要额外的操作。比如总保持队列中有一个元素是不可写的，此时如果读写指针指向同一位置，则缓冲区为空。如果读指针位于写指针的相邻后一个位置，则缓冲区为满。

当然还有其他实现，比如不存储写指针，而是存储队列中写入的元素个数。每次写入的时候，通过读指针加上元素个数（需要取模）来计算出写指针的位置。brpc的实现就是这种方案。

brpc中的有界队列

// A thread-unsafe bounded queue(ring buffer). It can push/pop from both
// sides and is more handy than thread-safe queues in single thread. Use
// boost::lockfree::spsc_queue or boost::lockfree::queue in multi-threaded
// scenarios.

看注释可以发现，BoundedQueue是一个线程不安全的有界队列。所以BoundedQueue不支持多线程间的并发读写，不过也因此BoundedQueue的代码足够简单。简单到给你10分钟到时间就可以看完，基本上是一个“裸”的ring buffer实现。当然也不是说BoundedQueue不能应用在多线程读写的场景，只是说你需要在外围添加额外的线程同步控制逻辑，比如下文我们也会提到的RemoteTaskQueue。

接下来我们来阅读一下他的源码： 代码路径：butil\containers\bounded_queue.h

模板声明与数据成员

先看一下模板声明及其中的数据成员：

template <typename T>
class BoundedQueue {
public:
    ...
private:
    // Since the space is possibly not owned, we disable copying.
    DISALLOW_COPY_AND_ASSIGN(BoundedQueue);
    
    ...
    
    uint32_t _count;  // 当前元素个数
    uint32_t _cap;    // 容量
    uint32_t _start;  // 起始位置
    StorageOwnership _ownership; // 枚举类型，所有权。表示有界队列是否持有该数据
    void* _items;     // 数据指针
};

StorageOwnership的定义：

enum StorageOwnership { OWNS_STORAGE, NOT_OWN_STORAGE };

顾名思义，枚举值OWNS_STORAGE表示有界队列持有数据存储的所有权，NOT_OWN_STORAGE则没有。

构造函数

看一下构造函数，它有三个重载。第一个：

    // You have to pass the memory for storing items at creation.
    // The queue contains at most memsize/sizeof(T) items.
    BoundedQueue(void* mem, size_t memsize, StorageOwnership ownership)
        : _count(0)
        , _cap(memsize / sizeof(T))
        , _start(0)
        , _ownership(ownership)
        , _items(mem) {
        DCHECK(_items);
    };

支持传入数据所有权ownership，数据地址mem以及大小memsize。

    // Construct a queue with the given capacity.
    // The malloc() may fail silently, call initialized() to test validity
    // of the queue.
    explicit BoundedQueue(size_t capacity)
        : _count(0)
        , _cap(capacity)
        , _start(0)
        , _ownership(OWNS_STORAGE)
        , _items(malloc(capacity * sizeof(T))) {
        DCHECK(_items);
    };

只传入容量capacity，这种表示队列持有数据存储的所有权，在构造时用malloc来分配内存。当然鉴于malloc不一定总能成功，故而使用DCHECK来做一把校验。TODO,失败了如何？

push相关函数

push()

    // Push |item| into bottom side of this queue.
    // Returns true on success, false if queue is full.
    bool push(const T& item) {
        if (_count < _cap) {
            new ((T*)_items + _mod(_start + _count, _cap)) T(item);
            ++_count;
            return true;
        }
        return false;
    }

    // Push a default-constructed item into bottom side of this queue
    // Returns address of the item inside this queue
    T* push() {
        if (_count < _cap) {
            return new ((T*)_items + _mod(_start + _count++, _cap)) T();
        }
        return NULL;
    }

如果队列没满，则使用placement new进行构造。常规的new是不需要自己指定对象分配的堆地址的，而placement new则可以在指定的内存位置上构造对象。这里内存位置通过起始位置+元素个数来定位到。

    // This is faster than % in this queue because most |off| are smaller
    // than |cap|. This is probably not true in other place, be careful
    // before you use this trick.
    static uint32_t _mod(uint32_t off, uint32_t cap) {
        while (off >= cap) {
            off -= cap;
        }
        return off;
    }

这个模运算使用了一个比较Trick的技巧，他的限制是off永远能保证在[0, 2*cap)之间。

push_top()

   // Push |item| into top side of this queue
   // Returns true on success, false if queue is full.
   bool push_top(const T& item) {
       if (_count < _cap) {
           _start = _start ? (_start - 1) : (_cap - 1);
           ++_count;
           new ((T*)_items + _start) T(item);
           return true;
       }
       return false;
   }    
   
   // Push a default-constructed item into top side of this queue
   // Returns address of the item inside this queue
   T* push_top() {
       if (_count < _cap) {
           _start = _start ? (_start - 1) : (_cap - 1);
           ++_count;
           return new ((T*)_items + _start) T();
       }
       return NULL;
   }

elim_push()

    // Push |item| into bottom side of this queue. If the queue is full,
   // pop topmost item first.
   void elim_push(const T& item) {
       if (_count < _cap) {
           new ((T*)_items + _mod(_start + _count, _cap)) T(item);
           ++_count;
       } else {
           ((T*)_items)[_start] = item;
           _start = _mod(_start + 1, _cap);
       }
   }

pop相关函数

pop是push的逆操作，这里大家可以选择性的阅读。

pop()

  // Pop top-most item from this queue
  // Returns true on success, false if queue is empty
  bool pop() {
      if (_count) {
          --_count;
          ((T*)_items + _start)->~T();
          _start = _mod(_start + 1, _cap);
          return true;
      }
      return false;
  }

  // Pop top-most item from this queue and copy into |item|.
  // Returns true on success, false if queue is empty
  bool pop(T* item) {
      if (_count) {
          --_count;
          T* const p = (T*)_items + _start;
          *item = *p;
          p->~T();
          _start = _mod(_start + 1, _cap);
          return true;
      }
      return false;
  }

值得一提的是pop() 和 push()是对称的一对操作，但是他们的函数参数却并不相同。在有参版的push()中，其参数是const T&，而pop()的参数是T*。 这其中的差异是我们需要通过pop_bottom()的出参拿到元素。好的C++编码规范（比如谷歌编码规范）都指明，当函数参数做出参的时候用指针，而作为入参的时候使用const &。

pop_bottom()

    // Pop bottom-most item from this queue
    // Returns true on success, false if queue is empty
    bool pop_bottom() {
        if (_count) {
            --_count;
            ((T*)_items + _mod(_start + _count, _cap))->~T();
            return true;
        }
        return false;
    }

    // Pop bottom-most item from this queue and copy into |item|.
    // Returns true on success, false if queue is empty
    bool pop_bottom(T* item) {
        if (_count) {
            --_count;
            T* const p = (T*)_items + _mod(_start + _count, _cap);
            *item = *p;
            p->~T();
            return true;
        }
        return false;
    }

当然push系列函数中的elim_push()在pop中并没有对应的逆操作，这也不难理解。但队列已满的时候，我实在想插入数据可以移除队头元素，但是如果队列为空的时候，我们实在还想从中取出一个元素，那么则是没有什么办法。

清空队列

    void clear() {
        for (uint32_t i = 0; i < _count; ++i) {
            ((T*)_items + _mod(_start + i, _cap))->~T();
        }
        _count = 0;
        _start = 0;
    }

这里主要做的是，遍历所有元素，取显式调用其析构函数：->~T()。这个操作是由于我们的对象是通过placement new构造的，->~T() 就是其必须成对的析构操作。就像new必须有delete一般。但和new与delete不同的是，不管是placement new 还是 ->~T() 都没有对本对象去做堆中内存的分配和释放！（当然如果T中有成员，其实还是会涉及到堆内存分配和释放的，但那是针对其中的成员，并不是T本身）

讲到这，我们来看一下析构函数。

析构函数

    ~BoundedQueue() {
        clear();
        if (_ownership == OWNS_STORAGE) {
            free(_items);
            _items = NULL;
        }
    }

首先调用前面的clear()函数，然后需要对数据存储的所有权做一个判断，只有明确是自己持有的情况下，采取free()释放掉内存，和malloc()成对。而对于非自己持有的，就不必了。

一些视图操作

本节主要是介绍对于有界队列的一些数据与元信息进行观测的函数。

top()

    // Get address of top-most item, NULL if queue is empty
    T* top() { 
        return _count ? ((T*)_items + _start) : NULL; 
    }
    const T* top() const { 
        return _count ? ((const T*)_items + _start) : NULL; 
    }

    // Randomly access item from top side.
    // top(0) == top(), top(size()-1) == bottom()
    // Returns NULL if |index| is out of range.
    T* top(size_t index) {
        if (index < _count) {
            return (T*)_items + _mod(_start + index, _cap);
        }
        return NULL;   // including _count == 0
    }
    const T* top(size_t index) const {
        if (index < _count) {
            return (const T*)_items + _mod(_start + index, _cap);
        }
        return NULL;   // including _count == 0
    }

bottom()

    // Get address of bottom-most item, NULL if queue is empty
    T* bottom() { 
        return _count ? ((T*)_items + _mod(_start + _count - 1, _cap)) : NULL; 
    }
    const T* bottom() const {
        return _count ? ((const T*)_items + _mod(_start + _count - 1, _cap)) : NULL; 
    }
    
    // Randomly access item from bottom side.
    // bottom(0) == bottom(), bottom(size()-1) == top()
    // Returns NULL if |index| is out of range.
    T* bottom(size_t index) {
        if (index < _count) {
            return (T*)_items + _mod(_start + _count - index - 1, _cap);
        }
        return NULL;  // including _count == 0
    }
    const T* bottom(size_t index) const {
        if (index < _count) {
            return (const T*)_items + _mod(_start + _count - index - 1, _cap);
        }
        return NULL;  // including _count == 0
    }

initialized()

检查队列是否已经初始化

    // True if the queue was constructed successfully.
    bool initialized() const { return _items != NULL; }

empty()和full()

    bool empty() const { return !_count; }
    bool full() const { return _cap == _count; }

size()和capacity()

    // Number of items
    size_t size() const { return _count; }

    // Maximum number of items that can be in this queue
    size_t capacity() const { return _cap; }

max_capacity()

    // Maximum value of capacity()
    size_t max_capacity() const { return (1UL << (sizeof(_cap) * 8)) - 1; }

这个函数的内容其实是编译期间就能确定的。也就是说返回的是固定值。sizeof(_cap)返回的是_cap的字节数，乘以8则是因为一个字节（byte）等于8个位（bit）。所以sizeof(_cap) * 8)表示的就是uint32_t类型所占的bit数，1UL << (sizeof(_cap) * 8)) - 1的结果就是uint32_t的每一位都是1，也就是uint32_t所能表示的最大值。

这个函数应该是实现的类似vector的max_size()接口。只不过这里命名成了max_capacity()。vector中的max_size()表示的是vector能存储的最大元素个数，有这个函数的原因是因为vector是能够动态扩容的。而有界队列，其实是不存在动态扩容的。所以这个max_capacity()其实也没太大用。

在brpc中的应用

BoundedQueue 在brpc中有大量使用。比如实现bthread时，每个做worker用的pthread都有一个TaskGroup，而这个TaskGroup中有两个队列：

class TaskGroup {

...
    WorkStealingQueue<bthread_t> _rq;
    RemoteTaskQueue _remote_rq;
...   
};

其中的_remote_rq的类RemoteTaskQueue就是通过BoundedQueue存储的。

class RemoteTaskQueue {
public:
    RemoteTaskQueue() {}

    int init(size_t cap) {
        const size_t memsize = sizeof(bthread_t) * cap;
        void* q_mem = malloc(memsize);
        if (q_mem == NULL) {
            return -1;
        }
        butil::BoundedQueue<bthread_t> q(q_mem, memsize, butil::OWNS_STORAGE);
        _tasks.swap(q);
        return 0;
    }

由于BoundedQueue构造函数并不能很好的处理malloc失败的情况。所以这里是在外部用malloc一段内存，然后再传给BoundedQueue来管理（所有权是：OWNS_STORAGE）

class RemoteTaskQueue {
// ...
    bool pop(bthread_t* task) {
        if (_tasks.empty()) {
            return false;
        }
        _mutex.lock();
        const bool result = _tasks.pop(task);
        _mutex.unlock();
        return result;
    }

    bool push(bthread_t task) {
        _mutex.lock();
        const bool res = push_locked(task);
        _mutex.unlock();
        return res;
    }

    bool push_locked(bthread_t task) {
        return _tasks.push(task);
    }

可以看到在往RemoteTaskQueue中push和pop数据的时候，要自己添加加锁和解锁的代码，因为我们开头就已经讲过了：

BoundedQueue 不是线程安全的！

push_locked()函数，由于没有在内部加锁，所以需要调用RemoteTaskQueue的使用方在更外围加锁。看下这个函数被唯一调用过的地方：

void TaskGroup::ready_to_run_remote(bthread_t tid, bool nosignal) {
    _remote_rq._mutex.lock();
    while (!_remote_rq.push_locked(tid)) {
        flush_nosignal_tasks_remote_locked(_remote_rq._mutex);
        LOG_EVERY_SECOND(ERROR) << "_remote_rq is full, capacity="
                                << _remote_rq.capacity();
        ::usleep(1000);
        _remote_rq._mutex.lock();
    }
   ...
   ...
   }

原文地址：https://cloud.tencent.com/developer/article/1927521

版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 dio@foxmail.com 举报，一经查实，本站将立刻删除。

C++存储存储编程算法

相关推荐

如何学习编程？新手菜鸟如何学编程？
学习编程是顺着互联网的发展潮流，是一件好事。新手如何学习编程？其实不难，不过在学习编程之前你得先了解你的目的是什么？这个很重要，因为目的决定你的发展方向、决定你的发展速度。
it行业都是什么工作？it行业主要做什么？
IT行业是什么工作做什么？IT行业的工作有：产品策划类、页面设计类、前端与移动、开发与测试、营销推广类、数据运营类、运营维护类、游戏相关类等，根据不同的分类下面有细分了不同的岗位。
女生学Java好就业吗？女生适合学Java编程吗？
女生学Java好就业吗？女生适合学Java编程吗？目前有不少女生学习Java开发，但要结合自身的情况，先了解自己适不适合去学习Java，不要盲目的选择不适合自己的Java培训班进行学习。只要肯下功夫钻研，多看、多想、多练
mysql-connect-confused
Can’t connect to local MySQL server through socket \'/var/lib/mysql/mysql.sock问题 1.进入mysql路径
oracle基本命令
oracle基本命令 一、登录操作 1.管理员登录 # 管理员登录 sqlplus / as sysdba 2.普通用户登录
shell判断服务是否启动
一、背景 因为项目中需要通北京网络，所以需要连vpn，但是服务器有时候会断掉，所以写个shell脚本每五分钟去判断是否连接，于是就有下面的shell脚本。
SQL BETWEEN 操作符
BETWEEN 操作符选取介于两个值之间的数据范围内的值。这些值可以是数值、文本或者日期。
​​windows上传ipa到开发者中心app store的方法
假如你已经使用过苹果开发者中心上架app，你肯定知道在苹果开发者中心的web界面，无法直接提交ipa文件，而是需要使用第三方工具，将ipa文件上传到构建版本，开...
SQL 入门篇之什么是别名？
下面的 SQL 语句指定了两个别名，一个是 name 列的别名，一个是 country 列的别名。**提示：**如果列名称包含空格，要求使用双引号或方括号：
windows上传ipa文件到app store的方法​
在使用H5混合开发的app打包后，需要将ipa文件上传到appstore进行发布，就需要去苹果开发者中心进行发布。​
SQL AND & OR 运算符
+----+--------------+---------------------------+-------+---------+
C++ 数组
数组的声明并不是声明一个个单独的变量，比如 number0、number1、...、number99，而是声明一个数组变量，比如 numbers，然后使用 nu...
最全iOS 应用上架流程提交到AppStore
第一步：到appuploader官网下载辅助工具和iCloud驱动，使用前面创建的AppID登录。
SQL ALTER TABLE 语句
如需删除表中的列，请使用下面的语法（请注意，某些数据库系统不允许这种在数据库表中删除列的方式）：
dism操作巨慢
前不久在制作win11pe，制作了一版，1.26GB，太大了，不满意，想再裁剪下，发现这次dism mount正常，commit或discard巨慢，以前都很快...
三种方法卸载Symantec Endpoint Protection 赛门铁克
赛门铁克各个版本概览：https://knowledge.broadcom.com/external/article?legacyId=tech163829
python3.6、python3.10用什么版本的pip
实测Python 3.6.6用pip 21.3.1，再高就报错了，Python 3.10.7用pip 22.3.1是可以的
博通和诺顿
Broadcom Corporation （博通公司，股票代号AVGO）是全球领先的有线和无线通信半导体公司。其产品实现向家庭、 办公室和移动环境以及在这些环境...
高版本c4d在低版本windows系统上打开后不显示窗格
发现个问题，server2016上安装了c4d这些版本，低版本的正常显示窗格，但红色圈出的高版本c4d打开后不显示窗格，
基础网络windows不要使用TAT、如何卸载TAT
TAT：https://cloud.tencent.com/document/product/1340
pythonJavaScriptjavaHTMLPHPreactjsC#AndroidCSSNode.jssqlrpython-3.xMysqLjQueryc++pandasFlutterangularIOSdjangolinuxswifttypescript路由器JSON路由器设置无线路由器h3c华三华三路由器设置华三路由器电脑软件教程arraysdocker软件图文教程Cvue.jslaravelspring-boot