操作系統下spinlock鎖解析、模擬及損耗分析

1. 關於spinlock

   我們在知道什么是spinlock之前，還需要知道為什么需要這個spinlock?
   spinlock本質就是鎖，提到鎖，我們就回到了多線程編程的混沌初期，為了實現多線程編程，操作系統引入了鎖。通過鎖能夠保證在多核多線程情況下，對臨界區資源進行保護，從而保證操作數據的一致性。

2. 鎖

   那么我們來溫習下操作系統中5個知名的鎖概念，每個技術都有適合自己的應用場景，此處引入介紹不再進一步深入展開。

• 信號量(Semaphore)

  Linux中的信號量是一種睡眠鎖。如有一個任務試圖獲得一個已被持有的信號量時，信號量會將其推進等待隊列，然后讓其睡眠。當持有信號量的進程將信號量開釋后，在等待隊列中的一個任務將被喚醒，從而便可以獲得這個信號量。
  信號量分為二元信號量和多元信號量，所謂二元信號量就是指該信號量只有兩個狀態，要么被占用，要么空閑;而多元信號量則允許同時被N個線程占有，超出N個外的占用請求將被阻塞。信號量是“系統級別”的，即同一個信號量可以被不同的進程訪問。

• 互斥量 (Mutex)

  和二元信號量類似,不同的是互斥量的獲取和釋放必須是在同一個線程中進行的。如果一個線程不能去釋放一個並不是它所占有的互斥量。而信號量是可以由其它線程進行釋放的。

• 臨界區(Critical Section)

  把獲取臨界區的鎖稱為進入臨界區，而把鎖的釋放稱為離開臨界區。Spinlock就是為了保護這臨界區。

• 讀寫鎖(Read-Write Lock)

  如果程序大部分時間都是在讀取，使用前面的鎖時，每次讀也要申請鎖的，會導致其他線程就無法再對此段數據進行同步讀取。我們知道對數據進行讀取時，不存在數據同步問題的，那么這些讀鎖就影響了程序的性能。讀寫鎖的出現就是為了解決這個問題的。
  讀寫鎖，有兩種獲取方式：共享(Shared)或獨占 (Exclusive)。如果當前讀寫鎖處於空閑狀態，那么當多個線程同時以共享方式訪問該讀寫鎖時，都可以成功;而如果一個線程以獨占的方式訪問該讀寫鎖，那么它會等待所有共享訪問都結束后才可以成功。在讀寫鎖被獨占的過程中，再次共享和獨占請求訪問該鎖，都會進行等待狀態。

• 條件變量(Condition Variable)

  條件變量相當於一種通知機制。多個線程可以設置等待該條件變量，一旦另外的線程設置了該條件變量(相當於喚醒條件變量)后，多個等待的線程就可以繼續執行了。
  以上是操作系統相關的幾個概念，信號量也好互斥量也罷，只是不同的手段來實現資源的保護，實際還是根據真實應用需求的來選擇。

3. Spinlock

   我們來看下spinlock， spinlock叫做自旋鎖，最初針對SMP系統，實現在SMP多處理器情況下臨界區保護。
   主要作用是給臨界數據加鎖，從而保護臨界數據不被同時訪問，實現多任務的同步。如果臨界數據當前不可訪問，那么就自旋直到可以訪問為止。
   自旋鎖和互斥鎖存在差異的是自旋鎖不會引起調用者睡眠，如果自旋鎖無法獲取，那么調用者一直循環檢測自旋鎖直到釋放。
   spinlock的工作方式本身就體現了它的優缺點，優點是執行速度快，不涉及上下文切換；缺點是耗費CPU資源。
   在Linux內核中，自旋鎖通常用於包含內核數據結構的操作，可以看到在許多內核數據結構中都嵌入有spinlock，這些大部分就是用於保證它自身被操作的原子性（原子操作atomic operation為"不可被中斷的一個或一系列操作"，最后其實是通過底層硬件來保證的），在操作這樣的結構體時都經歷這樣的過程：上鎖-操作-解鎖。
   因為在現代處理器系統中，考慮到中斷、內核搶占以及其他處理器的訪問，所以spinlock自旋鎖應該阻止在代碼運行過程中出現的其他並發干擾。

• 中斷

  中斷會觸發中斷例程的執行，如果中斷例程訪問了臨界區，這就可能會有大量中斷進來不斷觸發中斷例程來進入臨界區，那么臨界區的原子性就被打破了。所以如果在中斷例程中存在訪問某個臨界區的代碼，那么就必須用中斷禁用spinlock保護。（不同的中斷類型（硬件中斷和軟件中斷）對應於不同版本的自旋鎖實現）

• 內核搶占

  我們知道Linux內核在2.6以后，支持內核搶占。這種情況下進入臨界區就需要避免因內核搶占造成的並發，使用禁用搶占（preempt_disable()）的spinlock，結束后再開啟搶占（preempt_enable()）。

