template<typename T, int N>    // 链表结点称之为chunk_t    struct chunk_t    {        T values[N]; //每个chunk_t可以容纳N个T类型的元素，以后就以一个chunk_t为单位申请内存        chunk_t *prev;        chunk_t *next;    };

93     //  Adds an element to the back end of the queue.94     inline void push()95     {96         back_chunk = end_chunk;97         back_pos = end_pos; //99         if (++end_pos != N) //end_pos!=N表明这个chunk节点还没有满100             return;101102         chunk_t *sc = spare_chunk.xchg(NULL); // 为什么设置为NULL？因为如果把之前值取出来了则没有spare chunk了，所以设置为NULL103         if (sc)                               // 如果有spare chunk则继续复用它104         {105             end_chunk->next = sc;106             sc->prev = end_chunk;107         }108         else // 没有则重新分配109         {110             // static int s_cout = 0;111             // printf("s_cout:%d\n", ++s_cout);112             end_chunk->next = (chunk_t *)malloc(sizeof(chunk_t)); // 分配一个chunk113             alloc_assert(end_chunk->next);114             end_chunk->next->prev = end_chunk;115         }116         end_chunk = end_chunk->next;117         end_pos = 0;118     }

102         chunk_t *sc = spare_chunk.xchg(NULL);

154     //  Removes an element from the front end of the queue.155     inline void pop()156     {157         if (++begin_pos == N) // 删除满一个chunk才回收chunk158         {159             chunk_t *o = begin_chunk;160             begin_chunk = begin_chunk->next;161             begin_chunk->prev = NULL;162             begin_pos = 0;163164             //  'o' has been more recently used than spare_chunk,165             //  so for cache reasons we'll get rid of the spare and166             //  use 'o' as the spare.167             chunk_t *cs = spare_chunk.xchg(o); //由于局部性原理，总是保存最新的空闲块而释放先前的空闲快168             free(cs);169         }170     }

37     //  Initialises the pipe. 38     inline ypipe_t() 49     //  The destructor doesn't have to be virtual. It is mad virtual 50     //  just to keep ICC and code checking tools from complaining. 

 51     inline virtual ~ypipe_t() 52     { 53     }
 67     // 写入数据，incomplete参数表示写入是否还没完成，在没完成的时候不会修改flush指针，即这部分数据不会让读线程看到。 68     inline void write(const T &value_, bool incomplete_);
 92     inline bool unwrite(T *value_);
104     // 刷新所有已经完成的数据到管道，返回false意味着读线程在休眠，在这种情况下调用者需要唤醒读线程。105     // 批量刷新的机制， 写入批量后唤醒读线程；106     // 反悔机制 unwrite107     inline bool flush(); 
136     //  Check whether item is available for reading.137     // 这里面有两个点，一个是检查是否有数据可读，一个是预取138     inline bool check_read();163     //  Reads an item from the pipe. Returns false if there is no value.164     //  available.165     inline bool read(T *value_)178     //  Applies the function fn to the first elemenent in the pipe179     //  and returns the value returned by the fn.180     //  The pipe mustn't be empty or the function crashes.181     inline bool probe(bool (*fn)(T &))189 protected:190     //  Allocation-efficient queue to store pipe items.191     //  Front of the queue points to the first prefetched item, back of192     //  the pipe points to last un-flushed item. Front is used only by193     //  reader thread, while back is used only by writer thread.194     yqueue_t queue;195 196     //  Points to the first un-flushed item. This variable is used197     //  exclusively by writer thread.198     T *w; //指向第一个未刷新的元素,只被写线程使用199 200     //  Points to the first un-prefetched item. This variable is used201     //  exclusively by reader thread.202     T *r; //指向第一个还没预提取的元素，只被读线程使用203 204     //  Points to the first item to be flushed in the future.205     T *f; //指向下一轮要被刷新的一批元素中的第一个206 207     //  The single point of contention between writer and reader thread.208     //  Points past the last flushed item. If it is NULL,209     //  reader is asleep. This pointer should be always accessed using210     //  atomic operations.211     atomic_ptr_t c; //读写线程共享的指针，指向每一轮刷新的起点（看代码的时候会详细说）。当c为空时，表示读线程睡眠（只会在读线程中被设置为空）212 213     //  Disable copying of ypipe object.214     ypipe_t(const ypipe_t &);215     const ypipe_t &operator=(const ypipe_t &);

