2. 程式人生 > >C++11實現自旋鎖

C++11實現自旋鎖

阿新 • • 發佈：2019-01-01

http://blog.poxiao.me/p/spinlock-implementation-in-cpp11/

自旋鎖(Spinlock)

自旋鎖是一種用於保護多執行緒共享資源的鎖，與一般的互斥鎖(mutex)不同之處在於當自旋鎖嘗試獲取鎖的所有權時會以忙等待(busy waiting)的形式不斷的迴圈檢查鎖是否可用。在多處理器環境中對持有鎖時間較短的程式來說使用自旋鎖代替一般的互斥鎖往往能提高程式的效能。

自旋鎖的原理

自旋鎖有兩種基本狀態：

  1. 鎖定狀態
    鎖定狀態又稱不可用狀態，當自旋鎖被某個執行緒持有時就是鎖定狀態，在自旋鎖被釋放之前其他執行緒不能獲得鎖的所有權。
  2. 可用狀態
    當自選鎖未被任何執行緒持有時的狀態就是可用狀態。

假設某自旋鎖內部使用bool型別的flag變數來標識自旋鎖的狀態。當flag為true表示鎖定狀態，為false表示可用狀態。
獲取自旋鎖的一種可行的流程如下

其中灰色方框中的3步應是一個不可分割的原子操作。
釋放自旋鎖時只需以原子操作的形式將flag置為false。

使用C++11的原子操作實現自旋鎖

C++11提供了對原子操作的支援，其中std::atomic是標準庫提供的一個原子類模板。可以這樣來宣告一個自旋鎖互斥物件spin_mutex。

1
2
3
4
5
6
7
8
9
10
11
 
#include <atomic>

class spin_mutex {
  std::atomic<bool
 
> flag = ATOMIC_VAR_INIT(false);
public:
  spin_mutex() = default;
  spin_mutex(const spin_mutex&) = delete;
  spin_mutex& operator= (const spin_mutex&) = delete;
  void lock(); // 獲取自旋鎖
  void unlock(); // 釋放自旋鎖
};

對於lock函式，若根據前面的流程圖來實現，需要CAS(compare and swap)的原子操作，可以使用std::atomic類模板的成員函式compare_exchange_strong。該成員函式的4個過載(overload)宣告如下

1
2
3
4
5
6
7
8
9
10
 
bool compare_exchange_strong(T& expected, T desired,
                              std::memory_order success,
                              std::memory_order failure); // overload 1
bool compare_exchange_strong(T& expected, T desired,
                              std::memory_order success,
                              std::memory_order failure) volatile; // overload 2
bool compare_exchange_strong(T& expected, T desired,
                              std::memory_order order = std::memory_order_seq_cst); // overload 3
bool compare_exchange_strong(T& expected, T desired,
                              std::memory_order order = std::memory_order_seq_cst) volatile; // overload 4

