C/C++编程：ZeroMQ无锁队列ypipe_t实现

末蓝、 2022-08-28 05:47 215阅读 0赞

mailbox\_t的底层实际上使用了管道ypipe\_t来存储命令。而ypipe\_t实质上是一个无锁队列（ 由于普通锁的粒度比较大，以至于在并发量高的环境下，锁对于并发性能影响很大，所以zmq内部用了无锁队列），其底层使用了yqueue\_t队列，ypipe\_t是对yqueue\_t的再包装，所以我们先来看看yqueue\_t是怎么实现的。  
![在这里插入图片描述][20ada8f0feb44c7aa68933e2573624d0.png]

# yqueue\_t #

yqueue是ZMQ底层实现的一个高效的队列数据结构。它的主要特点有两个：

*  动态增长的时候，它能够尽可能少的进行内存分配和内存释放。
 *  允许生产者线程和消费者线程不加锁的进行生产和消费。

注意：用户必须保证不在空的yqueue上进行出队(pop)操作，并且在未进行同步的情况下，生产者和消费者不应该同时访问同一个元素。

`yqueue_t`是一个高效的队列，高效体现：

*  在它的内存配置上，尽量少的内存申请，尽量重用将要释放的内存。
 *  其次，容器的设计都会涉及到这点高效的内存配置器，像sgi stl容器的内存配置器，使用了内存池，预先分配一块比较大的内存，用不同大小的桶管理，容器申请内存时会从想要的桶里拿一块内存，释放内存时又把内存回收到相应的桶里，这样就能做到尽量少的malloc调用。yqueue\_t并没有使用内存池，但是利用了同样的思想，一次性分配一个chunk\_t减少内存分配次数，并用spare\_chunk管理将要释放的块用于内存回收

先看一下yqueue\_t的整个概况，源码位于Yqueue.hpp

*  yqueue\_t是一个模板类。其内部实现了一个链表结构用于存储数据。在分配内存的时候yqueue\_t并不是一次分配一个数据大小的内存，而是根据模板参数N，一次分配N个数据大小的内存，这样做减少了alloc/dealloc的次数，提高了效率。
 *  yqueue\_t中的元素出队列的时候其内存并不删除，而是返回到了空闲区，在下次有数据入队的时候首先查看空闲区域是否有内存区，如果有的话那么直接使用，不重新分配内存，如果没有空闲去再重新申请内存。
 *  yueue\_t允许同时一个线程push/back，另外一个线程pop/front。但需要保证不能在空队列上pop（具体原因后续说明）。yueue\_t两个线程不会同时操作同一个元素（如何保证两个线程不会操作同一个元素？后续说明）。

//  T is the type of the object in the queue.队列中元素的类型
        //  N is granularity(粒度) of the queue，简单来说就是yqueue_t一个结点可以装载N个T类型的元素，可以猜想yqueue_t的一个结点应该是个数组
        template <typename T, int N> class yqueue_t
        {
        
        public:
            inline yqueue_t ();//  Create the queue.
            inline ~yqueue_t ();//  Destroy the queue.   
            inline T &front ();//  Returns reference to the front element of the queue. If the queue is empty, behaviour is undefined.
            inline T &back ();//  Returns reference to the back element of the queue.If the queue is empty, behaviour is undefined.
            inline void push ();//  Adds an element to the back end of the queue.
            inline void pop ();//  Removes an element from the front of the queue.
            inline void unpush ()//  用于回滚操作，暂时先不管这个函数，用到再说
        private:
            //  Individual memory chunk to hold N elements.
            struct chunk_t
            {
        
                 T values [N];
                 chunk_t *prev;
                 chunk_t *next;
            };
    
            chunk_t *begin_chunk;
            int begin_pos;
            chunk_t *back_chunk;
            int back_pos;
            chunk_t *end_chunk;
            int end_pos;
    
            atomic_ptr_t<chunk_t> spare_chunk;  //空闲块（我把所有元素都已经出队的块称为空闲块），读写线程的共享变量
        };

可以看到

*  yqueue\_t是采用双向链表实现的（避免FIFO队列对内存过度频繁的分配和释放），内部由⼀个⼀个chunk\_t组成 ，每个chunk\_t可以容纳N个T类型的元素，以后就以一个chunk\_t为单位申请内存
 *  yqueue\_t采用了一个spare\_chunk，保存最近一次访问过后释放或分配出来的chunk，提高空闲chunk的cache亲和性。并且spare\_chunk使用了lock-free的atomic\_ptr\_t。
 *  因为yqueue\_t是双向链表实现，所以不支持随机访问

yqueue\_t的实现，每次批量分配⼀批元素，减少内存的分配和释放（解决不断动态内存分配的问题）：

*  有了chunk\_t来管理数据，这样每次需要新分配元素的时候，如果当前已经没有可用元素，可以一次性分配一个chunk\_t，每个chunk\_t能存储N个元素
 *  在数据出队列后，队列有多余空间的时候，回收的chunk也不是⻢上释放，⽽是根据局部性原理先回收到spare\_chunk⾥⾯，当再次需要分配chunk\_t的时候从spare\_chunk中获取。  
    ![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAT2NlYW4mJlN0YXI_size_20_color_FFFFFF_t_70_g_se_x_16]

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相关的数据不可见。  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAT2NlYW4mJlN0YXI_size_20_color_FFFFFF_t_70_g_se_x_16 1]  
如上图中，有三块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，根据上面的描述，扩容（写线程的事）和出队列（写线程的事）都会用到这个变量，所以这个变量是读写共享的，有同步的语义，zmq用了`atomic_ptr_t<T>`来做同步，

## atomic\_ptr\_t ##

atomic\_ptr\_t同样可以看成是一个指针，结构上atomic\_ptr\_t内含一个指针，提供了两个原子操作和一个非原子操作，在yqueue\_t中就需要用到其中一个原子操作xchg。

template <typename T> class atomic_ptr_t
    {
        
    public:
        atomic_ptr_t ()   //默认构造函数，会把内部指针ptr设置为null。
        ~atomic_ptr_t ()  //析构函数，什么操作都没有。
        void set (T *ptr_)  /*设置内部指针ptr为ptr_。
                            非线程安全函数，在使用的时候需要用户自己保证没有其他的线程正在操作ptr_。*/
        T *xchg (T *val_)  //线程安全，原子操作，把val_赋值给内部成员变量ptr_，并返回之前旧的ptr_。
        inline T *cas (T *cmp_, T *val_)
                         /*线程安全函数，原子操作。
                         对象内部变量ptr与cmp_做比较，如果相等，则把val_赋值给ptr，并返回旧的ptr的值。
                         如果不相等，则ptr不发生变化，返回ptr的值。
                         */
    }

*  set函数，把私有成员ptr指针设置成参数ptr\_的值，不是一个原子操作，需要使用者确保执行set过程没有其他线程使用ptr的值
 *  xchg函数，把私有成员ptr指针设置成参数val\_的值，并返回ptr设置之前的值。原子操作，操作系统保证线程安全
 *  cas函数，把私有成员ptr指针与参数cmp\_指针比较，如果相等，就把ptr设置为参数val\_的值，返回ptr设置之前的值；如果直接返回ptr值。原子操作，操作系统保证线程安全

在实现上xchg和cas函数就是包装了各种cpu提供的xchg和cas原子操作，想了解原理的可以查一查着方面的资料，这里只需要知道有这个功能就可以了。

有了这个指针，就可以保证单个读线程和单个写线程时的线程安全了。

接着，来看下yqueue\_t是如何构造、push、pop的。后面我会把begin\_chunk和begin\_pos合起来成为队头指针，back\_chunk和back\_pos合起来成为队尾指针，end\_chunk和end\_pos合起来称为容器指针

ZMQ的队列操作和C++的queue“很不一样”，在使用该数据结构时要非常小心。

*  yqueue的push并不接收参数，而只是将队列扩容，真正要插入元素需要在调用push方法之后，调用yqueue.back() = val。
 *  yqueue的pop并不真正把数据删除，而只是将队头“指针”向后移动。并且该操作不返回队头的元素。想得到队头的元素需要在调用pop方法之前，调用T val = yqueue.front()。
 *  yqueue的unpush是独有的，与pop类似，在调用unpush之前，要先调用T garbage = yqueue.front()，同时，调用者还要保证以下两个条件：
    
     *  队列不空
     *  释放要撤销的元素占用的资源

## 构造yqueue\_t ##

inline yqueue_t ()
            {
        
                 begin_chunk = (chunk_t*) malloc (sizeof (chunk_t));
                 alloc_assert (begin_chunk);
                 begin_pos = 0;
                 back_chunk = NULL;//back_chunk总是指向队列中最后一个元素所在的链表结点，现在还没有元素，所以初始为空
                 back_pos = 0;
                 end_chunk = begin_chunk;//end_chunk总是指向链表的最后一个结点
                 end_pos = 0;
            }

