使用C 11原子量實現自旋鎖

自旋鎖是一種基礎的同步原語，用於保障對共享資料的互斥訪問。與互斥鎖的相比，在獲取鎖失敗的時候不會使得執行緒阻塞而是一直自旋嘗試獲取鎖。當執行緒等待自旋鎖的時候，cpu不能做其他事情，而是一直處於輪詢忙等的狀態。自旋鎖主要適用於被持有時間短，執行緒不希望在重新排程上花過多時間的情況。實際上許多其他型別的鎖在底層使用了自旋鎖實現，例如多數互斥鎖在試圖獲取鎖的時候會先自旋一小段時間，然後才會休眠。如果在持鎖時間很長的場景下使用自旋鎖，則會導致cpu在這個執行緒的時間片用盡之前一直消耗在無意義的忙等上，造成計算資源的浪費。

cas（compare and swap），即比較並替換，實現併發演算法時常用到的一種技術，這種操作提供了硬體級別的原子操作（通過鎖匯流排的方式）。cas操作的原型可以認為是：

bool cas(v, a, b)

其中v代表記憶體中的變數，a代表期待的值，b表示新值。當v的值與a相等時，將v與b的值交換。邏輯上可以用下面的偽**表示：

bool cas(v, a, b)
return false;
}

需要強調的是上面的操作是原子的，要麼不做，要麼全部完成。

那麼已經擁有cas操作的情況下如何實現乙個自旋鎖呢？首先回憶自旋鎖的用途，本質上我們是希望能夠讓乙個執行緒在不滿足進入臨界區的條件時，不停的忙等輪詢，直到可以執行的時候再繼續（進入臨界區）執行。那麼，我們可能自然的想到使用乙個bool變數來表示是否可以進入臨界區，例如以下面的偽**的邏輯：

while(flag == true);
flag = true;
/*do something ...
*/flag = false;
...

這樣做的直觀想法是當flag為true的時候表示已經有執行緒處於臨界區內，只有當flag為fasle時才能進入，而在進入的時候立即將flag置為true。但是這樣做明顯存在乙個問題，判斷flag為false和設定flag為true並不是乙個不可分割的整體，有可能出現類似下面這樣的時序, 假設最初flag為false：

step

thread 1

thread 2

1while(flag == true);

2while(flag == true);

3flag = true

4flag = true

5do something

do something

6flag = false

7flag = false

step是虛構的步驟，do something為一系列指令，這裡寫在一起表示併發執行。這裡可以看出由於thread1讀取判斷flag的值與修改flag的值是兩個獨立的操作，中間插入了thread2的判斷操作，最終使得有兩個執行緒同時進入了臨界區，這與我們的期望相悖。那麼如何解決呢？如果能將讀取判斷與修改的操作合二為一，變成乙個不可分割的整體，那麼自然就不可能出現這種交錯的場景。對於這樣乙個整體操作，我們希望它能讀取記憶體中變數的值，並且當其等於特定值的時候，修改它為我們需要的另乙個值。嗯......沒錯，這樣我們就得到了cas操作。

現在可以重新修改我們的同步方式，不停的進行期望flag為false的cas操作 cas(flag, flase, b) (這裡b為true)，直到其返回成功為止，再進行臨界區中的操作，離開臨界區時將flag置為false。

b = true;
while(!cas(flag, false, b));
//do something
flag = false;

現在，判斷操作與寫入操作已經成為了乙個整體，當乙個執行緒的cas操作成功的時候會阻止其他執行緒進入臨界區，到達互斥訪問的目的。

現在我們已經可以使用cas操作來解決臨界區的互斥訪問的問題了，但是如果每次都這樣寫一遍實在太過麻煩，因此可以進行一些封裝使得使用更加方便，也就是說...可以封裝成自旋鎖。我們可以用乙個類來表示，將乙個bool值作為類的資料成員，同時將cas操作和賦值操作作為其成員函式，cas操作其實就是加鎖操作，而後面的賦值操作就是解鎖操作。

按照上面的思路，接下來用 c++ 11 引入標準庫的原子量來實現乙個自旋鎖並且進行測試。

首先，我們需要乙個bool值來表示鎖的狀態，這裡直接使用標準庫中的原子量 atomic(c++ 11的原子量可以參考：，在我的平台（cygwin64、gcc7.3）上 atomic的成員函式is_lock_free（）返回值為true，是無鎖的實現（如果內部使用了鎖來實現的話那還叫什麼自旋鎖 = =）。實際上在大多數平台上 atomic都是無鎖的，如果不確定的話也可以使用c++標準規定必須為無鎖實現的atomic_flag。

接下來，我們需要兩個原子操作，cas和賦值，c++11標準庫在原子量的成員函式中直接提供了這兩個操作。

//cas
std::atomic::compare_exchange_weak( t& expected, t desired,
std::memory_order order =
std::memory_order_seq_cst ),
std::atomic::compare_exchange_strong( t& expected, t desired,
std::memory_order order =
std::memory_order_seq_cst )
//賦值
void store( t desired, std::memory_order order = std::memory_order_seq_cst )

compare_exchange_weak 與 compare_exchange_strong 主要的區別在於記憶體中的值與expected相等的時候，cas操作是否一定能成功，compare_exchange_weak有概率會返回失敗，而compare_exchange_strong則一定會成功。因此，compare_exchange_weak必須與迴圈搭配使用來保證在失敗的時候重試cas操作。得到的好處是在某些平台上compare_exchange_weak效能更好。按照上面的模型，我們本來就要和while搭配使用，可以使用compare_exchange_weak。最後記憶體序的選擇沒有特殊需求直接使用預設的std::memory_order_seq_cst。而賦值操作非常簡單直接，這個呼叫一定會成功（只是賦值而已 = =），沒有返回值。

實現**非常短，下面是源**：

#include class spinlock 
void lock()
}void unlock()
private:
std::atomicflag_;
};

如上面所說，lock操作不停的嘗試cas操作直到成功為止，unlock操作則將bool標誌位復原。使用方式如下：

spinlock mylock;
mylock.lock();
//do something
mylock.unlock();

接下來，我們進行正確性測試，以經典的i++ 問題為例：

#include #include #include //自旋鎖類定義
class spinlock 
void lock()
}void unlock()
private:
std::atomicflag_;
};//每個執行緒自增次數
const int kincnum = 1000000;
//執行緒數
const int kworkernum = 10;
//自增計數器
int count = 0;
//自旋鎖
spinlock spinlock;
//每個執行緒的工作函式
void inccounter()
}int main()
else
return 0;
}

上面的**中建立了10個執行緒對共享的全域性變數count分別進行一百萬次++操作，然後驗證結果是否正確，最終執行的輸出為：

spinlock inc mytest start start 10 workers_every worker inc 1000000 count_: 0 workers_ end count_: 10000000

spinlock inc mytest passed

從結果中可以看出我們實現的自旋鎖起到了保護臨界區（這裡就是i++ ）的作用，count最後的值等於每個執行緒執行自增的數目之和。作為對比，可以去掉inccounter中的加鎖解鎖操作：

void inccounter()
}

執行後的輸出為：

spinlock inc mytest start start 10 workers_every worker inc 1000000 count_: 0 workers_ end count_: 7254522

spinlock inc mytest failed

結果由於多個執行緒同時執行 i++ 造成結果錯誤。

到這裡，我們就通過 c++ 11的原子量實現了乙個簡單的自旋鎖。這裡只是對c++原子量的乙個小使用，無論是自旋鎖本身還是原子量都還有許多值得**的地方。

使用C 11原子量實現自旋鎖

c 11自旋鎖的實現

C 互斥量原子鎖自旋鎖等比較

c 11 實現訊號量

使用C 11原子量實現自旋鎖

c 11自旋鎖的實現

C 互斥量 原子鎖 自旋鎖等比較

c 11 實現訊號量

相關推薦

C 互斥量原子鎖自旋鎖等比較