无锁消息队列的实现

template<typename T>
 class atomic_ptr_t {
 public:
     void set(T *ptr_);      // ⾮原⼦操作，设置该指针的值，使用者保证安全
     T *xchg(T *val_);       // 原⼦操作，设置⼀个新的值，然后返回旧的值
     T *cas(T *cmp_, T *val_);// 原⼦操作
 private:
     volatile T *ptr;
 };

//  Perform atomic 'compare and swap' operation on the pointer.
 //  The pointer is compared to 'cmp' argument and if they are equal, 
 //  its value is set to 'val'. Old value of the pointeris returned.
 inline T *cas (T *cmp_, T *val_) {
     T *old;
     __asm__ volatile (
         "lock; cmpxchg %2, %3"
         : "=a" (old), "=m" (ptr)
         : "r" (val_), "m" (ptr), "0" (cmp_)
         : "cc");
     return old;
 }

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 可以容纳 N 个 T 类型的元素
     chunk_t *prev;  // 指向前一个 chunk_t
     chunk_t *next;  // 指向后一个 chunk_t
 };

atomic_ptr_t<chunk_t> spare_chunk;  // 空闲块

chunk_t *begin_chunk;
 int begin_pos;       
 chunk_t *back_chunk; 
 int back_pos;         
 chunk_t *end_chunk;   
 int end_pos;

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，刚分配出来的chunk
     end_pos = 0;             // 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);
     }
 }

template<typename T, int N>
 class ypipe_t {
 public:
     inline ypipe_t();
     inline virtual ~ypipe_t();
     inline void write(const T &value_, bool incomplete_);
     inline bool unwrite(T *value_);
     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;
     atomic_ptr_t<T> c;
     ypipe_t(const ypipe_t &);
     const ypipe_t &operator=(const ypipe_t &);
 };

// 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;

//  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());
 }

// 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 never flushed down the stream.
 inline void write(const T &value_, bool incomplete_) {
     //  Place the value to the queue, add new terminator element.
     // 写入数据：通过 back 获取可写位置并写入数据；通过 push 更新下一个可写位置
     queue.back() = value_;
     queue.push();
     //  Move the "flush up to here" poiter.
     // 判断写操作是否已经完成
     // 1、若写入完成，则刷新 flush 指针
     if (!incomplete_) {    
         f = &queue.back(); // 记录要刷新的位置
     }
     // 2、若写入还未完成，不修改 flush 指针，read 没有数据
 }

// 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;
     }
 }

// 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;
 }

// 尝试预取数据，更新 r = 旧的 c (w=f)，分为两种情况
 // 1. 如果此时还没有数据写入，c == &queue.front(), 将 c 置为 NULL
 // 2. 如果此时已经有数据写入，c != &queue.front()，返回 c
 r = c.cas(&queue.front(), NULL); // 尝试预取数据，r 指向可以读取到的位置

// 判断是否成功预取数据
 // 若队列中无数据可读，r == front，当前队列为空；r == NULL，初始时队列为空
 if (&queue.front() == r || !r) 
     return false;
 return true;

// 第一次 read 时，r == front，这里判断不成立
 // 判断是否前几次调用 read 函数时已经预取到数据了
 if (&queue.front() != r && r) 
     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;
 }

template <typename ELEM_T, QUEUE_INT Q_SIZE>
 bool ArrayLockFreeQueue<ELEM_T, Q_SIZE>::enqueue(const ELEM_T &a_data) 
 {
     QUEUE_INT currentWriteIndex; 
     QUEUE_INT currentReadIndex;  
     // 1、获取可写入的位置(CAS)
     do 
     {
          // 获取当前读写指针的位置
         currentWriteIndex = m_writeIndex;
         currentReadIndex = m_readIndex;
          // 判断队列是否满了，(write + 1) % Q_SIZE == read   
         if(countToIndex(currentWriteIndex + 1) ==
             countToIndex(currentReadIndex))
         {
             return false;   
         }
     // 获取可写入的位置(CAS)，cas 更新失败则继续循环
     } while(!CAS(&m_writeIndex, currentWriteIndex, (currentWriteIndex+1)));
     // 2、写入数据
     m_thequeue[countToIndex(currentWriteIndex)] = a_data;  
     // 3、更新可读的位置(CAS)，多线程环境下生产者线程按序更新可读位置
     while(!CAS(&m_maximumReadIndex, currentWriteIndex, (currentWriteIndex + 1)))
     {
         sched_yield(); // cas 更新失败后，让出 cpu
     }
     // 原子操作；队列中元素数量+1
     AtomicAdd(&m_count, 1);
     return true;
 }

emplate <typename ELEM_T, QUEUE_INT Q_SIZE>
 bool ArrayLockFreeQueue<ELEM_T, Q_SIZE>::dequeue(ELEM_T &a_data)
 {
     QUEUE_INT currentMaximumReadIndex;
     QUEUE_INT currentReadIndex;
     do
     {
          // 获取当前可读位置[m_readIndex,  m_maximumReadIndex)
         currentReadIndex = m_readIndex;
         currentMaximumReadIndex = m_maximumReadIndex;
         // 若队列为空，返回 false
         if(countToIndex(currentReadIndex) ==
             countToIndex(currentMaximumReadIndex))      
         {
             return false;
         }
         // 读取数据
         a_data = m_thequeue[countToIndex(currentReadIndex)]; 
          // 更新可读的位置(CAS)
         if(CAS(&m_readIndex, currentReadIndex, (currentReadIndex + 1)))
         {
             AtomicSub(&m_count, 1); // 原子操作，元素-1
             return true;
         }
     } while(true);
     assert(0);
     return false;
 }

// ArrayLockFreeQueue.h
 #ifndef _ARRAYLOCKFREEQUEUE_H___
 #define _ARRAYLOCKFREEQUEUE_H___
 #include <stdint.h>
 #ifdef _WIN64
 #define QUEUE_INT int64_t
 #else
 #define QUEUE_INT unsigned long
 #endif
 #define ARRAY_LOCK_FREE_Q_DEFAULT_SIZE 65535
 template <typename ELEM_T, QUEUE_INT Q_SIZE = ARRAY_LOCK_FREE_Q_DEFAULT_SIZE>
 class ArrayLockFreeQueue
 {
 public:
     ArrayLockFreeQueue();
     virtual ~ArrayLockFreeQueue();
     QUEUE_INT size();
     bool enqueue(const ELEM_T &a_data); 
     bool dequeue(ELEM_T &a_data);
     bool try_dequeue(ELEM_T &a_data);
 private:
     ELEM_T m_thequeue[Q_SIZE];                  
     volatile QUEUE_INT m_count;         
     volatile QUEUE_INT m_writeIndex;    
     volatile QUEUE_INT m_readIndex;     
     volatile QUEUE_INT m_maximumReadIndex;
     inline QUEUE_INT countToIndex(QUEUE_INT a_count);
 };
 #include "ArrayLockFreeQueueImp.h"
 #endif
 // ArrayLockFreeQueueImp.h
 #ifndef _ARRAYLOCKFREEQUEUEIMP_H___
 #define _ARRAYLOCKFREEQUEUEIMP_H___
 #include "ArrayLockFreeQueue.h"
 #include <assert.h>
 #include "atom_opt.h"
 template <typename ELEM_T, QUEUE_INT Q_SIZE>
 ArrayLockFreeQueue<ELEM_T, Q_SIZE>::ArrayLockFreeQueue() :
     m_writeIndex(0),
     m_readIndex(0),
     m_maximumReadIndex(0)
 {
     m_count = 0;
 }
 template <typename ELEM_T, QUEUE_INT Q_SIZE>
 ArrayLockFreeQueue<ELEM_T, Q_SIZE>::~ArrayLockFreeQueue()
 {}
 template <typename ELEM_T, QUEUE_INT Q_SIZE>
 inline QUEUE_INT ArrayLockFreeQueue<ELEM_T, Q_SIZE>::countToIndex(QUEUE_INT a_count)
 {
     return (a_count % Q_SIZE);      
 }
 template <typename ELEM_T, QUEUE_INT Q_SIZE>
 QUEUE_INT ArrayLockFreeQueue<ELEM_T, Q_SIZE>::size()
 {
     QUEUE_INT currentWriteIndex = m_writeIndex;
     QUEUE_INT currentReadIndex = m_readIndex;
     if(currentWriteIndex>=currentReadIndex)
         return currentWriteIndex - currentReadIndex;
     else
         return Q_SIZE + currentWriteIndex - currentReadIndex;
 }
 template <typename ELEM_T, QUEUE_INT Q_SIZE>
 bool ArrayLockFreeQueue<ELEM_T, Q_SIZE>::enqueue(const ELEM_T &a_data) 
 {
     QUEUE_INT currentWriteIndex; 
     QUEUE_INT currentReadIndex;
     do 
     {
         currentWriteIndex = m_writeIndex;
         currentReadIndex = m_readIndex;
         if(countToIndex(currentWriteIndex + 1) ==
             countToIndex(currentReadIndex))
         {
             return false;   
         }
     } while(!CAS(&m_writeIndex, currentWriteIndex, (currentWriteIndex+1)));
     m_thequeue[countToIndex(currentWriteIndex)] = a_data; 
     while(!CAS(&m_maximumReadIndex, currentWriteIndex, (currentWriteIndex + 1)))
     {
         sched_yield();      
     }
     AtomicAdd(&m_count, 1);
     return true;
 }
 template <typename ELEM_T, QUEUE_INT Q_SIZE>
 bool ArrayLockFreeQueue<ELEM_T, Q_SIZE>::try_dequeue(ELEM_T &a_data)
 {
     return dequeue(a_data);
 }
 template <typename ELEM_T, QUEUE_INT Q_SIZE>
 bool ArrayLockFreeQueue<ELEM_T, Q_SIZE>::dequeue(ELEM_T &a_data)
 {
     QUEUE_INT currentMaximumReadIndex;
     QUEUE_INT currentReadIndex;
     do
     {
         currentReadIndex = m_readIndex;
         currentMaximumReadIndex = m_maximumReadIndex;
         if(countToIndex(currentReadIndex) ==
             countToIndex(currentMaximumReadIndex))      
         {
             return false;
         }
         a_data = m_thequeue[countToIndex(currentReadIndex)]; 
         if(CAS(&m_readIndex, currentReadIndex, (currentReadIndex + 1)))
         {
             AtomicSub(&m_count, 1); // 真正读取到了数据，元素-1
         }
     } while(true);
     assert(0);
     return false;
 }
 #endif