compare_exchange_strong能夠自動比較*this與expected的值，如果二者相等,會將*this的值修改為desired的值(執行read-modify-write操作)，否則將expected的值修改為*this的值。
當*this被改變時compare_exchange_strong返回true,否則返回false。
這4個過載中還有3個型別為std::memory_order的引數，這些引數用於指定記憶體序(memory order)，對記憶體序的介紹超出了本文的範圍，之後會在《C++11之多執行緒(五、記憶體序(Memory Order)）》中較為詳細的介紹，這裡僅做簡單介紹。
簡單的講就是程式的記憶體訪問順序在編譯期(程式碼優化)和執行期(CPU的亂序執行)可能會被重新排序，這種重新排序往往是為了提高程式的執行速度，並且在單處理器的情況下這種重新排序不會對程式的正確性產生影響，但在多處理器的多執行緒環境中就可能得到非預期的結果。memory order就是用來指定原子操作周圍(指前後)的非原子操作的記憶體訪問如何被排序和同步。std::memory_order_seq_cst是標準庫中原子操作記憶體序的預設值，這是最為嚴格的memory order對效能有一定的損害，若不苛求效能可以總是使用這個值。
簡單介紹完了memory order再回到compare_exchange_strong的函式宣告中來。
在過載1和2中success引數用於指定當比較成功(*this與expected值相等時)執行read-modify-write操作的記憶體序，failure引數用於指定當比較失敗時執行load操作的記憶體序。
在過載3和4中order引數同時指定sccuess和failure時的記憶體序。
下面兩種呼叫方式是等價的。

1
2
 
flag.compare_exchange_strong(expected, desired, order);
flag.compare_exchange_strong(expected, desired, order, order);

對於unlock函式，可以使用std::atomic類模板的成員函式store來以原子操作的方式將flag置false。

1
2
 
void store(T desired, memory_order = std::memory_order_seq_cst);
void store(T desired, memory_order = std::memory_order_seq_cst) volatile;

使用預設記憶體序的自旋鎖完整的實現如下

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
 
class spin_mutex {
  std::atomic<bool> flag = ATOMIC_VAR_INIT(false);
public:
  spin_mutex() = default;
  spin_mutex(const spin_mutex&) = delete;
  spin_mutex& operator= (const spin_mutex&) = delete;
  void lock() {
    bool expected = false;
    while(!flag.compare_exchange_strong(expected, true))
      expected = false;
  }
  void unlock() {
    flag.store(false);
  }
};

下面的程式碼演示了使用自旋鎖保護變數num

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
 
int num = 0;
spin_mutex sm;

void thread_proc()
{

  for(int i = 0; i < 100000; ++i) {
    sm.lock();
    ++num;
    sm.unlock();
  }
}

int main()
{
  std::thread td1(thread_proc), td2(thread_proc);
  td1.join();
  td2.join();
  std::cout << num << std::endl;
  return 0;
}

指定記憶體序提高效能

通過指定記憶體序代替預設的std::memory_order_seq_cst可以提高效能。C++11的std::memory_order有6個列舉值。

Value 描述
std::memory_order_relaxed 沒有任何同步和排序限制，只需保證操作是原子的。
std::memory_order_consume 一個指定了此值的load操作在受影響的記憶體位置上執行consume操作，此操作使得另一個在同一記憶體位置執行了release操作的執行緒在此之前對資料依賴(data-dependent)的記憶體位置的寫操作為當前執行緒可見。
std::memory_order_acquire 一個指定了此值的load操作在受影響的記憶體位置上執行acquire操作，此操作使得另一個在同一記憶體位置執行了release操作的執行緒在此之前對任意的記憶體位置的寫操作為當前執行緒可見。
std::memory_order_release 一個指定了此值的store操作在受影響的記憶體位置上執行release操作，此操作使得此執行緒之前對資料依賴(data-dependent)的記憶體位置的寫操作為另一個在此之後通過對同一記憶體位置執行consume操作的執行緒可見，使得此執行緒之前對任意的記憶體位置的寫操作為另一個在此之後通過對同一個記憶體位置執行acquire操作的執行緒可見。
std::memory_order_acq_rel 一個指定了此值的read-modify-write操作，在讀階段(相當於load)對受影響的記憶體位置執行acquire操作，在寫階段(相當於store)對同一記憶體位置執行release操作。
std::memory_order_seq_cst 順序一致性，所有的執行緒觀察到的整個程式中記憶體修改順序是一致的。

在指定memory order前需要明確自旋鎖的責任:自旋鎖除了要避免多執行緒重入，還要保證一個執行緒在持有自旋鎖期間對記憶體的寫操作要能夠被另一個執行緒在獲得自旋鎖的時候可觀測到(可見)。
對照前面memory order列舉值的表，優先考慮效能。可以在unlock中對flag呼叫store置false的執行release操作。而在lock中對flag呼叫compare_exchange_strong的時候執行acquire操作。
使用Rlease-Acquire記憶體序的自旋鎖的完整實現如下

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
 
class spin_mutex {
  std::atomic<bool> flag = ATOMIC_VAR_INIT(false);
public:
  spin_mutex() = default;
  spin_mutex(const spin_mutex&) = delete;
  spin_mutex& operator= (const spin_mutex&) = delete;
  void lock() {
    bool expected = false;
    while(!flag.compare_exchange_strong(expected, true, std::memory_order_acquire))
      expected = false;
  }
  void unlock() {
    flag.store(false, std::memory_order_release);
  }
};

其中flag.compare_exchange_strong(expected, true, std::memory_order_acquire)等價於flag.compare_exchange_strong(expected, true, std::memory_order_acquire, std::memory_order_acquire)。獲得自旋鎖的所有權的充分必要條件是當且僅當compare_exchange_strong比較成功且成功改變flag的值為true，也就是說當比較失敗時是沒有必要執行acquire操作的。
於是lock函式可以再修改成

1
2
3
4
5
 
void lock() {
  bool expected = false;
  while(!flag.compare_exchange_strong(expected, true, std::memory_order_acquire, std::memory_order_relaxed))
    expected = false;
}

不使用CAS(compare-and-swap)的實現

前面的CAS實現中lock函式中的原子操作需要3個步驟:

  1. 取flag的值；
  2. 比較flag的值；
  3. 置flag一個新值。

由於自旋鎖只有兩個狀態，事實上獲取自旋鎖可以使用下面的流程


其中灰色矩形中的2步應是一個不可分割的原子操作:

  1. 取flag的值;
  2. 置flag一個新值。

這樣就將原子操作中需要做的3步減少到2步了。
std::atomic類模板的exchange成員函式能夠以原子的方式對其進行賦值並返回舊值。

1
2
 
T exchange(T desired, memory_order = std::memory_order_seq_cst);
T exchange(T desired, memory_order = std::memory_order_seq_cst) volatile;

使用exchange實現的自旋鎖

1
2
3
4
5
6
7
8
9
10
11
12
13
14
 
class spin_mutex {
  std::atomic<bool> flag = ATOMIC_VAR_INIT(false);
public:
  spin_mutex() = default;
  spin_mutex(const spin_mutex&) = delete;
  spin_mutex& operator= (const spin_mutex&) = delete;
  void lock() {
    while(flag.exchange(true, std::memory_order_acquire))
      ;
  }
  void unlock() {
    flag.store(false, std::memory_order_release);
  }
};

使用std::atomic_flag的實現

C++11並不要求std::atomic的實現必須是無鎖的(lock-free)，可以通過使用std::atomic類模板的成員函式is_lock_free來檢查atomic物件是不是無鎖的。如果自旋鎖內部的flag不是無鎖的型別那麼這個自旋鎖就沒有存在的意義了。所幸C++11提供了一個無鎖的二值(bool)原子型別std::atomic_flag。使用std::atomic_flag就可以實現一個真正有用的自旋鎖了。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
 
class spin_mutex {
  std::atomic_flag flag = ATOMIC_FLAG_INIT;
public:
  spin_mutex() = default;
  spin_mutex(const spin_mutex&) = delete;
  spin_mutex& operator= (const spin_mutex&) = delete;
  void lock() {
    while(flag.test_and_set(std::memory_order_acquire))
      ;
  }
  void unlock() {
    flag.clear(std::memory_order_release);
  }
};

與標準庫的std::mutex一樣這裡實現的spin_mutex同樣不建議直接去呼叫lock和unlock函式，而推薦使用std::lock_guard來自動管理自旋鎖。完整的實現和使用程式碼如下

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
 
#include <atomic>
#include <thread>
#include <mutex>
#include <iostream>

class spin_mutex {
  std::atomic_flag flag = ATOMIC_FLAG_INIT;
public:
  spin_mutex() = default;
  spin_mutex(const spin_mutex&) = delete;
  spin_mutex& operator= (const spin_mutex&) = delete;
  void lock() {
    while(flag.test_and_set(std::memory_order_acquire))
      ;
  }
  void unlock() {
    flag.clear(std::memory_order_release);
  }
};

int num = 0;
spin_mutex sm;

void thread_proc()
{

  for(int i = 0; i < 100000; ++i) {
    std::lock_guard<spin_mutex> lock(sm);
    ++num;
  }
}

int main()
{
  std::thread td1(thread_proc), td2(thread_proc);
  td1.join();
  td2.join();
  std::cout << num << std::endl;
  return 0;
}

相關推薦

C++11實現自旋鎖

http://blog.poxiao.me/p/spinlock-implementation-in-cpp11/ 自旋鎖(Spinlock) 自旋鎖是一種用於保護多執行緒共享資源的鎖，與一般的互斥鎖(mutex)不同之處在於當自旋鎖嘗試獲取鎖的所有權時會以忙等待(busy waiting)的形式不斷的迴圈

自旋鎖與互斥鎖的對比、手工實現自旋鎖

                    本文之前，我只是對自旋鎖有所瞭解，知道它是做什麼的，但是沒有去測試實現過，甚至以為自旋鎖只有kernel用這個，今天才發現POSIX有提供自旋鎖的介面。下面我會分析一下自旋鎖，並程式碼實現自旋鎖和互斥鎖的效能對比，以及利用C++11實現自旋鎖。 一：自旋鎖（s

利用CAS實現自旋鎖

什麼是自旋鎖 自旋鎖是一種十分常見的鎖結構，自旋鎖在申請資源但是申請不到的情況下並不會掛起，而是會選擇持續申請。這種鎖結果適用於每個執行緒佔用較少時間的鎖，並且執行緒阻塞狀態切換的代價遠高於等待的代價時使用。 CAS操作 CAS操作底層給予組合語言實現，是

C++ 11 自旋鎖

c++ chan c-s pre eas load exchange spin lock unlock // Spin lock implementation. // BasicLockable. // Async-signal safe. // unlock()

C# lock 語法糖實現原理--《.NET Core 底層入門》之自旋鎖，互斥鎖，混合鎖，讀寫鎖

在多執行緒環境中，多個執行緒可能會同時訪問同一個資源，為了避免訪問發生衝突，可以根據訪問的複雜程度採取不同的措施 原子操作適用於簡單的單個操作，無鎖演算法適用於相對簡單的一連串操作，而執行緒鎖適用於複雜的一連串操作 ### 原子操作 修改狀態要麼成功且狀態改變，要麼失敗且狀態不變，並且外部只能觀察到修改

自旋鎖，讀寫鎖和順序鎖的實現原理

並且 保護 表達 min 返回 create creat rwlock ini 常用的同步原語鎖，到多核處理器時代鎖已經是必不可少的同步方式之一了。無論設計多優秀的多線程數據結構，都避不開有競爭的臨界區，此時高效的鎖顯得至關重要。鎖的顆粒度是框架/程序設計者所關註的，

併發程式設計（三）： 使用C++11實現無鎖stack（lock-free stack)

