基於循環數組的無鎖隊列

在之前的兩篇博客（線程安全的無鎖RingBuffer的實現，多個寫線程一個讀線程的無鎖隊列實現）中，分別寫了在只有一個讀線程、一個寫線程的情況下，以及只有一個寫線程、兩個讀線程的情況下，不采用加鎖技術，甚至原子運算的循環隊列的實現。但是，在其他的情況下，我們也需要盡可能高效的線程安全的隊列的實現。本文實現了一種基於循環數組和原子運算的無鎖隊列。采用原子運算（compare and swap）而不是加鎖同步，可以很大的提高運行效率。之所以用循環數組，是因為這樣在使用過程中不需要反復開辟內存空間，可以達到較好的效率。本文的實現參考了論文Implementing Lock-Free Queues所提到的方法。但是，在這篇論文中，作者並沒有給出具體的實現。本文的實現也和論文中的方法有所不同。本文的實現基於Windows操作系統，代碼在Visual Studio 2010下測試通過。

本文的實現基於compare and swap這個原子操作，簡寫為CAS。其原型為

long CAS(long* ptr, long old_value, long new_value).

其操作為，如果ptr所指向的變量和old_value相等，則將其置為new_value. 否則什么也不做。返回值為ptr所指向的變量。

本文提出的方法的基本原理如下。代碼會在其后附上。

首先，我們約定，有一個特殊的數值，叫做EMPTY，存入數組的所有數都不能是這個數值。在一開始，將整個buffer的值都初始化為EMPTY.

第二，以兩個變量，readCnt和writeCnt，來記錄讀和寫的次數。這兩個數模數組長度，就可以得到下一次讀和寫的位置。如果readCnt和writeCnt模buffer大小相等，則說明當前隊列為控；而如果在寫的位置不是EMPTY，則說明buffer已滿。

第三，也就是最重要的，線程同步的原理。在進行寫buffer操作時，首先通過CAS操作將buffer對應位置置為指定值，而writeCnt不變，因此其他線程無法在同樣的位置進行寫操作，這樣防止了其他寫線程的覆蓋。而由於此時writeCnt還沒有變，讀線程此時也無法讀該位置的數據，這樣防止了其他讀線程的沖突。接下來，第二步，則是通過CAS操作，將writeCnt加一。可以看到，代碼中在第一個CAS操作失敗的情況下，會執行一個增大writeCnt的原子操作。這樣做的目的是避免某個線程停掉導致其他所有線程都停掉。

而在進行寫操作時，首先通過判斷readCnt和writeCnt是否相等來判斷當前隊列是否已滿。這樣做的原因如前所述，是為了避免和寫線程之間的沖突。接下來，如果當前位置可以讀，則通過一個CAS操作將readCnt加一。這樣，其他讀線程就無法再讀這個位置。而由於這個位置值還不是EMPTY，其他寫線程也無法寫這個位置。接下來，在讀走數值之后，再通過CAS操作將buffer中的這個位置置為EMPTY. 此時，寫線程可以在這個位置寫入數據。

但是，這個實現其實還有點問題。如果讀線程比較多，例如，讀線程個數和數組長度一樣，就可能兩個讀線程同時讀同一個位置。這樣的讀沖突這個程序是無法避免的。另外，如果讀數據過程中有一個讀線程停掉了，那么其他線程在讀或者寫到這個位置時，也會被阻塞。

代碼如下。

#include <windows.h>
#include <stdlib.h>

#define CAS(ptr, oldval, newval) (InterlockedCompareExchange(ptr, newval, oldval))

static const long EMPTY=-1;

class NBQueue {
public:
  NBQueue()
    : queueSize(0)
    , buffer(NULL)
    , readCnt(0)
    , writeCnt(0){}

  ~NBQueue() {
    uninit();
  }

  bool init(int size) {
    queueSize = size;
    try {
      buffer = new long[queueSize];
    }
    catch (...) {
      queueSize = 0;
      buffer = NULL;
      return false;
    }
    for (int i = 0; i < queueSize; i++) {
      buffer[i] = EMPTY;
    }
    return true;
  }
  void uninit() {
    if (buffer != NULL) {
      delete[] buffer;
      buffer = NULL;
    }
    queueSize = 0;
  }

  bool put(long v) {
    bool succ;
    long tail;
    long index;
    do {
      tail = writeCnt;
      index = tail % queueSize;
      if (buffer[index] != EMPTY) {
        return false;
      }
      long oldval = CAS(&buffer[index], EMPTY, v);
      succ = oldval == EMPTY;
      if (!succ) {
        CAS(&writeCnt, tail, tail + 1);
      }
    } while (!succ);

    CAS(&writeCnt, tail, tail + 1);

    return true;
  }

  bool get(long* v) {
    long head;
    bool succ;
    do {
      head = readCnt;
      if (head == writeCnt) {
        return false;
      }
      long oldval = CAS(&readCnt, head, head + 1);
      succ = oldval == head;
    } while (!succ);

    long index = head % queueSize;
    *v = buffer[index];
    CAS(&buffer[index], *v, EMPTY);

    return true;
  }

private:
  long* buffer;
  int queueSize;
  long readCnt;
  long writeCnt;
};