• 多處理器的訪問

  在多個物理處理器系統，肯定會有多個進程的並發訪問內存。這樣就需要在內存加一個標志，每個需要進入臨界區的代碼都必須檢查這個標志，看是否有進程已經在這個臨界區中。當然每個系統都有一套自己的實現方案。
  其實在內核代碼中針對以上幾點都設計了針對的spinlock版本，開發者只要根據不同場景選擇不同版本即可。

4. 內核代碼定義

   與spinlock 相關的文件可以查看內核源碼中的include/linux文件夾，主要是include/linux/spinlock.h提供spinlock通用的spinlock和rwlock申明。定義了不同的spinlock版本，例如,以下下函數均定義在spinlock.h文件中。
   如果在中斷例程中不會操作臨界區，可以用如下版本
   static __always_inline void spin_lock(spinlock_t *lock)
   static __always_inline void spin_unlock(spinlock_t *lock)
   在軟件中斷中操作臨界區使用如下spinlock版本：
   static inline void spin_lock_bh(spinlock_t *lock)
   static inline void spin_unlock_bh(spinlock_t *lock)
   如果在硬件中斷中操作臨界區使用如下spinlock版本：
   static inline void spin_lock_irq(spinlock_t *lock)
   static inline void spin_unlock_irq(spinlock_t *lock)
   如果在控制硬件中斷的時候需要同時保存中斷狀態使用如下spinlock版本：
   spin_lock_irqsave(lock, flags)
   spin_unlock_irqrestore(spinlock_t *lock, unsigned long flags)
   獲得自旋鎖和釋放自旋鎖有好幾個版本，對開發同學來說知道什么樣的情況下使用什么版本的獲得和釋放鎖的宏是非常必要的。
   例如：如果被保護的臨界資源只在進程上下文訪問和軟中斷（包括tasklet、timer）中訪問，那么對於這種情況，對共享資源的訪問必須使用spin_lock_bh和spin_unlock_bh來處理。不過使用spin_lock_irq和spin_unlock_irq以及spin_lock_irqsave和spin_unlock_irqrestore也可以，它們會同時失效本地硬中斷，隱式地也失效了軟中斷。但是使用spin_lock_bh和spin_unlock_bh是最合適的，它比其他兩個快。
   spin_lock阻止在不同CPU上的執行單元對共享資源的同時訪問以及不同進程上下文互相搶占導致的對共享資源的非同步訪問，而中斷失效和軟中斷失效卻是為了阻止在同一CPU上軟中斷或中斷對共享資源的非同步訪問。
   此外根據內核配置中CONFIG_SMP和CONFIG_DEBUG_SPINLOCK，會定義不同的函數,如下:
   #if defined(CONFIG_SMP) || defined(CONFIG_DEBUG_SPINLOCK)
   # include <linux/spinlock_api_smp.h>
   #else
   # include <linux/spinlock_api_up.h>
   #endif

5. 關於spinlock注意點

   a) 因為spin_lock的使用會浪費CPU資源（因為busy_loop），為了盡可能地消除spin_lock的負面影響，使用spin_lock保護臨界區代碼盡可能精煉，確保能盡早從臨界區出來。
   b)如果臨界區可能包含引起睡眠的代碼則不能使用自旋鎖，否則可能引起死鎖。萬一進程在臨界區引發睡眠后，那么后面嗷嗷待哺的那些正在spinlock的進程咋辦，它們正等着進入臨界區呢？等到猴年馬月呢，就觸發死鎖了。
   c)此外頻繁的檢測鎖會讓流水線上充滿讀操作引起CPU對流水線的重排，從而進一步影響性能。如果在循環的中加個pause指令，讓cpu暫停N個周期，則可以釋放cpu的一些計算資源，讓同一個核心上的另一個超線程使用，提升性能。針對這塊可以翻閱Intel的官方材料：
   64-ia-32-architectures-optimization-manual.pdf
   以上內容介紹了操作系統中的鎖的類型、種類，以及spinlock鎖的工作機制和注意點。下面我們聚焦在於當系統中出現spinlock高的時候如何找到問題元凶。

