详解高性能无锁队列的实现-1

bool CAS( int * pAddr, int nExpected, int nNew )
atomically {
  if ( *pAddr == nExpected ) {
    *pAddr = nNew ;
    return true ;
 }
  else
    return false ;
}

volatile int a;
  a = 1;
  while (!CAS(&a, 1, 2))
 {
   ;
 }

template<typename T>
class atomic_ptr_t {
public:
    void set(T *ptr_);    // ⾮原⼦操作，设置该指针的值，使用者保证安全
    T *xchg(T *val_);     // 原⼦操作，设置⼀个新的值，然后返回旧的值
    // 原来的值(ptr指向)如果和 comp_的值相同则更新为val_,并返回原来的ptr
    //   ○ 如果相等返回ptr设置之前的值，并把ptr更新为参数val_的值，；
    //   ○ 如果不相等直接返回ptr值。
    T *cas(T *cmp_, T *val_);// 原⼦操作
private:
    volatile T *ptr;
};

struct chunk_t 
 {
    T values[N];  //每个chunk_t可以容纳N个T类型的元素，以后就以一个chunk_t为单位申请内存
    chunk_t *prev;
    chunk_t *next;
 };

template<typename T, int N> // T 队列中元素类型，N 粒度
class yqueue_t {
public:
    inline yqueue_t();
    inline ~yqueue_t();
    inline T &front();   // Returns reference to the front element of the queue.
    inline T &back();    // Returns reference to the back element of the queue.
    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() // Removes element from the back end of the queue。
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; //空闲块，读写线程的共享变量
};

  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;

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

// 返回队列头部元素的引用，调用者可以通过该引用更新元素，结合pop实现出队列操作。
inline T &front() // 返回的是引用，是个左值，调用者可以通过其修改容器的值
{
    return begin_chunk->values[begin_pos];
}
// 返回队列尾部元素的引用，调用者可以通过该引用更新元素，结合push实现插入操作。
// 如果队列为空，该函数是不允许被调用。
inline T &back() // 返回的是引用，是个左值，调用者可以通过其修改容器的值
{
    return back_chunk->values[back_pos];
}

//  Adds an element to the back end of the queue.
inline void push() {
    back_chunk = end_chunk;
    back_pos = end_pos; 
    // 若执行 push 操作后该结点是否还有空间
    // 1、若该 chunk 结点仍有空间，无需扩容
    if (++end_pos != N) return;
    // 2、若该 chunk 结点没有空间了，需要扩容
    // 为什么设置为NULL? 因为如果把 spare chunk 取出则没有了，所以设置为NULL
    chunk_t *sc = spare_chunk.xchg(NULL);
    // 判断是否有 spare_chunk
    // 2.1、如果有spare_chunk，则继续复用它
    if (sc) {
        end_chunk->next = sc;
        sc->prev = end_chunk;
    }
    // 2.2、如果没有spare_chunk，则重新分配
    else {
        end_chunk->next = (chunk_t *)malloc(sizeof(chunk_t)); // 分配一个chunk
        alloc_assert(end_chunk->next);
        end_chunk->next->prev = end_chunk;  
    }
    // 更新 end_chunk 和 end_pos 
    end_chunk = end_chunk->next;
    end_pos = 0;
}

//  Removes an element from the front end of the queue.
inline void pop() {
    // 判断是否需要释放该 chunk，删除满一个chunk才会回收chunk
    if (++begin_pos == N) {
        chunk_t *o = begin_chunk;
        // 重新设置当前 chunk
        begin_chunk = begin_chunk->next;
        begin_chunk->prev = NULL;
        begin_pos = 0;
        // spare_chunk 只保留一个 chunk
        // 由于局部性原理，总是保存最新的空闲块而释放先前的空闲快
        chunk_t *cs = spare_chunk.xchg(o); 
        free(cs);
    }
}

/*
    Copyright (c) 2007-2013 Contributors as noted in the AUTHORS file
    This file is part of 0MQ.
    0MQ is free software; you can redistribute it and/or modify it under
    the terms of the GNU Lesser General Public License as published by
    the Free Software Foundation; either version 3 of the License, or
    (at your option) any later version.
    0MQ is distributed in the hope that it will be useful,
    but WITHOUT ANY WARRANTY; without even the implied warranty of
    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
    GNU Lesser General Public License for more details.
    You should have received a copy of the GNU Lesser General Public License
    along with this program.  If not, see <http://www.gnu.org/licenses/>.
*/
#ifndef __ZMQ_YQUEUE_HPP_INCLUDED__
#define __ZMQ_YQUEUE_HPP_INCLUDED__
#include <stdlib.h>
#include <stddef.h>
// #include "err.hpp"
#include "atomic_ptr.hpp"
//  yqueue is an efficient queue implementation. The main goal is
//  to minimise number of allocations/deallocations needed. Thus yqueue
//  allocates/deallocates elements in batches of N.
//
//  yqueue allows one thread to use push/back function and another one
//  to use pop/front functions. However, user must ensure that there's no
//  pop on the empty queue and that both threads don't access the same
//  element in unsynchronised manner.
//
//  T is the type of the object in the queue. 队列中元素的类型
//  N is granularity(粒度) of the queue (how many pushes have to be done till  actual memory allocation is required).
//  即是yqueue_t一个结点可以装载N个T类型的元素， yqueue_t的一个结点是一个数组
template <typename T, int N>
class yqueue_t
{
public:
    //  创建队列.
    inline yqueue_t()
    {
        begin_chunk = (chunk_t *)malloc(sizeof(chunk_t));
        alloc_assert(begin_chunk);
        begin_pos = 0;
        back_chunk = NULL; //back_chunk总是指向队列中最后一个元素所在的chunk，现在还没有元素，所以初始为空
        back_pos = 0;
        end_chunk = begin_chunk; //end_chunk总是指向链表的最后一个chunk
        end_pos = 0;
    }
    //  销毁队列.
    inline ~yqueue_t()
    {
        while (true)
        {
            if (begin_chunk == end_chunk)
            {
                free(begin_chunk);
                break;
            }
            chunk_t *o = begin_chunk;
            begin_chunk = begin_chunk->next;
            free(o);
        }
        chunk_t *sc = spare_chunk.xchg(NULL);
        free(sc);
    }
    //  Returns reference to the front element of the queue.
    //  If the queue is empty, behaviour is undefined.
    // 返回队列头部元素的引用，调用者可以通过该引用更新元素，结合pop实现出队列操作。
    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.
    // 返回队列尾部元素的引用，调用者可以通过该引用更新元素，结合push实现插入操作。
    // 如果队列为空，该函数是不允许被调用。
    inline T &back() // 返回的是引用，是个左值，调用者可以通过其修改容器的值
    {
        return back_chunk->values[back_pos];
    }
    //  Adds an element to the back end of the queue.
    inline void push()
    {
        back_chunk = end_chunk;
        back_pos = end_pos; //
        if (++end_pos != N) //end_pos!=N表明这个chunk节点还没有满
            return;
        chunk_t *sc = spare_chunk.xchg(NULL); // 为什么设置为NULL？ 因为如果把之前值取出来了则没有spare chunk了，所以设置为NULL
        if (sc)                               // 如果有spare chunk则继续复用它
        {
            end_chunk->next = sc;
            sc->prev = end_chunk;
        }
        else // 没有则重新分配
        {
            // static int s_cout = 0;
            // printf("s_cout:%d\n", ++s_cout);
            end_chunk->next = (chunk_t *)malloc(sizeof(chunk_t)); // 分配一个chunk
            alloc_assert(end_chunk->next);
            end_chunk->next->prev = end_chunk;  
        }
        end_chunk = end_chunk->next;
        end_pos = 0;
    }
    //  Removes element from the back end of the queue. In other words
    //  it rollbacks last push to the queue. Take care: Caller is
    //  responsible for destroying the object being unpushed.
    //  The caller must also guarantee that the queue isn't empty when
    //  unpush is called. It cannot be done automatically as the read
    //  side of the queue can be managed by different, completely
    //  unsynchronised thread.
    // 必须要保证队列不为空，参考ypipe_t的uwrite
    inline void unpush()
    {
        //  First, move 'back' one position backwards.
        if (back_pos) // 从尾部删除元素
            --back_pos;
        else
        {
            back_pos = N - 1; // 回退到前一个chunk
            back_chunk = back_chunk->prev;
        }
        //  Now, move 'end' position backwards. Note that obsolete end chunk
        //  is not used as a spare chunk. The analysis shows that doing so
        //  would require free and atomic operation per chunk deallocated
        //  instead of a simple free.
        if (end_pos) // 意味着当前的chunk还有其他元素占有
            --end_pos;
        else
        {
            end_pos = N - 1; // 当前chunk没有元素占用，则需要将整个chunk释放
            end_chunk = end_chunk->prev;
            free(end_chunk->next);
            end_chunk->next = NULL;
        }
    }
    //  Removes an element from the front end of the queue.
    inline void pop()
    {
        if (++begin_pos == N) // 删除满一个chunk才回收chunk
        {
            chunk_t *o = begin_chunk;
            begin_chunk = begin_chunk->next;
            begin_chunk->prev = NULL;
            begin_pos = 0;
            //  'o' has been more recently used than spare_chunk,
            //  so for cache reasons we'll get rid of the spare and
            //  use 'o' as the spare.
            chunk_t *cs = spare_chunk.xchg(o); //由于局部性原理，总是保存最新的空闲块而释放先前的空闲快
            free(cs);
        }
    }
private:
    //  Individual memory chunk to hold N elements.
    // 链表结点称之为chunk_t
    struct chunk_t
    {
        T values[N]; //每个chunk_t可以容纳N个T类型的元素，以后就以一个chunk_t为单位申请内存
        chunk_t *prev;
        chunk_t *next;
    };
    //  Back position may point to invalid memory if the queue is empty,
    //  while begin & end positions are always valid. Begin position is
    //  accessed exclusively be queue reader (front/pop), while back and
    //  end positions are accessed exclusively by queue writer (back/push).
    chunk_t *begin_chunk; // 链表头结点
    int begin_pos;        // 起始点
    chunk_t *back_chunk;  // 队列中最后一个元素所在的链表结点
    int back_pos;         // 尾部
    chunk_t *end_chunk;   // 拿来扩容的，总是指向链表的最后一个结点
    int end_pos;
    //  People are likely to produce and consume at similar rates.  In
    //  this scenario holding onto the most recently freed chunk saves
    //  us from having to call malloc/free.
    atomic_ptr_t<chunk_t> spare_chunk; //空闲块（把所有元素都已经出队的块称为空闲块），读写线程的共享变量
    //  Disable copying of yqueue.
    yqueue_t(const yqueue_t &);
    const yqueue_t &operator=(const yqueue_t &);
};
#endif

template<typename T, int N>
class ypipe_t {
public:
    inline ypipe_t();
    inline virtual ~ypipe_t();
    // 写入数据，incomplete参数表示写入是否还没完成，在没完成的时候不会修改flush指针，即这部分数据不会让读线程看到。
    inline void write(const T &value_, bool incomplete_);
    inline bool unwrite(T *value_);
    // 刷新所有已经完成的数据到管道，返回false意味着读线程在休眠，在这种情况下调用者需要唤醒读线程。
    inline bool flush();
    // 这里面有两个点，一个是检查是否有数据可读，一个是预取
    inline bool check_read();
    inline bool read(T *value_);
    inline bool probe(bool (*fn)(T &));
protected:
    yqueue_t<T, N> queue;
    T *w;       //指向第一个未刷新的元素,只被写线程使用 
    T *r;       //指向第一个还没预提取的元素，只被读线程使用
    T *f;       //指向下一轮要被刷新的一批元素中的第一个
    //读写线程共享的指针，指向每一轮刷新的起点（看代码的时候会详细说）。当c为空时，表示读线程睡眠（只会在读线程中被设置为空）
    atomic_ptr_t<T> c;
    ypipe_t(const ypipe_t &);
    const ypipe_t &operator=(const ypipe_t &);
};

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.
  // (unless pipe is dead, in which case c is set to NULL).
  r = w = f = &queue.back();//就是让r、w、f、c四个指针都指向这个end迭代器
  c.set(&queue.back()); // 保存[0]索引的位置
}

