1. 程式人生 > >自旋鎖與互斥鎖的對比、手工實現自旋鎖

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

               

    本文之前,我只是對自旋鎖有所瞭解,知道它是做什麼的,但是沒有去測試實現過,甚至以為自旋鎖只有kernel用這個,今天才發現POSIX有提供自旋鎖的介面。下面我會分析一下自旋鎖,並程式碼實現自旋鎖和互斥鎖的效能對比,以及利用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核心程序排程關係

    現在我們就來說一說之前的問題,如果臨界區可能包含引起睡眠的程式碼則不能使用自旋鎖,否則可能引起死鎖:     那麼為什麼訊號量保護的程式碼可以睡眠而自旋鎖會死鎖呢?     先看下自旋鎖的實現方法吧,自旋鎖的基本形式如下:
  1.     spin_lock(&mr_lock):
  2.     //critical region
  3.     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,然後計算花費的時間。
  1. #include <iostream>
  2. #include <thread>
  3. #include <pthread.h>
  4. #include <sys/time.h>
  5. #include <unistd.h>
  6. int num = 0;
  7. pthread_spinlock_t spin_lock;
  8. int64_t get_current_timestamp()
  9. {
  10.     struct timeval now = {0, 0};
  11.     gettimeofday(&now, NULL);
  12.     return now.tv_sec * 1000 * 1000 + now.tv_usec;
  13. }
  14. void thread_proc()
  15. {
  16.     for(int i=0; i<100000000; ++i){
  17.         pthread_spin_lock(&spin_lock);
  18.         ++num;
  19.         pthread_spin_unlock(&spin_lock);
  20.     }  
  21. }
  22. int main()
  23. {
  24.     pthread_spin_init(&spin_lock, PTHREAD_PROCESS_PRIVATE);//maybe PHREAD_PROCESS_PRIVATE or PTHREAD_PROCESS_SHARED
  25.     int64_t start = get_current_timestamp();
  26.     std::thread t1(thread_proc), t2(thread_proc);
  27.     t1.join();
  28.     t2.join();
  29.     std::cout<<"num:"<<num<<std::endl;
  30.     int64_t end = get_current_timestamp();
  31.     std::cout<<"cost:"<<end-start<<std::endl;
  32.     pthread_spin_destroy(&spin_lock);
  33.     return 0;
  34. }

3.mutex程式碼

  1. #include <iostream>
  2. #include <thread>
  3. #include <pthread.h>
  4. #include <sys/time.h>
  5. #include <unistd.h>
  6. int num = 0;
  7. pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
  8. int64_t get_current_timestamp()
  9. {  
  10.     struct timeval now = {0, 0};
  11.     gettimeofday(&now, NULL);
  12.     return now.tv_sec * 1000 * 1000 + now.tv_usec;
  13. }
  14. void thread_proc()
  15. {
  16.     for(int i=0; i<1000000; ++i){
  17.         pthread_mutex_lock(&mutex);
  18.         ++num;
  19.         pthread_mutex_unlock(&mutex);
  20.     }  
  21. }
  22. int main()
  23. {
  24.     int64_t start = get_current_timestamp();
  25.    std::thread t1(thread_proc), t2(thread_proc);
  26.     t1.join();
  27.     t2.join();
  28.     std::cout<<"num:"<<num<<std::endl;
  29.     int64_t end = get_current_timestamp();
  30.     std::cout<<"cost:"<<end-start<<std::endl;
  31.     pthread_mutex_destroy(&mutex);    //maybe you always foget this
  32.     return 0;
  33. }

4.結果分析

得出的結果如圖,num是最終結果,cost是花費時間,單位為us,main2是使用spin lock,     顯然,在臨界區只有++num這一條語句的情況下,spin lock相對花費的時間短一些,實際上它們有可能接近的情況,取決於CPU的排程情況,但始終會是spin lock執行的效率在本情況中花費時間更少。     我修改了兩個程式中臨界區的程式碼,改為:
  1.    for(int i=0; i<1000000; ++i){
  2.         pthread_spin_lock(&spin_lock);
  3.         ++num;
  4.         for(int i=0; i<100; ++i){
  5.             //do nothing
  6.         }  
  7.         pthread_spin_unlock(&spin_lock);
  8.     }  
    另一個使用mutex的程式也加了這麼一段,然後結果就與之前的情況大相徑庭了:     實驗結果是如此的明顯,僅僅是在臨界區內加了一個10圈的迴圈,spin lock就需要花費比mutex更長的時間了。     所以,spin lock雖然lock/unlock的效能更好(花費很少的CPU指令),但是它只適應於臨界區執行時間很短的場景。實際開發中,程式設計師如果對自己程式的鎖行為不是很瞭解,否則使用spin lock不是一個好主意。更保險的方法是使用mutex,如果對效能有進一步的要求,那麼再考慮spin lock。

七:使用C++實現自主實現自旋鎖

由於前面原理已經很清楚了,現在直接給程式碼如下:
  1. #pragma once
  2. #include <atomic>
  3. class spin_lock {
  4. private:
  5.     std::atomic<bool> flag = ATOMIC_VAR_INIT(false);
  6. public:
  7.     spin_lock() = default;
  8.     spin_lock(const spin_lock&) = delete;
  9.     spin_lock& operator=(const spin_lock) = delete;
  10.     void lock(){   //acquire spin lock
  11.         bool expected = false;
  12.         while(!flag.compare_exchange_strong(expected, true));
  13.             expected = false;   
  14.     }  
  15.     void unlock(){   //release spin lock
  16.         flag.store(false);
  17.     }  
  18. };
測試檔案,僅給出關鍵部分: