消息队列优化 (3) -- grpc MPMCQueue 简介及各队列性能对比

[我的 blog 原地址：消息队列优化 -- grpc MPMCQueue 简介及各队列性能对比，写于 2020-08-24，是关于消息队列优化的一些思考，年初写过两篇，这是后来学习 mpmcq 的一些笔记。也在知乎发过，最近会持续努力同步更新到 InfoQ～]

又忍不住写一篇，是因为我们的开源项目 workflow 需要而接触到的：多生产者多消费者消息队列。

作为目前性能最优秀的 http server，内部的队列通常是我们的性能瓶颈突破口所在。

在谢爷的帮助下，我上周也学习了 grpc 内部的 MPMCQueue，并测了一些结果供大家参考～

1. 背景

以前这个系列已经写过的内容包括： 

• 单队列的多种实现：基本队列、基本队列锁外唤醒版本、google linked blocking queue、以及我们 workflow 内部使用的 double list queue 

• 单生产者单消费者队列的扩展：无锁的 kfifo + CAS 

• work stealing（没细讲）

我等鶸渣是不可能有能力创造一种队列做法的，但是谢爷可以，上述的 double list queue 就是他设计出来的，并且在多数测试场景中（包括今天要介绍的 mpmc queue）性能一骑绝尘～

源码当然是可以在 workflow 的 github 里围观到，请允许我务必再贴一次链接～！

workflow

上述的“多数”测试场景，是因为再优秀的队列，性能都是吞吐和长尾做 trade off 的，搞一些事情去做整体吞吐的优化，必然会导致不需要这些优化的场景下性能略差一丢丢。

这个系列的前两篇文章分别在：

消息队列优化(1) -- 鶸的介绍篇

消息队列优化(2) -- 几种基本实现

里边有介绍到为什么我的工作会涉及到优化消息队列。

今天要记录一下的是本鶸学习 mpmc queue 的笔记。

2. MpmcQueue 的基本介绍

mpmcqueue.cc

grpc 的 MpmcQueue 接口很简单，具体实现是用 InfLenFIFOQueue 来做的，从名字可以看到：

• 保序：先进先出； 

• 不限长度：这个有好有不好，只是一种 feature 的实现而已；

class InfLenFIFOQueue : public MPMCQueueInterface {  ...  void Put(void* elem);  void* Get(gpr_timespec* wait_time = nullptr);  int count() const { return count_.Load(MemoryOrder::RELAXED); }  ...};

传统的 basic queue 我们介绍过，我们会用一个条件变量：pthread_cond_t 去管理所有等待的人，这个条件变量内部实现可以简单设想一下，基本就是内部有个等待队列，等待队列被锁保护（所以我们 pthread_cond_wait()的时候需要传入一个 mutex）哪个线程要等就挂到等待队列上。然后东西来了，就按顺序叫醒等待的线程，如果 broadcast 就是叫醒所有人。

而 MpmcQueue 最大的特点在于，它不用这一个条件变量。它自己管等待队列，而挂上去的不是各个线程/上下文，而是各个 cond 本身，cond 在这里的作用就不涉及排队了，只涉及被唤醒的功能。

  // Returns pointer to the waiter that should be waken up next, should be the  // last added waiter.  Waiter* TopWaiter();
  Mutex mu_;        // Protecting lock  Waiter waiters_;  // Head of waiting thread queue

然后我们必须看看这个 waiter 的定义，以及等待队列如何做的。

这个等待队列和我理解的传统的 pthread_cond_t 内部实现相反，它是后进先出，后来的人先被唤醒，google 的说法是这样可以更好地利用 cache line。

  // Node for waiting thread queue. Stands for one waiting thread, should have  // exact one thread waiting on its CondVar.  // Using a doubly linked list for waiting thread queue to wake up waiting  // threads in LIFO order to reduce cache misses.  struct Waiter {    CondVar cv;    Waiter* next;    Waiter* prev;  };
  // Pushs waiter to the front of queue, require caller held mutex  void PushWaiter(Waiter* waiter);
  // Removes waiter from queue, require caller held mutex  void RemoveWaiter(Waiter* waiter);

然后来看生产者的做法：