// Flush all the completed items into the pipe. Returns false if
// the reader thread is sleeping. In that case, caller is obliged to
// wake the reader up before using the pipe again.
// 刷新所有已经完成的数据到管道，返回false说明读线程在休眠，此时需要调用者唤醒读线程。
// 批量刷新的机制， 写入批量后唤醒读线程；
inline bool flush() {
    // If there are no un-flushed items, do nothing.
    // 1、没有新元素加入，不需要更新
    if (w == f) {
        return true;
    }
    // Try to set 'c' to 'f'.
    // 尝试将 c 设置为 f
    // 注意：参考 read 函数，读端没有读取到数据，则 c = NULL，
    // cas 操作：比较 c 和 w 是否相等。相等则 c = f，否则不做任何操作。返回旧的 c 值
    // 1、若 c != w，说明读端没有读取到数据(c = NULL), 不做任何操作，返回NULL
    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.
        c.set(f); // 更新 c，非原子操作（因为读线程睡眠，不存在竞争）
        w = f;    // 更新 w
        return false; // 返回 false 需要唤醒读线程，这需要写业务去做处理
    }
    // 2、若 c == w，读端还有数据可读取，则自动更新 c = f（原子操作），返回c(w)
    else {
        //  Reader is alive. Nothing special to do now. Just move
        //  the 'first un-flushed item' pointer to 'f'.
        w = f;    // 更新 w
        return true;
    }
}

// Check whether item is available for reading.
// 这里面有两个点，一个是检查是否有数据可读，一个是预取
inline bool check_read() {
    // Was the value prefetched already? If so, return.
    // 第一次 read 时，r == front，这里判断不成立
    // 判断是否在前几次调用 read 函数时已经预取到数据了，return true
    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).
    // 尝试预取数据，更新 r = 旧的 c (w=f)，分为两种情况
    // 1. 如果此时还没有数据写入，c == &queue.front(), 将 c 置为 NULL
    // 2. 如果此时已经有数据写入，c != &queue.front()，返回 c 的位置 
    r = c.cas(&queue.front(), NULL); // 尝试预取数据，r 指向可以读取到的位置
    //  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.
    // 判断是否成功预取数据
    // 若队列中无数据可读，r == front，当前队列为空；r == 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.
    // 读取操作：front + pop
    *value_ = queue.front();
    queue.pop();
    return true;
}

/*
    Copyright (c) 2007-2013 Contributors as noted in the AUTHORS file
    This file is part of 0MQ.
    0MQ is free software; you can redistribute it and/or modify it under
    the terms of the GNU Lesser General Public License as published by
    the Free Software Foundation; either version 3 of the License, or
    (at your option) any later version.
    0MQ is distributed in the hope that it will be useful,
    but WITHOUT ANY WARRANTY; without even the implied warranty of
    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
    GNU Lesser General Public License for more details.
    You should have received a copy of the GNU Lesser General Public License
    along with this program.  If not, see <http://www.gnu.org/licenses/>.
*/
#ifndef __ZMQ_YPIPE_HPP_INCLUDED__
#define __ZMQ_YPIPE_HPP_INCLUDED__
#include "atomic_ptr.hpp"
#include "yqueue.hpp"
//  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:
    //  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.
        //  (unless pipe is dead, in which case c is set to NULL).
        r = w = f = &queue.back(); //就是让r、w、f、c四个指针都指向这个end迭代器
        c.set(&queue.back());
    }
    //  The destructor doesn't have to be virtual. It is mad virtual
    //  just to keep ICC and code checking tools from complaining.
    inline virtual ~ypipe_t()
    {
    }
    //  Following function (write) deliberately copies uninitialised data
    //  when used with zmq_msg. Initialising the VSM body for
    //  non-VSM messages won't be good for performance.