*  yqueue的内存动态分配都是以chunk\_t为单位，其内部是包含N个元素的数组（ZMQ使用的N值是256，通过这样的设置，能够削弱约99.6%的由内存分配带来的影响）。

## pop ##

inline void pop() {
        
        if (++begin_pos == N) {
        
            // 当前块已全部出队（废弃），用临时指针o指向
            chunk_t *o = begin_chunk;
            // 将队头指向下一块的开始，剔除旧块，并复位偏移量
            begin_chunk = begin_chunk->next;
            begin_chunk->prev = NULL;
            begin_pos = 0;
    
            // 根据“最近经常使用”内存分页算法，将O设为重用块效率最高
            // 也就是说，如果马上扩容时，不需要进行内存页面替换操作
            // geek，利用操作系统内存管理，追求性能到极致
            chunk_t *cs = spare_chunk.xchg(o);  //由于局部性原理，总是保存最新的空闲块而释放先前的空闲快
            // 别忘了释放旧的、不用的块，这回是真的没用了
            free(cs);
        }
    }

主要是链表的基本操作。pop虽然只有几行代码，却也有两个点需要注意：

*  pop掉的元素，其销毁工作交给调用者完成
 *  空闲块的保存，要求是原子操作。这得想明白为什么。原因是，空闲块是读写线程的共享变量，需要做同步，我们会在push中看到，push使用了spare\_chunk。

## push ##

inline void push ()
            {
        
                back_chunk = end_chunk;
                back_pos = end_pos;
    
                if (++end_pos != N)//end_pos==N表明这个链表结点已经满了
                    return;
    
                chunk_t *sc = spare_chunk.xchg (NULL);
                if (sc) {
        
                    end_chunk->next = sc;
                    sc->prev = end_chunk;
                } else {
        
                    end_chunk->next = (chunk_t*) malloc (sizeof (chunk_t));
                    alloc_assert (end_chunk->next);
                    end_chunk->next->prev = end_chunk;
                }
                end_chunk = end_chunk->next;
                end_pos = 0;
            }

push操作并未真正的push一个元素，只是把队尾指针指向容器指针，然后让容器指针加1，所以，这二者的值总是差1，二者的关系在第2节中的四个图中可以看的更清楚。扩容的条件是容器指针到达了容器尾（所以end\_chunk是拿来扩容的），扩容时先去spare\_chunk拿之前废弃的块（所有元素都被pop的块），拿到了就重用，没拿到就得重新申请。同样需要注意，拿空闲块需要做同步操作。

当end\_pos==N时，需要扩容，如下：  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAT2NlYW4mJlN0YXI_size_17_color_FFFFFF_t_70_g_se_x_16]

## front、back ##

这两个函数需要注意的点是，返回的是引用，是个左值，调用者可以通过二者修改容器的值。

//  Returns reference to the front element of the queue.
            //  If the queue is empty, behaviour is undefined.
            inline T &front ()
            {
        
                 return begin_chunk->values [begin_pos];
            }
    
            //  Returns reference to the back element of the queue.
            //  If the queue is empty, behaviour is undefined.
            inline T &back ()
            {
        
                return back_chunk->values [back_pos];
            }

总的来说yqueue\_t还是比较好理解，现在可以来看一看ypipe\_t的实现。

# ypipe\_t #

