自旋鎖與互斥鎖的對比、C--11 實現自旋鎖

這篇文章分析一下自旋鎖，並代碼實現自旋鎖和互斥鎖的性能對比，以及利用 C++11 實現自旋鎖。

一：自旋鎖（spin lock）

自旋鎖是一種用於保護多線程共享資源的鎖，與一般互斥鎖（mutex）不同之處在於當自旋鎖嘗試獲取鎖時以忙等待（busy waiting）的形式不斷地循環檢查鎖是否可用。

在多 CPU 的環境中，對持有鎖較短的程序來說，使用自旋鎖代替一般的互斥鎖往往能夠提高程序的性能。

最後加粗的句子很重要，本文將針對該結論進行驗證。

下面是 man 手冊中對自旋鎖 pthread_spin_lock() 函數的描述：

DESCRIPTION The pthread_spin_lock() function shall lock the spin lock referenced by lock. The calling thread shall acquire the lock if it is not held by another thread. Otherwise, the thread shall spin (that is, shall not return from the pthread_spin_lock() call) until the lock becomes available. The results are undefined if the calling thread holds the lock at the time the call is made. The pthread_spin_trylock() function shall lock the spin lock referenced by lock if it is not held by any thread. Otherwise, the function shall fail. The results are undefined if any of these functions is called with an uninitialized spin lock.

可以看出，自選鎖的主要特徵：當自旋鎖被一個線程獲得時，它不能被其它線程獲得。如果其他線程嘗試去 phtread_spin_lock() 獲得該鎖，那麼它將不會從該函數返回，而是一直自旋（spin），直到自旋鎖可用爲止。

使用自旋鎖時要注意：

由於自旋時不釋放 CPU，因而持有自旋鎖的線程應該儘快釋放自旋鎖，否則等待該自旋鎖的線程會一直在哪裏自旋，這就會浪費 CPU 時間。
持有自旋鎖的線程在 sleep 之前應該釋放自旋鎖以便其他咸亨可以獲得該自旋鎖。內核編程中，如果持有自旋鎖的代碼 sleep 了就可能導致整個系統掛起。(下面會解釋）

使用任何鎖都需要消耗系統資源（內存資源和 CPU 時間），這種資源消耗可以分爲兩類：

1. 建立鎖所需要的資源

2. 當線程被阻塞時所需要的資源

POSIX 提供的與自旋鎖相關的函數有以下幾個，都在 <pthread.h> 中。

int pthread_spin_init(pthread_spinlock_t *lock, int pshared);

初始化 spin lock，當線程使用該函數初始化一個未初始化或者被 destroy 過的 spin lock 有效。該函數會爲 spin lock 申請資源並且初始化 spin lock 爲 unlocked 狀態。

有關第二個選項是這麼說的：

If the Thread Process-Shared Synchronization option is supported and the value of pshared is PTHREAD_PROCESS_SHARED, the implementation shall permit the spin lock to be operated upon by any thread that has access to the memory where the spin lock is allocated, even if it is allocated in memory that is shared by multiple processes. If the Thread Process-Shared Synchronization option is supported and the value of pshared is PTHREAD_PROCESS_PRIVATE, or if the option is not supported, the spin lock shall only be operated upon by threads created within the same process as the thread that initialized the spin lock. If threads of differing processes attempt to operate on such a spin lock, the behav‐ ior is undefined.

所以，如果初始化 spin lock 的線程設置第二個參數爲 PTHREAD_PROCESS_SHARED，那麼該 spin lock 不僅被初始化線程所在的進程中所有線程看到，而且可以被其他進程中的線程看到，PTHREAD_PROESS_PRIVATE 則只被同一進程中線程看到。如果不設置該參數，默認爲後者。

int pthread_spin_destroy(pthread_spinlock_t *lock);

銷燬 spin lock，作用和 mutex 的相關函數類似，就不翻譯了：

The pthread_spin_destroy() function shall destroy the spin lock referenced by lock and release any resources used by the lock. The effect of subsequent use of the lock is undefined until the lock is reinitialized by another call to pthread_spin_init(). The results are undefined if pthread_spin_destroy() is called when a thread holds the lock, or if this function is called with an uninitialized thread spin lock.

不過和 mutex 的 destroy 函數一樣有這樣的性質（當初害慘了我）：

The result of referring to copies of that object in calls to pthread_spin_destroy(), pthread_spin_lock(), pthread_spin_try‐ lock(), or pthread_spin_unlock() is undefined.

int pthread_spin_lock(pthread_spinlock_t *lock);

加鎖函數，功能上文都說過了，不過這麼一點值得注意：

EBUSY A thread currently holds the lock. These functions shall not return an error code of [EINTR].

int pthread_spin_trylock(pthread_spinlock_t *lock);

還有這個函數，這個一般很少用到。

int pthread_spin_unlock(pthread_spinlock_t *lock);

解鎖函數。不是持有鎖的線程調用或者解鎖一個沒有 lock 的 spin lock 這樣的行爲都是 undefined 的。

二：自旋鎖和互斥鎖的區別

從實現原理上來講，Mutex 屬於 sleep-waiting 類型的鎖。例如在一個雙核的機器上有兩個線程 (線程 A 和線程 B)，它們分別運行在 Core0 和 Core1 上。假設線程 A 想要通過 pthread_mutex_lock 操作去得到一個臨界區的鎖，而此時這個鎖正被線程 B 所持有，那麼線程 A 就會被阻塞(blocking)，Core0 會在此時進行上下文切換(Context Switch) 將線程 A 置於等待隊列中，此時 Core0 就可以運行其他的任務 (例如另一個線程 C) 而不必進行忙等待。而 Spin lock 則不然，它屬於 busy-waiting 類型的鎖，如果線程 A 是使用 pthread_spin_lock 操作去請求鎖，那麼線程 A 就會一直在 Core0 上進行忙等待並不停的進行鎖請求，直到得到這個鎖爲止。

如果大家去查閱 Linux glibc 中對 pthreads API 的實現 NPTL(Native POSIX Thread Library) 的源碼的話 (使用”getconf GNU_LIBPTHREAD_VERSION” 命令可以得到我們系統中 NPTL 的版本號)，就會發現 pthread_mutex_lock()操作如果沒有鎖成功的話就會調用 system_wait()的系統調用並將當前線程加入該 mutex 的等待隊列裏。而 spin lock 則可以理解爲在一個 while(1)循環中用內嵌的彙編代碼實現的鎖操作(印象中看過一篇論文介紹說在 linux 內核中 spin lock 操作只需要兩條 CPU 指令，解鎖操作只用一條指令就可以完成)。有興趣的朋友可以參考另一個名爲 sanos 的微內核中 pthreds API 的實現：mutex.c spinlock.c，儘管與 NPTL 中的代碼實現不盡相同，但是因爲它的實現非常簡單易懂，對我們理解 spin lock 和 mutex 的特性還是很有幫助的。

對於自旋鎖來說，它只需要消耗很少的資源來建立鎖；隨後當線程被阻塞時，它就會一直重複檢查看鎖是否可用了，也就是說當自旋鎖處於等待狀態時它會一直消耗 CPU 時間。

對於互斥鎖來說，與自旋鎖相比它需要消耗大量的系統資源來建立鎖；隨後當線程被阻塞時，線程的調度狀態被修改，並且線程被加入等待線程隊列；最後當鎖可用時，在獲取鎖之前，線程會被從等待隊列取出並更改其調度狀態；但是在線程被阻塞期間，它不消耗 CPU 資源。

因此自旋鎖和互斥鎖適用於不同的場景。自旋鎖適用於那些僅需要阻塞很短時間的場景，而互斥鎖適用於那些可能會阻塞很長時間的場景。

三：自旋鎖與 linux 內核進程調度關係

現在我們就來說一說之前的問題，如果臨界區可能包含引起睡眠的代碼則不能使用自旋鎖，否則可能引起死鎖：

那麼爲什麼信號量保護的代碼可以睡眠而自旋鎖會死鎖呢？

先看下自旋鎖的實現方法吧，自旋鎖的基本形式如下：

    spin_lock(&mr_lock):
    
    //critical region
 
    spin_unlock(&mr_lock);

跟蹤一下 spin_lock(&mr_lock) 的實現

#define spin_lock(lock) _spin_lock(lock)
#define _spin_lock(lock) __LOCK(lock)
#define __LOCK(lock) \
do { preempt_disable(); __acquire(lock); (void)(lock); } while (0)

注意到 “preempt_disable()”，這個調用的功能是“關搶佔”（在 spin_unlock 中會重新開啓搶佔功能）。從中可以看出，使用自旋鎖保護的區域是工作在非搶佔的狀態；即使獲取不到鎖，在“自旋” 狀態也是禁止搶佔的。瞭解到這，我想咱們應該能夠理解爲何自旋鎖保護的代碼不能睡眠了。試想一下，如果在自旋鎖保護的代碼中間睡眠，此時發生進程調度，則可能另外一個進程會再次調用 spinlock 保護的這段代碼。而我們現在知道了即使在獲取不到鎖的 “自旋” 狀態，也是禁止搶佔的，而 “自旋” 又是動態的，不會再睡眠了，也就是說在這個處理器上不會再有進程調度發生了，那麼死鎖自然就發生了。

總結下自旋鎖的特點：

單 CPU 非搶佔內核下：自旋鎖會在編譯時被忽略（因爲單 CPU 且非搶佔模式情況下，不可能發生進程切換，時鐘只有一個進程處於臨界區（自旋鎖實際沒什麼用了）
單 CPU 搶佔內核下：自選鎖僅僅當作一個設置搶佔的開關（因爲單 CPU 不可能有併發訪問臨界區的情況，禁止搶佔就可以保證臨街區唯一被擁有）
多 CPU 下：此時才能完全發揮自旋鎖的作用，自旋鎖在內核中主要用來防止多處理器中併發訪問臨界區，防止內核搶佔造成的競爭。

四：linux 發生搶佔的時間

linux 搶佔發生的時間，搶佔分爲用戶搶佔和內核搶佔。

用戶搶佔在以下情況下產生：

從系統調用返回用戶空間
從中斷處理程序返回用戶空間

內核搶佔會發生在：

當從中斷處理程序返回內核空間的時候，且當時內核具有可搶佔性
當內核代碼再一次具有可搶佔性的時候（如：spin_unlock 時）
如果內核中的任務顯示的調用 schedule()

基本的進程調度就是發生在時鐘中斷後，並且發現進程的時間片已經使用完了，則發生進程搶佔。通常我們會利用中斷處理程序返回內核空間的時候可進行內核搶佔這個特性來提高一些 I/O 操作的實時性，如：當 I/O 事件發生的時候，對應的中斷處理程序被激活，當它發現有進程在等待這個 I/O 事件的時候，它會激活等待進程，並且設置當前正在執行進程的 need_resched 標誌，這樣在中斷處理程序返回的時候，調度程序被激活，原來在等待 I/O 事件的進程（很可能）獲得執行權，從而保證了對 I/O 事件的相對快速響應（毫秒級）。可以看出，在 I/O 事件發生的時候，I/O 事件的處理進程會搶佔當前進程，系統的響應速度與調度時間片的長度無關。

五：spin_lock 和 mutex 實際效率對比

1.++i 是否需要加鎖?

我分別使用 POSIX 的 spin_lock 和 mutex 寫了兩個累加的程序，啓動了兩個線程，並利用時間戳計算它們執行完累加所用的時間。

下面這個是使用 spin_lock 的代碼，我啓動兩個線程同時對 num 進行 ++，使用 spin_lock 保護臨界區，實際上可能會有疑問 ++i（++i 和 ++num 本文中是一個意思）爲什麼還要加鎖？

i++ 需要加鎖是很明顯的事情，對 i++ 的操作的印象是，它一般是三步曲，從內存中取出 i 放入寄存器中，在寄存器中對 i 執行 inc 操作，然後把 i 放回內存中。這三步明顯是可打斷的，所以需要加鎖。

但是 ++i 可能就有點猶豫了。實際上印象流是不行的，來看一下 i++ 和 ++i 的彙編代碼，其實他們是一樣的，都是三步，我只上一個圖就行了，如下：

所以 ++i 也不是原子操作，在多核的機器上，多個線程在讀取內存中的 i 時，可能讀取到同一個值，這就導致多個線程同時執行 + 1，但實際上它們得到的結果是一樣的，即 i 只加了一次。還有一點：這幾句彙編正說明了 ++i 和 i++i 對於效率是一樣的，不過這只是針對內建 POD 類型而言，如果是 class 的話，我們都寫過類的 ++ 運算符的重載，如果一個類在單個語句中不寫 ++i，而是寫 i++ 的話，那無疑效率會有很大的損耗。（有點跑題）

2.spin_lock 代碼

首先是 spin_lock 實現兩個線程同時加一個數，每個線程均 ++num，然後計算花費的時間。

#include <iostream>
#include <thread>
 
#include <pthread.h>
#include <sys/time.h>
#include <unistd.h>
 
int num = 0;
pthread_spinlock_t spin_lock;
 
int64_t get_current_timestamp()
{
    struct timeval now = {0, 0};
    gettimeofday(&now, NULL);
    return now.tv_sec * 1000 * 1000 + now.tv_usec;
}
 
void thread_proc()
{
    for(int i=0; i<100000000; ++i){
        pthread_spin_lock(&spin_lock);
        ++num;
        pthread_spin_unlock(&spin_lock);
    }   
}
 
int main()
{
    pthread_spin_init(&spin_lock, PTHREAD_PROCESS_PRIVATE);//maybe PHREAD_PROCESS_PRIVATE or PTHREAD_PROCESS_SHARED
 
    int64_t start = get_current_timestamp();
 
    std::thread t1(thread_proc), t2(thread_proc);
    t1.join();
    t2.join();
 
    std::cout<<"num:"<<num<<std::endl;
    int64_t end = get_current_timestamp();
    std::cout<<"cost:"<<end-start<<std::endl;
 
    pthread_spin_destroy(&spin_lock);
 
    return 0;
}

3.mutex 代碼

#include <iostream>
#include <thread>
 
#include <pthread.h>
#include <sys/time.h>
#include <unistd.h>
 
int num = 0;
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
 
int64_t get_current_timestamp()
{   
    struct timeval now = {0, 0}; 
    gettimeofday(&now, NULL);
    return now.tv_sec * 1000 * 1000 + now.tv_usec;
}
 
void thread_proc()
{
    for(int i=0; i<1000000; ++i){
        pthread_mutex_lock(&mutex);
        ++num;
        pthread_mutex_unlock(&mutex);
    }   
}
 
int main()
{
    int64_t start = get_current_timestamp();
   std::thread t1(thread_proc), t2(thread_proc);
    t1.join();
    t2.join();
    std::cout<<"num:"<<num<<std::endl;
    int64_t end = get_current_timestamp();
    std::cout<<"cost:"<<end-start<<std::endl;
 
    pthread_mutex_destroy(&mutex);    //maybe you always foget this
 
    return 0;
}

結果分析

得出的結果如圖，num 是最終結果，cost 是花費時間，單位爲 us，main2 是使用 spin lock，

顯然，在臨界區只有 ++num 這一條語句的情況下，spin lock 相對花費的時間短一些，實際上它們有可能接近的情況，取決於 CPU 的調度情況，但始終會是 spin lock 執行的效率在本情況中花費時間更少。

我修改了兩個程序中臨界區的代碼，改爲：

   for(int i=0; i<1000000; ++i){
        pthread_spin_lock(&spin_lock);
        ++num;
        for(int i=0; i<100; ++i){
            //do nothing
        }   
        pthread_spin_unlock(&spin_lock);
    }

另一個使用 mutex 的程序也加了這麼一段，然後結果就與之前的情況大相徑庭了：

實驗結果是如此的明顯，僅僅是在臨界區內加了一個 10 圈的循環，spin lock 就需要花費比 mutex 更長的時間了。

所以， spin lock 雖然 lock/unlock 的性能更好（花費很少的 CPU 指令），但是它只適應於臨界區運行時間很短的場景。實際開發中，程序員如果對自己程序的鎖行爲不是很瞭解，否則使用 spin lock 不是一個好主意。更保險的方法是使用 mutex，如果對性能有進一步的要求，那麼再考慮 spin lock。

六：使用 C++ 實現自主實現自旋鎖

由於前面原理已經很清楚了，現在直接給代碼如下：

#pragma once
 
#include <atomic>
 
class spin_lock {
private:
    std::atomic<bool> flag = ATOMIC_VAR_INIT(false);
public:
    spin_lock() = default;
    spin_lock(const spin_lock&) = delete;
    spin_lock& operator=(const spin_lock) = delete;
    void lock(){   //acquire spin lock
        bool expected = false;
        while(!flag.compare_exchange_strong(expected, true));
            expected = false;    
    }   
    void unlock(){   //release spin lock
        flag.store(false);
    }   
};

測試文件，僅給出關鍵部分：

int num = 0;
spin_lock sm; 
 
void thread_proc()
{
    for(int i=0; i<10000000; ++i){
        sm.lock();
        ++num;
        sm.unlock();
    }   
}

本文由 Readfog 進行 AMP 轉碼，版權歸原作者所有。
來源：https://mp.weixin.qq.com/s/rG_2uF9IpmquVCfp6SzD0g

猜你喜歡