#ifdef ZMQ_HAVE_OPENVMS
#pragma message save
#pragma message disable(UNINIT)
#endif
    //  Write an item to the pipe.  Don't flush it yet. If incomplete is
    //  set to true the item is assumed to be continued by items
    //  subsequently written to the pipe. Incomplete items are neverflushed down the stream.
    // 写入数据，incomplete参数表示写入是否还没完成，在没完成的时候不会修改flush指针，即这部分数据不会让读线程看到。
    inline void write(const T &value_, bool incomplete_)
    {
        //  Place the value to the queue, add new terminator element.
        queue.back() = value_;
        queue.push();
        //  Move the "flush up to here" poiter.
        if (!incomplete_)
        {
            f = &queue.back(); // 记录要刷新的位置     false 更新f
            // printf("1 f:%p, w:%p\n", f, w);
        }
        else
        {
            //  printf("0 f:%p, w:%p\n", f, w);
        }
    }
#ifdef ZMQ_HAVE_OPENVMS
#pragma message restore
#endif
    //  Pop an incomplete item from the pipe. Returns true is such
    //  item exists, false otherwise.
    inline bool unwrite(T *value_)
    {
        if (f == &queue.back())
            return false;
        queue.unpush();
        *value_ = queue.back();
        return true;
    }
    //  Flush all the completed items into the pipe. Returns false if
    //  the reader thread is sleeping. In that case, caller is obliged to
    //  wake the reader up before using the pipe again.
    // 刷新所有已经完成的数据到管道，返回false意味着读线程在休眠，在这种情况下调用者需要唤醒读线程。
    // 批量刷新的机制， 写入批量后唤醒读线程；
    // 反悔机制 unwrite
    inline bool flush()
    {
        //  If there are no un-flushed items, do nothing.
        if (w == f) // 不需要刷新，即是还没有新元素加入
            return true;
        //  Try to set 'c' to 'f'.
        // read时如果没有数据可以读取则c的值会被置为NULL
        if (c.cas(w, f) != w) // 尝试将c设置为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.
            c.set(f); // 更新w的位置
            w = f;
            return false; //线程看到flush返回false之后会发送一个消息给读线程，这需要写业务去做处理
        }
        else  // 读端还有数据可读取
        {
            //  Reader is alive. Nothing special to do now. Just move
            //  the 'first un-flushed item' pointer to 'f'.
            w = f;             // 只需要更新w的位置
            return true;
        }
    }
    //  Check whether item is available for reading.
    // 这里面有两个点，一个是检查是否有数据可读，一个是预取
    inline bool check_read()
    {
        //  Was the value prefetched already? If so, return.
        if (&queue.front() != r && r) //判断是否在前几次调用read函数时已经预取数据了return true;
            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).
        // 两种情况
        // 1. 如果c值和queue.front()， 返回c值并将c值置为NULL，此时没有数据可读
        // 2. 如果c值和queue.front()， 返回c值，此时可能有数据度的去
        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.
        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;
    }
    //  Applies the function fn to the first elemenent in the pipe
    //  and returns the value returned by the fn.
    //  The pipe mustn't be empty or the function crashes.
    inline bool probe(bool (*fn)(T &))
    {
        bool rc = check_read();
        // zmq_assert(rc);
        return (*fn)(queue.front());
    }
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为空时，表示读线程睡眠（只会在读线程中被设置为空）
    //  Disable copying of ypipe object.
    ypipe_t(const ypipe_t &);
    const ypipe_t &operator=(const ypipe_t &);
};
#endif