//  Lock-free queue implementation.
        //  Only a single thread can read from the pipe at any specific moment.
        //  Only a single thread can write to the pipe at any specific moment.
        //  T is the type of the object in the queue.
        //  N is granularity of the pipe, i.e. how many items are needed to
        //  perform next memory allocation.
        template <typename T, int N> class ypipe_t : public ypipe_base_t<T,N>   
    
        template <typename T, int N> class ypipe_base_t
        {
        
        public:
            virtual ~ypipe_base_t () {
        }
            virtual void write (const T &value_, bool incomplete_) = 0;
            virtual bool unwrite (T *value_) = 0;
            virtual bool flush () = 0;
            virtual bool check_read () = 0;
            virtual bool read (T *value_) = 0;
            virtual bool probe (bool (*fn)(T &)) = 0;
        };
        template <typename T, int N> class ypipe_t : public ypipe_base_t<T,N>
        {
        
        protected:
            //  Allocation-efficient queue to store pipe items.
            //  Front of the queue points to the first prefetched item, back of the pipe points to last un-flushed item. 
            //  Front is used only by reader thread, while back is used only by writer thread.
            yqueue_t <T, N> queue;//底层容器
    
            //  Points to the first un-flushed item. This variable is used exclusively by writer thread.
            T *w;//指向第一个未刷新的元素,只被写线程使用
    
            //  Points to the first un-prefetched item. This variable is used exclusively by reader thread.
            T *r;//指向第一个还没预提取的元素，只被读线程使用
    
            //  Points to the first item to be flushed in the future.
            T *f;//指向下一轮要被刷新的一批元素中的第一个
    
            //  The single point of contention between writer and reader thread.
            //  Points past the last flushed item. If it is NULL,reader is asleep. 
            //  This pointer should be always accessed using atomic operations.
            atomic_ptr_t <T> c;//读写线程共享的指针，指向每一轮刷新的起点（看代码的时候会详细说）。当c为空时，表示读线程睡眠（只会在读线程中被设置为空）
    }

ypipe\_t继承自ypipe\_base\_t，其提供了一组操作管道的接口，从ypipe\_t源码来看，他只是实现了这组接口，并没有提供其他的方法了。T、N在yqueue\_t中已经详细说过了。从数据成员来看，其底层使用了上面讲到的yqueue\_t，可以猜想，ypipe\_t的write、read等操作往yqueue\_t中写数据读数据。ypipe\_t开头的的注释中写道，ypipe\_t是一个无锁队列的实现，单个读单个写同时操作ypipe\_t是线程安全的。有没有发现这就是一个生产者消费者的问题，写线程是生产者，读线程是消费者，yqueue\_t就是缓冲区，由于只有一个生产者、一个消费者，并不涉及同类线程间的互斥，只需读线程和写线程同步就可以了，操作系统课程中的解法就是两个信号量+PV操作，涉及到锁，而这里是无锁，其实就是使用了数据成员中的三个指针w、r、c来实现的。至于f指针，是用来保证数据完整性的，zmq中一个完整的数据是可以分成多段往ypipe\_t中写的(下面源码的write函数incomplete\_参数)，只有数据写完整了才允许读线程去读数据，关于这点，可以在session与socket\_base\_t实例的通信中看到，后面会有文章专门详细介绍，我们前面说过的mailbox\_t底层也使用了ypipe\_t，但mailbox不会把命令分段，每次都是完整的数据。那么接下来就看看这几个指针时如何协同工作的吧。

先看ypipe\_t构造的时候做了什么事情：

//  Initialises the pipe.
    inline ypipe_t ()
    {
        
        //  Insert terminator element into the queue.
        queue.push ();//yqueue_t的尾指针加1，开始back_chunk为空，现在back_chunk指向第一个chunk_t块的第一个位置
    
        //  Let all the pointers to point to the terminator.
        r = w = f = &queue.back ();
        c.set (&queue.back ());
    }

在ypipe\_t中，back\_chunk+back\_pos类似vector的end迭代器，上面的注释"Let all the pointers to point to the terminator."也是这个意思，就是让r、w、f、c四个指针都指向这个end迭代器，有关这点在write的时候能看清晰的感受到。那么做完这一步，他们关系像下面这个样子：

ps.后面7个格子都属于同一个chunk\_t块（yqueue\_t介绍了chunk\_t结点内含一个数组）  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAT2NlYW4mJlN0YXI_size_15_color_FFFFFF_t_70_g_se_x_16]

[20ada8f0feb44c7aa68933e2573624d0.png]: /images/20220828/cf17c2a495d0493094e06f5b9a398949.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAT2NlYW4mJlN0YXI_size_20_color_FFFFFF_t_70_g_se_x_16]: /images/20220828/9cbe5faefbc64923b29b59560f933d1e.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAT2NlYW4mJlN0YXI_size_20_color_FFFFFF_t_70_g_se_x_16 1]: /images/20220828/e4c71e7d60a24fa8b69a05938aaed1b0.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAT2NlYW4mJlN0YXI_size_17_color_FFFFFF_t_70_g_se_x_16]: /images/20220828/0bb5e44ca0004551adb0719e5b89f9b5.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAT2NlYW4mJlN0YXI_size_15_color_FFFFFF_t_70_g_se_x_16]: /images/20220828/b0eb61b13ad94afeb0612627b0ebba0c.png