C++实现无锁队列

CAS等原子操作

在介绍无锁队列之前，我们需要知道一个重要的技术就是CAS操作，现在几乎所有的CPU指令都支持CAS等原子操作。有了这个原子操作，我们就可以用其来实现各种无锁的数据结构了。

• 本文用到的是GCC的CAS

  GCC4.1+版本中支持CAS的原子操作

bool __sync_bool_compare_and_swap (type *ptr, type oldval type newval, ...)
type __sync_val_compare_and_swap (type *ptr, type oldval type newval, ...)
           

bool __sync_bool_compare_and_swap (type* ptr, type oldval type newval, …)

type __sync_val_compare_and_swap (type* ptr, type oldval type newval, …)

这两个函数提供原子的比较和交换，如果ptr == oldval,就将newval写入ptr,

第一个函数在相等并写入的情况下返回true.

第二个函数返回操作之前的值。

• C++中的CAS

  c++11中的STL中的atomic类的函数可以让你跨平台

template< class T >
bool atomic_compare_exchange_weak( std::atomic* obj,
                                   T* expected, T desired );
template< class T >
bool atomic_compare_exchange_weak( volatile std::atomic* obj,
                                   T* expected, T desired );
           

无锁队列的链表实现

参考文章

这篇文章讲的非常详细，大家进去浏览完无锁队列如何一步一步进行实现，然后再出来看最终实现的完整版。

• 代码最终实现

#include <atomic>
#include <memory>
#include <vector>

template <typename T> class LockFreeQueue 
{
  // 用链表实现LockFreeQueue  Node定义节点的类型
private:
  struct Node 
  {
    T data_;
    Node *next_;
    Node() : data_(), next_(nullptr) {}
    Node(T &data) : data_(data), next_(nullptr) {}
    Node(const T &data) : data_(data), next_(nullptr) {}
  };

private:
  Node *head_, *tail_;

public:
  LockFreeQueue() : head_(new Node()), tail_(head_) {}
  ~LockFreeQueue() 
  {
    Node *tmp;
    while (head_ != nullptr) 
    {
      tmp = head_;
      head_ = head_->next_;
      delete tmp;
    }
    tmp = nullptr;
    tail_ = nullptr;
  }

  bool Try_Dequeue(T &data) 
  {
    Node *old_head, *old_tail, *first_node;
    while(true) 
    {
      // 取出头指针，尾指针，和第一个元素的指针
      old_head = head_;
      old_tail = tail_; // 前面两步的顺序非常重要，一定要在获取old_tail之前获取old_head
                        // 保证old_head_不落后于old_tail
      first_node = old_head->next_;

      // 下面的工作全是在确保上面三个指针的“一致性”
      // 1. 保证head_的一致性
      /* 
         判断head_指针是否已经移动
         如果head已经被改变我们需要重新再去获取它
      */
      if (old_head != head_) 
      {
        continue;
      }

      // 2. 保证尾指针落后于头指针 (尾指针与全局尾指针部分一致性保持)
      // 空队列或者全局尾指针落后
      // 落后是指其他线程的已经更新，而当前线程没有更新
      if (old_head == old_tail) 
      {
        //这里是队列为空，返回false
        if (first_node == nullptr) 
        {
          return false;
        }
        // 证明全局尾指针没有被更新，尝试更新一下 “主动操作”
        //尾指针落后了，所以我们要更新尾指针，然后在循环一次
        ::__sync_bool_compare_and_swap(&tail_, old_tail, first_node);
        continue;
      } 
      else // 前面的操作都是在确保全局尾指针在全局头指针之后，只有这样才能安全的删除数据
      {
        // 在CAS前先取出数据，防止其他线程dequeue造成数据的缺失
        data = first_node->data_;
        // 在取出数据后，如果head指针没有落后，移动 old_head 指针成功则退出
        if (::__sync_bool_compare_and_swap(&head_, old_head, first_node)) 
        {
          break;
        }
      }
    }
    delete old_head;
    return true;
  }

  void Enqueue(const T &data) 
  {
    Node *enqueue_node = new Node(data);
    Node *old_tail, *old_tail_next;

    while(true) 
    {
      //先取一下尾指针和尾指针的next
      old_tail = tail_;
      old_tail_next = old_tail->next_;

      //如果尾指针已经被移动了，则重新开始
      if (old_tail != tail_) 
      {
        continue;
      }

      // 判断尾指针是否指向最后一个节点
      if (old_tail_next == nullptr) 
      {
        if (::__sync_bool_compare_and_swap(&(old_tail->next_), old_tail_next,
                                           enqueue_node)) 
        {
          break;
        }
      } 
      else 
      {
        // 全局尾指针不是指向最后一个节点，就把全局尾指针向后移动

        // 全局尾指针不是指向最后一个节点，发生在其他线程已经完成节点添加操作，
        // 但是并没有更新最后一个节点，此时，当前线程的(tail_和old_tail是相等的，)
        // 可以更新全局尾指针为old_tail_next，如果其他线程不更新全局尾指针，
        // 那么当前线程会不断移动，直到  old_tail_next == nullptr 为true

        // 为什么这里不直接Continue 原因是:
        // 如果其他线程添加了节点，但是并没有更新
        // 全局尾节点，就会导致所有的线程原地循环等待，所以每一个线程必须要有一些
        // “主动的操作” 去获取尾节点，这种思想在 dequeue的时候也有体现
        ::__sync_bool_compare_and_swap(&(tail_), old_tail, old_tail_next);
        continue;
      }
    }
    // 重置尾节点, 也有可能已经被别的线程重置，那么当前线程就不用管了
    /*
      不用管是因为我们已经将节点成功添加，每一次向队列中加入数据的时候都要保证将tail指针移动到最尾端
    */
    ::__sync_bool_compare_and_swap(&tail_, old_tail, enqueue_node);
  }
};
           

写个测试程序验证一下我们的无锁队列是否正确

#include<vector>
#include<unistd.h>
#include<iostream>
#include<thread>

LockFreeQueue<int> q;

void func1()
{
    std::vector<int> vec{1, 2, 3, 4, 5, 6, 7, 8, 9, 10};
    for(int i = 0; i < vec.size(); ++i)
        q.Enqueue(vec[i]);    
}

void func2()
{
    std::vector<int> vec{11, 12, 13, 14, 15, 16, 17, 18, 19, 20};
    for(int i = 0; i < vec.size(); ++i)
    {
        q.Enqueue(vec[i]);
    }
}

void func3()
{
    int a;
    for(int i = 0; i < 10; ++i)
    {
        q.Try_Dequeue(a);
        std::cout << "func3 a = " << a << "\n";
    }
}

void func4()
{
    int a;
    for(int i = 0; i < 10; ++i)
    {
        q.Try_Dequeue(a);
        std::cout << "func4 a = " << a << "\n";
    }
}

int main()
{
    std::thread t1(func1);
    std::thread t2(func2);
    std::thread t3(func3);
    std::thread t4(func4);
    t1.join();
    t2.join();
    t3.join();
    t4.join();
}
           

运行结果如图所示

读取数据没有出错。