内容简介:本文基于zeromq 4.3.0版本,分析其无锁消息队列的实现。zeromq这个网络库,有以下几个亮点:zeromq内部运行着多个io线程,每个io线程内部有以下两个核心组件:
本文基于zeromq 4.3.0版本,分析其无锁消息队列的实现。
概述
zeromq这个网络库,有以下几个亮点:
- 从以往的面向TCP流的网络开发,变成了面向消息的开发。应用层关注的是什么类型的消息,库本身解决网络收发、断线重连等问题。
- 将这些消息的传输模式封装成几个模式,应用开发者只需要关注自己的业务符合什么模式,采用搭积木的方式就能构建起应用服务。
- 内部实现无锁消息队列用于对象间通信,类似actor模式。
基本架构
zeromq内部运行着多个io线程,每个io线程内部有以下两个核心组件:
- poller:即针对epoll、select等事件轮询器的封装。
- mailbox:负责接收消息的消息邮箱。
可以简单理解IO线程做的事情是:内部通过一个poller,监听着各种事件,其中包括针对IO线程的mailbox的消息,以及绑定在该IO线程上的IO对象的消息。
即这是一个per-thread-per-loop的线程设计,线程之间的通信通过消息邮箱来进行。
除了io线程之外,io对象也有mailbox,即如果想与某个IO对象通信也是通过该mailbox进行。由于消息邮箱是zeromq中的重要组成部分,下面将专门分析zeromq是如何实现的。
所有需要收发消息的对象都继承自object_t:
class object_t { public: object_t (zmq::ctx_t *ctx_, uint32_t tid_); void process_command (zmq::command_t &cmd_); private: zmq::ctx_t *ctx;// Context provides access to the global state. uint32_t tid;// Thread ID of the thread the object belongs to. void send_command (command_t &cmd_); }
而IO对象之间的命令通过command_t结构体来定义:
struct command_t { // Object to process the command. zmq::object_t *destination; enum type_t { ... } type; union { ... } args; };
可以看到,zeromq实现对象间相互通信依赖于mailbox,本文重点在分析其无锁队列的实现上。
使用无锁队列实现的消息邮箱
zeromq内部类似actor模型,每个actor内部有一个mailbox,负责收发消息,对外暴露的接口就是收发相关的send、recv接口。
负责收发消息的类是mailbox_t,内部实现使用了ypipe_t来实现无锁消息队列,而ypipe_t内部又使用了yqueue_t来实现队列,这个实现的目的是为了减少内部的分配。
下面根据上面这个图,自上而下分析邮箱的实现。
yqueue_t
yqueue_t的实现,每次能批量分配一批元素,减少内存的分配和释放。
yqueue_t内部由一个一个chunk组成,每个chunk保存N个元素,如下图:
有了chunk_t来管理数据,这样每次需要新分配元素的时候,如果当前已经没有可用元素,可以一次性分配一个chunk_t,这里面有N个元素;另外在回收的时候,也不是马上被释放,根据局部性原理可以先回收到spare_chunk里面,当再次需要分配chunk_t的时候从spare_chunk中获取。
yqueue_t内部有三个chunk_t类型指针以及对应的索引位置:
- begin_chunk/begin_pos:begin_chunk用于指向队列头的chunk,begin_pos用于指向队列第一个元素在当前chunk中的位置。
- back_chunk/back_pos:back_chunk用于指向队列尾的chunk,back_chunk用于指向队列最后一个元素在当前chunk的位置。
- end_chunk/end_pos:由于chunk是批量分配的,所以end_chunk用于指向分配的最后一个chunk位置。
注意不要混淆了back和end的作用, back_chunk/back_pos负责的是元素的存储,而end_chunk/end_pos负责的是chunk的分配 ,yqueue_t的back函数返回的是back_pos,而对外部而言,end相关的数据不可见。
如上图中:
- 有三块chunk,分别由begin_chunk、back_chunk、end_chunk组成。
- begin_pos指向begin_chunk中的第n个元素。
- back_pos指向back_chunk的最后一个元素。
- 由于back_pos已经指向了back_chunk的最后一个元素,所以end_pos就指向了end_chunk的第一个元素。
另外还有一个spare_chunk指针,用于保存释放的chunk指针,当需要再次分配chunk的时候,会首先查看这里,从这里分配chunk。这里使用了原子的cas操作来完成,利用了操作系统的局部性原理。
ypipe_t
ypipe_t在yqueue_t之上,构建了一个 单写单读的无锁队列 。
内部的元素有以下几个:
- yqueue_t
_queue:由yqueue_t实现的队列。 - T *_w:指向第一个没有被flush的元素,只能被写线程使用。
- T *_r:指向第一个未读的元素,只能被读线程使用。
- T *_f:指向第一个写入但是还没有被刷新的元素。
- atomic_ptr_t
_c:读写线程都能使用的指针,指向最后一个被刷新的元素。如果为空,那么读线程将休眠。
之所以除了写指针_w之外,还需要一个_f的刷新指针,原因在于:可能会分批次写入一堆数据,但是在没有写完毕之前,不希望被读线程看到,所以写入数据的时候由_w指针控制,而_f指针控制读线程可以看到哪些数据。
来看相关的几个对外API:
- void write (const T &value , bool incomplete ):写入数据,incomplete参数表示写入是否还没完成,在没完成的时候不会修改flush指针,即这部分数据不会让读线程看到。
- bool flush ():刷新所有已经完成的数据到管道,返回false意味着读线程在休眠,在这种情况下调用者需要唤醒读线程。
- bool read (T *value_):读数据,将读出的数据写入value指针中,返回false意味着没有数据可读。
以下面的场景来解释这个无锁队列相关的流程:
说明:以下场景忽略begin、back、end在不同chunk的情况,假设都在一个chunk完成的操作。
1、初始化
ypipe_t构造函数在初始化的时候,将push进去一个哑元素在队列尾部,然后_r、_w、_c、_f指针都同时指向队列头。 而经过这个操作之后,begin_pos和back_pos都为0,end_pos为1(因为push了一个元素)。
inline ypipe_t () { // Insert terminator element into the queue. // 先放入一个空元素 _queue.push (); // Let all the pointers to point to the terminator. // (unless pipe is dead, in which case c is set to NULL). _r = _w = _f = &_queue.back (); _c.set (&_queue.back ()); }
2、write(‘a’, true)
由于进行了push操作,因此back_pos更新为1,而end_pos更新为2。
写入一个元素a,同时incomplete为true,意味着写入还未完成,所以并没有更新flush指针,_w指针也没有在这个函数中被更新,因此当incomplete为true时不会更新上面的四个指针。
// incomplete_为true意味着这只是写入数据的一部分,此时不需要修改flush的指针指向 inline void write (const T &value_, bool incomplete_) { // 注意在这里写入数据的时候修改的是_f指针 // Place the value to the queue, add new terminator element. _queue.back () = value_; _queue.push (); // Move the "flush up to here" poiter. if (!incomplete_) // incomplete_为false表示写完毕数据了,可以修改flush指针指向 _f = &_queue.back (); }
3、write(‘b’, false)
由于进行了push操作,因此back_pos更新为1,而end_pos更新为2。
写入一个元素b,同时incomplete为false,意味着写入完成,此时需要修改flush指针指向队列尾,即新的back_pos位置2。
4、flush()
刷新数据操作,该操作中将更新_w以及_c指针。
更新_w指针的操作,由于只有写线程来完成,因此不需要加锁,_w指针用于与_f指针进行对比,快速知道是否有数据需要刷新,以唤醒读线程来继续读数据。
而_c指针,则是读写线程都可以操作,因此需要使用原子的CAS操作来修改,它的可能值有以下几种:
- NULL:读线程设置,此时意味着已经没有数据可读,读线程在休眠。
- 非零:写线程设置,这里又区分两种情况:
- 旧值为_w的情况下,cas(_w,_f)操作修改为_f,意味着如果原先的值为_w,则原子性的修改为_f,表示有更多已被刷新的数据可读。
- 在旧值为NULL的情况下,此时读线程休眠,因此可以安全的设置为当前_f指针的位置。
inline bool flush () { // If there are no un-flushed items, do nothing. // _w等于_f,意味着没有需要刷新的元素了,直接返回 if (_w == _f) return true; // Try to set 'c' to 'f'. // 如果c原来是_w,切换为_f,同时返回旧的值 // 如果返回值不是_w,意味着旧的值不是_w if (_c.cas (_w, _f) != _w) { // Compare-and-swap was unseccessful because 'c' is NULL. // This means that the reader is asleep. Therefore we don't // care about thread-safeness and update c in non-atomic // manner. We'll return false to let the caller know // that reader is sleeping. // cas操作返回不是_w,意味着_c指针为NULL // 这种情况下读线程在休眠,因此需要修改_w指针为_f并且返回false唤醒读线程 _c.set (_f); _w = _f; return false; } // Reader is alive. Nothing special to do now. Just move // the 'first un-flushed item' pointer to 'f'. // 到了这里意味着读线程没有在休眠,直接修改_w指针为_f _w = _f; return true; }
5、read(&ret)
第一次读操作,read函数返回true表示读到了数据,ret中保存的是’a’返回。
读操作首先进入check_read函数中检查是否有数据可读,做以下的判断:
- &_queue.front () != _r && _r:如果队列头不等于_r,而且_r不为NULL,意味着有预读的数据,这种情况下直接返回。
- 如果上面的条件不满足,意味着没有预读的数据。此时根据_c指针来判断是否有数据可读。使用原子的CAS操作,在_c为队列头的情况下重置为NULL,同时将_c的旧值返回到_r指针中,如果_r为队列头或者为NULL,则返回false表示没有数据可读。
- 否则,返回true意味着有数据可读。
而在check_read函数返回true表示有数据可读的情况下,read函数将pop出队列的头部数据,这个操作将begin_pos递增一位。
// 返回是否有数据可以读 inline bool check_read () { // Was the value prefetched already? If so, return. // 队列首元素位置不等于_r并且_r不为NULL,说明有元素可读 if (&_queue.front () != _r && _r) return true; // There's no prefetched value, so let us prefetch more values. // Prefetching is to simply retrieve the // pointer from c in atomic fashion. If there are no // items to prefetch, set c to NULL (using compare-and-swap). // 返回_c的旧值到_r中,同时如果_c为队列头,则设置为NULL _r = _c.cas (&_queue.front (), NULL); // If there are no elements prefetched, exit. // During pipe's lifetime r should never be NULL, however, // it can happen during pipe shutdown when items // are being deallocated. // 如果_c的旧值为队列头,或者_c的旧值为NULL,则没有数据可读 if (&_queue.front () == _r || !_r) return false; // There was at least one value prefetched. return true; } // Reads an item from the pipe. Returns false if there is no value. // available. inline bool read (T *value_) { // Try to prefetch a value. if (!check_read ()) return false; // There was at least one value prefetched. // Return it to the caller. *value_ = _queue.front (); _queue.pop (); return true; }
明白了以上的流程,具体解释第一次调用read(&ret)操作: * 在调用之前,_r指向队列头,由于_c不是指向队列头,所以_r = _c.cas (&_queue.front (), NULL)的操作并没有修改_c的值,只是将_r置为_c,然后check_read函数返回true表示有数据可读。 * 由于check_read函数返回true表示有数据可读,因此read函数中调用pop函数读出队列头数据,同时将begin_pos递增为1。
6、read(&ret)
第二次读操作,read函数返回true表示读到了数据,ret中保存的是’b’返回。
流程如下:
- 此时_r和_c为back_pos即索引位置2,而队列头为begin_pos索引位置1,因此有数据可读check_read返回true。
- 由于check_read函数返回true表示有数据可读,因此read函数中调用pop函数读出队列头数据,同时将begin_pos递增为2。
7、read(&ret)
第三次读操作(上图中没有给出),read函数返回false表示没有数据可读。
流程如下:
- 此时_r为back_pos即索引位置2,而队列头begin_pos也是2,因此check_read返回false表示没有数据可读。
总结ypipe_t的整体设计:
- 区分了几个指针,分别有以下不同的功能:
- _f:用于存放刷新数据的位置。只有写线程可以更新,在写入的数据未完成的情况下不会更新该指针。
- _w:用于存放写入数据的位置。只有写线程可以更新,只有在写入完成之后调用flush函数才会将该指针更新到_f。
- _r:用于存放读取数据的位置。只有读线程可以更新,如果_r不是队列头,则表示一直有数据可读;否则需要根据_c的值判断是否有数据可读。
- _c:指向最后一个被刷新数据的位置,读写线程都可以修改,如果为NULL表示没有数据可读。
mailbox_t
有了以上的介绍,实际理解起来mailbox_t的实现就比较简单了。但是前面分析ypipe_t的时候提到过,这个无锁队列的实现是单写单读的,而正常情况下,会有多个不同的线程同时往一个actor发消息,即需要的是多写多读的模式,来看mailbox_t中send函数的实现:
void zmq::mailbox_t::send (const command_t &cmd_) { // 这里需要加锁,因为是多写一读的邮箱 _sync.lock (); _cpipe.write (cmd_, false); const bool ok = _cpipe.flush (); _sync.unlock (); if (!ok) // flush操作返回false意味着读线程在休眠,signal发送信号唤醒读线程 _signaler.send (); }
可以从代码中看到,虽然ypipe_t的实现了一个单写单读的无锁队列,但是由于没有解决多写多读问题,还是需要在写入数据的时候加锁。 因此, zeromq号称的无锁消息队列设计,其实准确的说只是针对读写线程无锁,对于多个写线程而言还是有锁的 。
另外,由于在没有元素可读的情况下,读线程会休眠,因此需要一个唤醒读线程的机制,这里采用了signaler_t类型的成员变量_signaler,内部实现实际上一个pipe,向这个pipe写入一个字符用于唤醒读线程。
以上就是本文的全部内容,希望本文的内容对大家的学习或者工作能带来一定的帮助,也希望大家多多支持 码农网
猜你喜欢:- 这就是所谓的JavaScript异步!
- 开源软件漏洞频出,所谓的 Linus's Law 是否已过时?
- 区块链所谓的开源,也许并不是像你想的那个样子
- 【安全帮】彭博社指控华为网络设备有后门,但所谓的后门只不过是普通漏洞
- rabbitmq实现延时队列(死信队列)
- 消息队列(三)常见消息队列介绍
本站部分资源来源于网络,本站转载出于传递更多信息之目的,版权归原作者或者来源机构所有,如转载稿涉及版权问题,请联系我们。