6. spinlock問題模擬

   在操作系統中模擬spinlock我們借助POSIX threads。這個是在多核平台上進行並行編程的一套常用的API。Pthreads提供了多種鎖機制：
   (1) Mutex（互斥量）：pthread_mutex_*** (2) Spin lock（自旋鎖）：pthread_spin_*** (3) Condition Variable（條件變量）：pthread_con_*** (4) Read/Write lock（讀寫鎖）：pthread_rwlock_*** 首先定義一個自旋鎖：spinlock_t x;
   然后初始化：spin_lock_init(spinlock_t *x); //自旋鎖在使用前需要先初始化
   使用后銷毀它：spin_destroy(&lock);
   當然不過不想用這些API，可以自己實現spinlock，然后再調用之也行。
   #include 
   #include 
   #include <pthread.h>
   #include <sys/time.h>
   #include <unistd.h>
   int numcount = 0;
   pthread_spinlock_t lock;
   using namespace std;
   void thread_proc()
   {
   for (int i = 0; i < 100000000; ++i) {
   pthread_spin_lock(&lock);
   ++numcount;
   pthread_spin_unlock(&lock);
   }
   }
   int main()
   {
   pthread_spin_init(&lock, PTHREAD_PROCESS_PRIVATE); //maybe PHREAD_PROCESS_PRIVATE or PTHREAD_PROCESS_SHARED
   std::thread t1(thread_proc);
   t1.join();
   std::cout << "numcount:" << numcount << std::endl;
   pthread_spin_destroy(&lock);
   return 0;
   }
   編譯 #g++ -std=c++11 -lpthread spinlock_t.cpp
   執行（循環時間可以增大便於監控）之后我們發現：
   通過perf top可以看到95.11%是pthread_spin_lock.
   這是我們代碼中定義的pthread_spin_lock函數，該函數就是在我們使用使用的庫libpthread，就樣因果關系就對應起來了。
   不過為什么在sys中使用率幾乎是0%呢？
   因為我們在代碼中保護的是用戶層的數據，並沒有切入到內核態。如我們在前面所描述，Linux中自旋鎖是用於保護內核數據結構的，當到內核態時候不停自鎖就會被監控命令累積到sys上了。
   知道函數之后，就可以在源碼中找到對應的代碼位置進行分析了。
   此外可以使用pstack和gdb工具。
   用pstack可以顯示進程的棧跟蹤,用來來確定進程掛起的位置。
   GDB是GNU開源組織發布的程序調試工具，用來調試 C 和 C++ 程序。可以使程序開發者在程序運行時觀察程序的內部結構和內存的使用情況。

7. spinlock損耗

   然后在代碼中加上時間戳后，對比測試了一組數據。
   線程數量從1個線程計算增加到40個線程，每個線程的工作內容為累加到1000千萬，40線程會將結果累積到4億。計算每次i增加到i+1的平均時間，就可以理解成spinlock的損耗。我們發現時間變化如下，其中橫坐標為線程數量，縱坐標為每次加法的時間損耗，單位為us。
   
   我們發現在40線程下每次加法消耗的時間要比1個線程下每次加法消耗高出幾十倍來，雖然投入的CPU資源增加了，但是更多的是在spinlock上消耗了。
   通過這樣一組實驗，對spinlock損耗有進一步的認識，並可得出這樣一個結論：當一個臨界資源被更多的線程共享爭用時候，在並發增加時會導致平均每次操作的時間損耗增加。
   所以在一個共享資源爭用厲害的業務場景，在不優化爭用資源的情況下，一直增加負載反會讓業務響應性能更差。

8. 小結

   因為以上問題是基於自身模擬的問題，所以在定位思路上難免有作弊之嫌疑。不過通過了解spinlock，並深入如何使用spinlock之后，對自旋鎖本身有了更深刻的認識。后續我們看到spinlock情況的時候可以更加大膽的來找問題原因了。

9. 參考鏈接

http://m.blog.csdn.net/maray/article/details/8757030
http://cyningsun.github.io/06-01-2016/nehalem-arch.html
http://kb.cnblogs.com/page/105657/
Intel® 64 and IA-32 Architectures Optimization Reference Manual
http://blog.csdn.net/freeelinux/article/details/53695111
http://blog.chinaunix.net/uid-28711483-id-4995776.html
http://blog.chinaunix.net/uid-21411227-id-1826888.html
http://blog.poxiao.me/p/spinlock-implementation-in-cpp11/
https://www.intel.com/content/www/us/en/architecture-and-technology/64-ia-32-architectures-software-developer-manual-325462.html
http://blog.chinaunix.net/uid-20543672-id-3252604.html
http://blog.csdn.net/sahusoft/article/details/9265533

10. 附錄spinlock C++實現

#include 
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;
int
main ()
{
for (int i = 0; i < 10000000; ++i)
{
sm.lock ();
++num;
sm.unlock ();
}
return 0;
}
編譯命令：g++ -std=c++11 –lpthread **.cpp

附錄模擬mutex
mutex的使用方法和spinlock大同小異。
#include 
#include 
#include <pthread.h>
#include <sys/time.h>
#include <unistd.h>
int num = 0;
pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
void thread_proc ()
{
    for (int i = 0; i < 1000000; ++i)
    {
        pthread_mutex_lock (&mutex);
        ++num;
        pthread_mutex_unlock (&mutex);
    }
}
int
main ()
{
    std::thread t1 (thread_proc);
    t1.join ();
    std::cout << "num:" << num << std::endl;
    pthread_mutex_destroy (&mutex); //maybe you always     foget this
    return 0;
}

 

編譯命令：
g++ -std=c++11 -lpthread mutex.cpp

12. 附錄加上時間戳

#include 
#include 
#include <pthread.h>
#include <sys/time.h>
#include <unistd.h>
int numcount = 0;
pthread_spinlock_t lock;
using namespace std;
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(&lock);
++numcount;
pthread_spin_unlock(&lock);
}
}

int main()
{
pthread_spin_init(&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 << "numcount:" << numcount << std::endl;
int64_t end = get_current_timestamp();
std::cout << "cost:" << end - start << std::endl;
pthread_spin_destroy(&lock);
return 0;
}