C++11中CAS實現： template< class T> struct atomic { public: bool compare_exchange_weak( T& expected, T desired,                    

10.Linux核心設計與實現 P148---自旋鎖總結 (轉)

自旋鎖可分為用在單核處理器上和用在多核處理器上。單核處理器：用在單核處理器上，又可分為兩種：1.系統不支援核心搶佔此時自旋鎖什麼也不做，確實也不需要做什麼，因為單核處理器只有一個執行緒在執行，又不支援核心搶佔，因此資源不可能會被其他的執行緒訪問到。2.系統支援核心搶佔這種情況下，自旋鎖加鎖僅僅是禁止了核心搶佔

c++執行緒併發：mutex，atomic，自旋鎖spinlock，單執行緒

自旋鎖直接呼叫了boost庫的#include <boost/smart_ptr/detail/spinlock.hpp> 測試執行緒併發的情況下耗時，不太明白為什麼boost的spinl

c++11實現寫優先的讀寫鎖

先直接貼出程式碼： #ifndef __WRITE_FIRST_RW_LOCK_H #define __WRITE_FIRST_RW_LOCK_H #include <mutex> #include <condition_variable> cl

簡單的非公平自旋鎖以及基於排隊的公平自旋鎖的實現

基礎 什麼是自旋鎖 由於本文主要討論的都是自旋鎖，所以首先就需要弄明白什麼是自旋鎖。 自旋鎖最大的特徵，就是它會一直迴圈檢測鎖的狀態，當鎖處於被佔用的狀態時，不會將當前執行緒阻塞住，而是任由它繼續消耗CPU Cycles，直到發現需要的鎖處於可用狀態。

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

現象： （1）單執行緒無鎖速度最快，但應用場合受限； （2）多執行緒無鎖速度第二快，但結果不對，未保護臨界程式碼段； （3）多執行緒原子鎖第三快，且結果正確； （4）多執行緒互斥量較慢，慢與原子鎖近10倍，結果正確； （5）多執行緒自旋鎖最慢，慢與原子鎖30倍，結果

Java 中 synchronized 的實現原理及偏向鎖、輕量級鎖、自旋鎖、公平鎖簡介

    在多執行緒程式設計中，synchronized 一直都是元老級別的存在，很多人都稱之為重量級鎖。本文來簡單介紹synchronized的實現原理，以及為減少獲得鎖和釋放鎖所帶來的效能損耗而引進的偏向鎖與輕量級鎖。     Java中使用synchronized來實現

聊聊高併發（八）實現幾種自旋鎖（三）

在聊聊高併發（七）實現幾種自旋鎖（二） 這篇中介紹了兩種佇列鎖，一種是有界佇列鎖，一種是無界佇列鎖。其中無界佇列鎖CLHLock採用了連結串列的方式來組織多執行緒，使用了兩個ThreadLocal做指標指向自身的node和前一個node。它的特點是在前一個node的lock

聊聊高併發（六）實現幾種自旋鎖（一）

在聊聊高併發（五）理解快取一致性協議以及對併發程式設計的影響 我們瞭解了處理器快取一致性協議的原理，並且提到了它對併發程式設計的影響，“多個執行緒對同一個變數一直使用CAS操作，那麼會有大量修改操作，從而產生大量的快取一致性流量，因為每一次CAS操作都會發出廣播通知其他處理

Linux核心原始碼之自旋鎖的實現

1 Linux核心同步 Linux核心中有許多共享資源，這些共享資源是核心中程序都有機會訪問到的。核心對其中一些共享資源的訪問是獨佔的，因此需要提供機制對共享資源進行保護，確保任意時刻只有一個程序在訪問共享資源。自旋鎖就是一種共享資源保護機制，確

synchronized實現原理及其優化-（自旋鎖，偏向鎖，輕量鎖，重量鎖）

1.synchronized概述： 　　synchronized修飾的方法或程式碼塊相當於併發中的臨界區，即在同一時刻jvm只允許一個執行緒進入執行。synchronized是通過鎖機制實現同一時刻只允許一個執行緒來訪問共享資源的。另外synchronized鎖機制還可以保證執行緒併發執行的原子性，有序性，可

自旋鎖

pre 自旋 線程鎖 虛擬機 鎖競爭 pinning 取消 lock 等待 自旋鎖 自旋鎖可以使線程在沒有取得鎖的時候，可以不放棄CPU時間片，不被掛起，而轉去執行一個空循環，（即所謂的自旋，就是自己執行空循環），若在若幹個空循環後，線程如果可以獲得鎖，則繼續執行。若線程

本地自旋鎖與信號量/多服務臺自旋隊列-spin wait風格的信號量

我們 一件事 以及 睡眠 為我 這也 鎖改變 蘇打水 mic 周日傍晚，我去家附近的超市(...)買蘇打水，準備自制青檸蘇打。我感覺我做的比買的那個巴黎水要更爽口。由於天氣太熱，非常多人都去超市避暑去了，超市也不攆人，這仿佛是他們的策略。人過來避暑了，走的時候難免要買些東

java自旋鎖

是否 ext 並且 大量 需要 命中率 logs div 共享 一 Test-and-Set Lock 所謂測試設置是最基本的鎖，每個線程共用一把鎖，進入臨界區之前看沒有有線程在臨界區，如果沒有，則進入，並上鎖，如果有則等待。java實踐中利用了原子的設置state變量來保