void InfLenFIFOQueue::Put(void* elem) {  MutexLock l(&mu_);
  if (queue_tail_ == queue_head_ && curr_count != 0) {    // 因为内部使用循环数组，且实现为不限长度的队列    // 所以这里会两倍地扩容  }
  // 把东西放到数据区的末尾  queue_tail_->content = static_cast<void*>(elem);  queue_tail_ = queue_tail_->next;
  // 叫醒一个waiter_list上的消费者（如果有的话  TopWaiter()->cv.Signal();}

然后是消费者的做法：

如果没有东西，我就创建一个条件变量 cond，挂到等待队列 waiter_list 上。其他的做法和基本的消息队列 basic queue 一致：while 来检查保护条件变量的变量是否已经 ok～（哈哈哈哈真是太绕了

void* InfLenFIFOQueue::Get(gpr_timespec* wait_time) {  MutexLock l(&mu_);
  if (count_.Load(MemoryOrder::RELAXED) == 0) {    // 现创建一个cond    Waiter self;    PushWaiter(&self);    do {      self.cv.Wait(&mu_);    } while (count_.Load(MemoryOrder::RELAXED) == 0);    RemoveWaiter(&self); // 等到了就删掉自己  return PopFront();}

看看 PopFront 的做法，意思是当前能保证我被唤醒的时候至少有一个东西可以拿了。并且如果还有东西，我会叫醒 wait list 上的下一个人。

inline void* InfLenFIFOQueue::PopFront() {  void* result = queue_head_->content;  queue_head_ = queue_head_->next;  // Signal waiting thread  if (count_.Load(MemoryOrder::RELAXED) > 0) {    TopWaiter()->cv.Signal();  }  return result;}

3. 我的 demo 空跑耗时测试结果

我个人也对mpmcq的实现与double list+mpmcq这种自己管等待队列和 cond 的做法补充到了原先的队列 demo 中，扔到了 github 上：

holmes1412/queue

这里大概说几点：

1. double list空跑性能是最好的这个毫无疑问；

2. 第四第五个：mpmcq初版（锁外唤醒）和list_add_tail ，和第七第八，分别是 mpmc queue 的实现的加到 head 和加到 tail 的做法，以检查 google 说的 cache line 的优化，这两组对比都发现list_add_tail好像性能略优呀，并没有看到 cache line 的优化。

3. 最后一个是我周末闲来无事，把我们性能优秀的double list改了个自己管 wait list 版本(最后一项)，demo 空跑性能竟然最好。但是这个是不太合理的，因为 double list 本身的做法就保证了只有一个消费者会等在生产条件变量上，也就是说本来 wait list 上就是只有一个人。这个为什么测出来会更快，我还得研究研究。

4. srpc 略带序列化 QPS 测试结果

demo 版本空跑显然难以服众，而且，队列都是我写来练手的。接下来给大家看看谢爷写的 workflow 内部的 poller 在各种队列实现中，对 srpc 的小包性能测试结果。

这里补充一下 poller 基本原理：生产者为从 epoll 拿到消息的人，扔到消息队列中；消费者为用户线程，从消息队列里拿消息解析并返回给用户。

照样总结几点吧：

1. double list空跑性能是最好的这个毫无疑问；+1

2. 这里其实除了 qps，也记录了延迟，但基本结果和 qps 一致（qps 高的延迟都低），所以没有单独画延迟的图；

3. list_add_tail 再次比 add_head 要优秀！不知道 google 的 cache line 优化都在哪~~Q_Q~~ 

4. 这里的 mpmcq 有 cond 和*cond 版本，是因为谢爷实现的时候尝试过 waiter 带pthread_cond_t或者pthread_cond_t *，结果是pthread_cond_t * 的性能更好；

5. 人肉观察了 CPU 使用，基本 cpu 使用率为 11-12 个核，但是最后那项：pthread_cond_t +锁外唤醒+add_tail ，基本使用 13 个核。

5. srpc 带压缩和序列化 QPS 测试结果及其他

本来这里是想放 srpc + protobuf serialize 10000 字节数据 + Gzip 压缩的影响下，各队列的性能的，不过结果和我们组当前做的 http server benchmark 略有出入，所以等我们出一个严格的 benchmarck 之后再给大家看看～

今天所测的都是非 CAS 的单队列实现，并没有涉及到 work stealing 等多队列的做法，如果各位还有了解哪些内部基于单队列形式所实现的其他优化方案，都希望与我交流！我不允许这个世界上有我不知道的其他单队列形式的优化！(๑•̀ㅂ•́) ✧一脸好学 ing～～～之后我也会继续去研究一些关于无锁的技术，提升一下自己的视野～到时候再总结一下与大家分享。