无锁CAS_无所队列

1 volatile int a;
2 a = 1;
3
4 // this will loop while 'a' is not equal to 1.
5 // If it is equal to 1 the operation will atomically set a to
2 and return true
6 while (!CAS(&a, 1, 2))
7 {
8 ;
9 }

ypipe_t<int, 100> yqueue; // 设置元素类型为int , 每个chunk为100个元素
// 写入元素为count , false代表这次已经写完数据
yqueue.write(count, false); // enqueue的顺序是无法保证的，我们只能计算enqueue的个数
// flush后读端能看到更新后的数据，返回false(刷新之前队列为空),可以notify唤醒读端
// 返回true(说明队列本身有数据)
// flush才会真正触发cas的调用
yqueue.flush()
//读取数据   返回true:读到元素
//返回false: 消息队列为空，可以让出cup , 或者wait写端唤醒
yqueue.read(&value)
-----------------------------------------------------------
yqueue.write(count1, true); // 写入元素为count1
yqueue.write(count2, true); // true代表这次没写完
yqueue.write(count3, false);// false代表写完了
yqueue.flush(); // flush后读端能看到更新后的数据

// T is the type of the object in the queue.队列中元素的类型
// N is granularity(粒度) of the queue，简单来说就是yqueue_t一个结点可以装载N个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 ()// 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; //空闲块（我把所有元素都已经出队的块称为空闲
    //块），读写线程的共享变量
};

//  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表明这个链表结点已经满了
            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;
            end_chunk->next = (chunk_t *)malloc(sizeof(chunk_t)); // 分配一个chunk
            alloc_assert(end_chunk->next);
            end_chunk->next->prev = end_chunk;
            // printf("s_cout:%d\n", ++s_cout);
        }
        end_chunk = end_chunk->next;
        end_pos = 0;
    }

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

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;//指向下⼀轮要被刷新的⼀批元素中的第⼀个
    atomic_ptr_t<T> c;//读写线程共享的指针，指向每⼀轮刷新的起点（看代码的
//时候会详细说）。当c为空时，表示读线程睡眠（只会在读线程中被设置为空）
    ypipe_t(const ypipe_t &);
    const ypipe_t &operator=(const 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.
        //  (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());
    }

// 写入数据，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();      // 记录要刷新的位置
    }

//  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.
        //  w由写端去控制，只受f修改
        if (w == f)     // 不需要刷新，即是还没有新元素加入
            return true;
        //  Try to set 'c' to 'f'.
        // read时如果没有数据可以读取则c的值会被置为NULL,这点非常重要
        // 如果c为NULL,则此时w和null不相等,返回null,null和w不相等 走 return false ,
        // c被设置为f
        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 = f;
            return false;       // 线程看到flush返回false之后会发送一个消息给读线程，这个是需要写业务去做处理
        }
        //  Reader is alive. Nothing special to do now. Just move
        //  the 'first un-flushed item' pointer to 'f'.
        w = f;
        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).
        // 两种情况
        // 如果此时还没有写入数据，则c和&queue.front()相等，则c被设置为NULL
        // 如果此时已经写入数据，则c和&queue.front()不相等，返回w的位置
        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.
        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;
    }

1 template <typename ELEM_T, QUEUE_INT Q_SIZE = ARRAY_LOCK_FREE_Q_D
EFAULT_SIZE>
2 class ArrayLockFreeQueue
3 {
4 public:
5
6 ArrayLockFreeQueue();
7 virtual ~ArrayLockFreeQueue();
8
9 QUEUE_INT size();
10
11 bool enqueue(const ELEM_T &a_data); // ⼊队列
12
13 bool dequeue(ELEM_T &a_data); // 出队列
14
15 bool try_dequeue(ELEM_T &a_data); // 尝试⼊队列
16
17 private:
18
19 ELEM_T m_thequeue[Q_SIZE];
20
21 volatile QUEUE_INT m_count; // 队列的元素个数
22 volatile QUEUE_INT m_writeIndex;//新元素⼊列时存放位置在数组中的下标
23
24 volatile QUEUE_INT m_readIndex;// 下⼀个出列元素在数组中的下标
25
26 volatile QUEUE_INT m_maximumReadIndex; //最后⼀个已经完成⼊列操作
                                          //的元素在数组中的下标
27
28 inline QUEUE_INT countToIndex(QUEUE_INT a_count);
29 };

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)));
  // We know now that this index is reserved for us. Use it to save the data
    // 第一次cas竞争写入位置
  m_thequeue[countToIndex(currentWriteIndex)] = a_data;
  // update the maximum read index after saving the data. It wouldn't fail if there is only one thread 
  // inserting in the queue. It might fail if there are more than 1 producer threads because this
  // operation has to be done in the same order as the previous CAS
  while(!CAS(&m_maximumReadIndex, currentWriteIndex, (currentWriteIndex + 1)))
  {
     // this is a good place to yield the thread in case there are more
    // software threads than hardware processors and you have more
    // than 1 producer thread
    // have a look at sched_yield (POSIX.1b)
    sched_yield();    // 当线程超过cpu核数的时候如果不让出cpu导致一直循环在此。
  }
  AtomicAdd(&m_count, 1);
  return true;
}

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
  {
     // to ensure thread-safety when there is more than 1 producer thread
        // a second index is defined (m_maximumReadIndex)
    currentReadIndex = m_readIndex;
    currentMaximumReadIndex = m_maximumReadIndex;
    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)))
    {
      AtomicSub(&m_count, 1); // 真正读取到了数据
      return true;
    }
  } while(true);
  assert(0);
   // Add this return statement to avoid compiler warnings
  return false;
}