template<typename T, int N>    // 鏈表結(jié)點(diǎn)稱之為chunk_t    struct chunk_t    {        T values[N]; //每個(gè)chunk_t可以容納N個(gè)T類型的元素，以后就以一個(gè)chunk_t為單位申請(qǐng)內(nèi)存        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表明這個(gè)chunk節(jié)點(diǎn)還沒(méi)有滿100             return;101102         chunk_t *sc = spare_chunk.xchg(NULL); // 為什么設(shè)置為NULL？因?yàn)槿绻阎爸等〕鰜?lái)了則沒(méi)有spare chunk了，所以設(shè)置為NULL103         if (sc)                               // 如果有spare chunk則繼續(xù)復(fù)用它104         {105             end_chunk->next = sc;106             sc->prev = end_chunk;107         }108         else // 沒(méi)有則重新分配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)); // 分配一個(gè)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) // 刪除滿一個(gè)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     // 寫入數(shù)據(jù)，incomplete參數(shù)表示寫入是否還沒(méi)完成，在沒(méi)完成的時(shí)候不會(huì)修改flush指針，即這部分?jǐn)?shù)據(jù)不會(huì)讓讀線程看到。 68     inline void write(const T &value_, bool incomplete_);
 92     inline bool unwrite(T *value_);
104     // 刷新所有已經(jīng)完成的數(shù)據(jù)到管道，返回false意味著讀線程在休眠，在這種情況下調(diào)用者需要喚醒讀線程。105     // 批量刷新的機(jī)制， 寫入批量后喚醒讀線程；106     // 反悔機(jī)制 unwrite107     inline bool flush(); 
136     //  Check whether item is available for reading.137     // 這里面有兩個(gè)點(diǎn)，一個(gè)是檢查是否有數(shù)據(jù)可讀，一個(gè)是預(yù)取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; //指向第一個(gè)未刷新的元素,只被寫線程使用199 200     //  Points to the first un-prefetched item. This variable is used201     //  exclusively by reader thread.202     T *r; //指向第一個(gè)還沒(méi)預(yù)提取的元素，只被讀線程使用203 204     //  Points to the first item to be flushed in the future.205     T *f; //指向下一輪要被刷新的一批元素中的第一個(gè)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; //讀寫線程共享的指針，指向每一輪刷新的起點(diǎn)（看代碼的時(shí)候會(huì)詳細(xì)說(shuō)）。當(dāng)c為空時(shí)，表示讀線程睡眠（只會(huì)在讀線程中被設(shè)置為空）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     // 刷新所有已經(jīng)完成的數(shù)據(jù)到管道，返回false意味著讀線程在休眠，在這種情況下調(diào)用者需要喚醒讀線程。105     // 批量刷新的機(jī)制， 寫入批量后喚醒讀線程；106     // 反悔機(jī)制 unwrite107     inline bool flush()108     {109         //  If there are no un-flushed items, do nothing.110         if (w == f) // 不需要刷新，即是還沒(méi)有新元素加入111             return true;112 113         //  Try to set 'c' to 'f'.114         // read時(shí)如果沒(méi)有數(shù)據(jù)可以讀取則c的值會(huì)被置為NULL115         if (c.cas(w, f) != w) // 嘗試將c設(shè)置為f，即是準(zhǔn)備更新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之后會(huì)發(fā)送一個(gè)消息給讀線程，這需要寫業(yè)務(wù)去做處理126         }127         else  // 讀端還有數(shù)據(jù)可讀取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??//?讀端還有數(shù)據(jù)可讀取{//??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時(shí)如果沒(méi)有數(shù)據(jù)可以讀取則c的值會(huì)被置為NULLif (c.cas(w, f) != w) // 嘗試將c設(shè)置為f，即是準(zhǔn)備更新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之后會(huì)發(fā)送一個(gè)消息給讀線程，這需要寫業(yè)務(wù)去做處理}

 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     // 寫入數(shù)據(jù)，incomplete參數(shù)表示寫入是否還沒(méi)完成，在沒(méi)完成的時(shí)候不會(huì)修改flush指針，即這部分?jǐn)?shù)據(jù)不會(huì)讓讀線程看到。 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) //判斷是否在前幾次調(diào)用read函數(shù)時(shí)已經(jīng)預(yù)取數(shù)據(jù)了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，此時(shí)沒(méi)有數(shù)據(jù)可讀150         // 2. 如果c值和queue.front()， 返回c值，此時(shí)可能有數(shù)據(jù)度的去151         r = c.cas(&queue.front(), NULL); //嘗試預(yù)取數(shù)據(jù)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) //判斷是否成功預(yù)取數(shù)據(jù)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);//入隊(duì)列 25  26     bool dequeue(ELEM_T &a_data);//出隊(duì)列 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;   // 隊(duì)列已經(jīng)滿了  56         } 57         // 目的是為了獲取一個(gè)能寫入的位置 58     } while(!CAS(&m_writeIndex, currentWriteIndex, (currentWriteIndex+1))); 59     // 獲取寫入位置后 currentWriteIndex 是一個(gè)臨時(shí)變量，保存我們寫入的位置 60     // We know now that this index is reserved for us. Use it to save the data 61     m_thequeue[countToIndex(currentWriteIndex)] = a_data;  // 把數(shù)據(jù)更新到對(duì)應(yīng)的位置 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();      // 當(dāng)線程超過(guò)cpu核數(shù)的時(shí)候如果不讓出cpu導(dǎo)致一直循環(huán)在此。 74     } 75  76     AtomicAdd(&m_count, 1); 77  78     return true; 79  80 }

while(!CAS(&m_maximumReadIndex, currentWriteIndex, (currentWriteIndex + 1)){sched_yield();      // 當(dāng)線程超過(guò)cpu核數(shù)的時(shí)候如果不讓出cpu導(dǎo)致一直循環(huán)在此。}

 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)]; // 從臨時(shí)位置讀取的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); // 真正讀取到了數(shù)據(jù)，元素-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();      // 當(dāng)線程超過(guò)cpu核數(shù)的時(shí)候如果不讓出cpu導(dǎo)致一直循環(huán)在此。 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;