java多线程并发之旅-14-lock free queue 无锁队列[通俗易懂]

1.mov eax,[i]

2.add eax,1

3.mov [i],eax

EnQueue(x) //进队列
{ 
   
    //准备新加入的结点数据
    q = new record();
    q->value = x;
    q->next = NULL;
 
    do { 
   
        p = tail; //取链表尾指针的快照
    } while( CAS(p->next, NULL, q) != TRUE); //如果没有把结点链在尾指针上，再试
 
    CAS(tail, p, q); //置尾结点
}

EnQueue(x) //进队列改良版
{ 
   
    q = new record();
    q->value = x;
    q->next = NULL;
 
    p = tail;
    oldp = p
    do { 
   
        while (p->next != NULL)
            p = p->next;
    } while( CAS(p.next, NULL, q) != TRUE); //如果没有把结点链在尾上，再试
 
    CAS(tail, oldp, q); //置尾结点
}

DeQueue() //出队列
{ 
   
    do{ 
   
        p = head;
        if (p->next == NULL){ 
   
            return ERR_EMPTY_QUEUE;
        }
    while( CAS(head, p, p->next) != TRUE );
    return p->next->value;
}

你拿着一个装满钱的手提箱在飞机场，此时过来了一个火辣性感的美女，然后她很暖昧地挑逗着你，并趁你不注意的时候，把用一个一模一样的手提箱和你那装满钱的箱子调了个包，然后就离开了，你看到你的手提箱还在那，于是就提着手提箱去赶飞机去了。

SafeRead(q)
{ 
   
    loop:
        p = q->next;
        if (p == NULL){ 
   
            return p;
        }
 
        Fetch&Add(p->refcnt, 1);
 
        if (p == q->next){ 
   
            return p;
        }else{ 
   
            Release(p);
        }
    goto loop;
}

template

class LockFreeQueue { 
   

private:

ElementT * ring_array_;

int size_;

int head_index_;

int tail_index_;

}

do { 
   

  获取当前的tail_index_的值cur_tail_index；

  计算新的tail_index_的值：new_tail_index = (cur_tail_index + 1) % size;

} while(!CAS(tail_index_, cur_tail_index, new_tail_index));

插入元素到cur_tail_index;

char * flag_array_;

template
class LockFreeQueue {
    private:
        ElementT * ring_array_;
        char * flags_array_; // 标记位，标记某个位置的元素是否被占用
        // flags: 0：空节点；1：已被申请，正在写入
        // 2：已经写入，可以弹出;3,正在弹出操作;
        int size_;  // 环形数组的大小
        int element_num_; //队列中元素的个数
        int head_index_;
        int tail_index_;

    public:
        LockFreeQueue(int s = 0) {
            size_ = s;
            head_index_ = 0;
            tail_index_ = 0;
            element_num_ = 0;
        }
        ~LockFreeQueue() {}
    public:
        // 初始化queue。分配内存，设定size
        bool Init(void);
        const int GetSize(void) const {
            return size_;
        }

        const int GetElementNum(void) const {
            return element_num_;
        }

        // 入队函数
        bool EnQueue(const ElementT & ele);
        // 出队函数
        bool DeQueue(ElementT * ele);
};

// This function is NOT ThreadSafe!
// 应当在单线程环境中使用该函数
// OR should be called in the constructor...
template
bool LockFreeQueue::Init(void) {
    flags_array_ = new(std::nothrow) char[size_];
    if (flags_array_ == NULL)
        return false;
    memset(flags_array_, 0, size_);
    ring_array_ = reinterpret_cast(
            new(std::nothrow) char[size_ * sizeof(ElementT)]);
    if (ring_array_ == NULL)
        return false;
    memset(ring_array_, 0, size_ * sizeof(ElementT));
    return true;
}

// ThreadSafe
// 元素入队尾部
template
bool LockFreeQueue::EnQueue(const ElementT & ele) {
    if (!(element_num_ < size_))
        return false;
    int cur_tail_index = tail_index_;
    char * cur_tail_flag_index = flags_array_ + cur_tail_index;

    // 忙式等待
    // while中的原子操作：如果当前tail的标记为“”已占用(1)“，则更新cur_tail_flag_index,
    // 继续循环；否则，将tail标记设为已经占用
    while (!__sync_bool_compare_and_swap(cur_tail_flag_index, 0, 1)) {
        cur_tail_index = tail_index_;
        cur_tail_flag_index = flags_array_ +  cur_tail_index;
    }

    // 两个入队线程之间的同步
    // 取模操作可以优化
    int update_tail_index = (cur_tail_index + 1) % size_;

    // 如果已经被其他的线程更新过，则不需要更新；
    // 否则，更新为 (cur_tail_index+1) % size_;
    __sync_bool_compare_and_swap(&tail_index_, cur_tail_index, update_tail_index);

    // 申请到可用的存储空间
    *(ring_array_ + cur_tail_index) = ele;

    // 写入完毕
    __sync_fetch_and_add(cur_tail_flag_index, 1);

    // 更新size;入队线程与出队线程之间的协作
    __sync_fetch_and_add(&element_num_, 1);
    return true;
}

// ThreadSafe
// 元素出队头部
template
bool LockFreeQueue::DeQueue(ElementT * ele) {
    if (!(element_num_ > 0))
        return false;
    int cur_head_index = head_index_;
    char * cur_head_flag_index = flags_array_ + cur_head_index;
    while (!__sync_bool_compare_and_swap(cur_head_flag_index, 2, 3)) {
        cur_head_index = head_index_;
        cur_head_flag_index = flags_array_ + cur_head_index;
    }

    // 取模操作可以优化
    int update_head_index = (cur_head_index + 1) % size_;
    __sync_bool_compare_and_swap(&head_index_, cur_head_index, update_head_index);
    *ele = *(ring_array_ + cur_head_index);

    // 弹出完毕
    __sync_fetch_and_sub(cur_head_flag_index, 3);

    // 更新size
    __sync_fetch_and_sub(&element_num_, 1);
    return true;

}