101     //  Flush all the completed items into the pipe. Returns false if102     //  the reader thread is sleeping. In that case, caller is obliged to103     //  wake the reader up before using the pipe again.104     // 刷新所有已经完成的数据到管道，返回false意味着读线程在休眠，在这种情况下调用者需要唤醒读线程。105     // 批量刷新的机制， 写入批量后唤醒读线程；106     // 反悔机制 unwrite107     inline bool flush()108     {109         //  If there are no un-flushed items, do nothing.110         if (w == f) // 不需要刷新，即是还没有新元素加入111             return true;112 113         //  Try to set 'c' to 'f'.114         // read时如果没有数据可以读取则c的值会被置为NULL115         if (c.cas(w, f) != w) // 尝试将c设置为f，即是准备更新w的位置116         {117 118             //  Compare-and-swap was unseccessful because 'c' is NULL.119             //  This means that the reader is asleep. Therefore we don't120             //  care about thread-safeness and update c in non-atomic121             //  manner. We'll return false to let the caller know122             //  that reader is sleeping.123             c.set(f); // 更新为新的f位置124             w = f;125             return false; //线程看到flush返回false之后会发送一个消息给读线程，这需要写业务去做处理126         }127         else  // 读端还有数据可读取128         {129             //  Reader is alive. Nothing special to do now. Just move130             //  the 'first un-flushed item' pointer to 'f'.131             w = f;             // 更新f的位置132             return true;133         }134     }

else  // 读端还有数据可读取{//  Reader is alive. Nothing special to do now. Just move//  the 'first un-flushed item' pointer to 'f'.w = f;             // 更新f的位置return true;}

//  Try to set 'c' to 'f'.// read时如果没有数据可以读取则c的值会被置为NULLif (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'//  that reader is sleeping.c.set(f); // 更新为新的f位置w = f;return false; //线程看到flush返回false之后会发送一个消息给读线程，这需要写业务去做处理}

 64     //  Write an item to the pipe.  Don't flush it yet. If incomplete is 65     //  set to true the item is assumed to be continued by items 66     //  subsequently written to the pipe. Incomplete items are neverflushed down the stream. 67     // 写入数据，incomplete参数表示写入是否还没完成，在没完成的时候不会修改flush指针，即这部分数据不会让读线程看到。 68     inline void write(const T &value_, bool incomplete_) 69     { 70         //  Place the value to the queue, add new terminator element. 71         queue.back() = value_; 72         queue.push(); 73  74         //  Move the "flush up to here" poiter. 75         if (!incomplete_) 76         { 77             f = &queue.back(); // 记录要刷新的位置 78             // printf("1 f:%p, w:%p\n", f, w); 79         } 80         else 81         { 82             //  printf("0 f:%p, w:%p\n", f, w); 83         } 84     }

138     inline bool check_read()139     {140         //  Was the value prefetched already? If so, return.141         if (&queue.front() != r && r) //判断是否在前几次调用read函数时已经预取数据了return true;142             return true;143 144         //  There's no prefetched value, so let us prefetch more values.145         //  Prefetching is to simply retrieve the146         //  pointer from c in atomic fashion. If there are no147         //  items to prefetch, set c to NULL (using compare-and-swap).148         // 两种情况149         // 1. 如果c值和queue.front()， 返回c值并将c值置为NULL，此时没有数据可读150         // 2. 如果c值和queue.front()， 返回c值，此时可能有数据度的去151         r = c.cas(&queue.front(), NULL); //尝试预取数据152 153         //  If there are no elements prefetched, exit.154         //  During pipe's lifetime r should never be NULL, however,155         //  it can happen during pipe shutdown when items are being deallocated.156         if (&queue.front() == r || !r) //判断是否成功预取数据157             return false;158 159         //  There was at least one value prefetched.160         return true;161     }162 163     //  Reads an item from the pipe. Returns false if there is no value.164     //  available.165     inline bool read(T *value_)166     {167         //  Try to prefetch a value.168         if (!check_read())169             return false;170 171         //  There was at least one value prefetched.172         //  Return it to the caller.173         *value_ = queue.front();174         queue.pop();175         return true;176     }

 14 template  15 class ArrayLockFreeQueue 16 { 17 public: 18  19     ArrayLockFreeQueue(); 20     virtual ~ArrayLockFreeQueue(); 21  22     QUEUE_INT size(); 23  24     bool enqueue(const ELEM_T &a_data);//入队列 25  26     bool dequeue(ELEM_T &a_data);//出队列 27  28     bool try_dequeue(ELEM_T &a_data); 29  30 private: 31  32     ELEM_T m_thequeue[Q_SIZE]; 33  34     volatile QUEUE_INT m_count; 35     volatile QUEUE_INT m_writeIndex; 36  37     volatile QUEUE_INT m_readIndex; 38  39     volatile QUEUE_INT m_maximumReadIndex; 40  41     inline QUEUE_INT countToIndex(QUEUE_INT a_count); 42 };

#define CAS(a_ptr, a_oldVal, a_newVal) __sync_bool_compare_and_swap(a_ptr, a_oldVal, a_newVal)

(m_writeIndex+1) % Q_SIZE == m_readIndex

countToIndex(currentWriteIndex + 1) == countToIndex(currentReadIndex)

 m_readIndex == m_maximumReadIndex

 42 template <typename ELEM_T, QUEUE_INT Q_SIZE> 43 bool ArrayLockFreeQueue::enqueue(const ELEM_T &a_data) 44 { 45     QUEUE_INT currentWriteIndex;        // 获取写指针的位置 46     QUEUE_INT currentReadIndex; 47     // 1. 获取可写入的位置 48     do 49     { 50         currentWriteIndex = m_writeIndex; 51         currentReadIndex = m_readIndex; 52         if(countToIndex(currentWriteIndex + 1) == 53             countToIndex(currentReadIndex)) 54         { 55             return false;   // 队列已经满了  56         } 57         // 目的是为了获取一个能写入的位置 58     } while(!CAS(&m_writeIndex, currentWriteIndex, (currentWriteIndex+1))); 59     // 获取写入位置后 currentWriteIndex 是一个临时变量，保存我们写入的位置 60     // We know now that this index is reserved for us. Use it to save the data 61     m_thequeue[countToIndex(currentWriteIndex)] = a_data;  // 把数据更新到对应的位置 62  63     // 2. 更新可读的位置，按着m_maximumReadIndex+1的操作 64     // update the maximum read index after saving the data. It wouldn't fail if there is only one thread  65     // inserting in the queue. It might fail if there are more than 1 producer threads because this 66     // operation has to be done in the same order as the previous CAS 67     while(!CAS(&m_maximumReadIndex, currentWriteIndex, (currentWriteIndex + 1))) 68     { 69          // this is a good place to yield the thread in case there are more 70         // software threads than hardware processors and you have more 71         // than 1 producer thread 72         // have a look at sched_yield (POSIX.1b) 73         sched_yield();      // 当线程超过cpu核数的时候如果不让出cpu导致一直循环在此。 74     } 75  76     AtomicAdd(&m_count, 1); 77  78     return true; 79  80 }

while(!CAS(&m_maximumReadIndex, currentWriteIndex, (currentWriteIndex + 1)){sched_yield();      // 当线程超过cpu核数的时候如果不让出cpu导致一直循环在此。}

 88 template  89 bool ArrayLockFreeQueue::dequeue(ELEM_T &a_data) 90 { 91     QUEUE_INT currentMaximumReadIndex; 92     QUEUE_INT currentReadIndex; 93  94     do 95     { 96          // to ensure thread-safety when there is more than 1 producer thread 97         // a second index is defined (m_maximumReadIndex) 98         currentReadIndex = m_readIndex; 99         currentMaximumReadIndex = m_maximumReadIndex;100 101         if(countToIndex(currentReadIndex) ==102             countToIndex(currentMaximumReadIndex))      // 如果不为空，获取到读索引的位置103         {104             // the queue is empty or105             // a producer thread has allocate space in the queue but is 106             // waiting to commit the data into it107             return false;108         }109         // retrieve the data from the queue110         a_data = m_thequeue[countToIndex(currentReadIndex)]; // 从临时位置读取的111 112         // try to perfrom now the CAS operation on the read index. If we succeed113         // a_data already contains what m_readIndex pointed to before we 114         // increased it115         if(CAS(&m_readIndex, currentReadIndex, (currentReadIndex + 1)))116         {117             AtomicSub(&m_count, 1); // 真正读取到了数据，元素-1118             return true;119         }120     } while(true);121 122     assert(0);123      // Add this return statement to avoid compiler warnings124     return false;125 126 }

 67     while(!CAS(&m_maximumReadIndex, currentWriteIndex, (currentWriteIndex + 1))) 68     { 69          // this is a good place to yield the thread in case there are more 70         // software threads than hardware processors and you have more 71         // than 1 producer thread 72         // have a look at sched_yield (POSIX.1b) 73         sched_yield();      // 当线程超过cpu核数的时候如果不让出cpu导致一直循环在此。 74     }

template <typename ELEM_T>    bool ArrayLockFreeQueue::push(const ELEM_T &a_data)    {        uint32_t currentReadIndex;        uint32_t currentWriteIndex;
        currentWriteIndex = m_writeIndex;        currentReadIndex  = m_readIndex;        if (countToIndex(currentWriteIndex + 1) ==             countToIndex(currentReadIndex))        {            // the queue is full            return false;        }
        // save the date into the q        m_theQueue[countToIndex(currentWriteIndex)] = a_data;
        // increment atomically write index. Now a consumer thread can read        // the piece of data that was just stored         AtomicAdd(&m_writeIndex, 1);
        return true;    }
    template <typename ELEM_T>    bool ArrayLockFreeQueue::pop(ELEM_T &a_data)    {    uint32_t currentMaximumReadIndex;    uint32_t currentReadIndex;do    {        // m_maximumReadIndex doesn't exist when the queue is set up as        // single-producer. The maximum read index is described by the current        // write index        currentReadIndex        = m_readIndex;        currentMaximumReadIndex = m_writeIndex;
        if (countToIndex(currentReadIndex) ==             countToIndex(currentMaximumReadIndex))        {            // the queue is empty or            // a producer thread has allocate space in the queue but is             // waiting to commit the data into it            return false;        }
        // retrieve the data from the queue        a_data = m_theQueue[countToIndex(currentReadIndex)];
        // try to perfrom now the CAS operation on the read index. If we succeed        // a_data already contains what m_readIndex pointed to before we         // increased it        if (CAS(&m_readIndex, currentReadIndex, (currentReadIndex + 1)))        {            return true;        }
        // it failed retrieving the element off the queue. Someone else must        // have read the element stored at countToIndex(currentReadIndex)        // before we could perform the CAS operation        
    } while(1); // keep looping to try again!
    // Something went wrong. it shouldn't be possible to reach here    assert(0);
    // Add this return statement to avoid compiler warnings    return false;    }

template <typename ELEM_T>        inline uint32_t ArrayLockFreeQueue::size()    {        uint32_t currentWriteIndex = m_writeIndex;        uint32_t currentReadIndex  = m_readIndex;
        if (currentWriteIndex >= currentReadIndex)        {            return (currentWriteIndex - currentReadIndex);        }        else        {            return (m_totalSize + currentWriteIndex - currentReadIndex);        }    }

74   template <typename U> 75   inline bool enqueue(U &&item) 76   {    77     idx_t nodeIdx = allocate_node_for(std::forward(item)); 78        79     auto tail_ = tail.load(std::memory_order_relaxed); 80     while (!tail.compare_exchange_weak(tail_, nodeIdx, std::memory_order_release, std::memory_order_relaxed)) 81       continue; 82     get_node_at(tail_)->next.store(nodeIdx, std::memory_order_release); 83        84     return true; 85   }

87   inline bool try_dequeue(T &item) {…….125           add_node_to_free_list(head_, headNode);}

80     while (!tail.compare_exchange_weak(tail_, nodeIdx, std::memory_order_release, std::memory_order_relaxed))81       continue;