作業系統下spinlock鎖解析、模擬及損耗分析
http://linuxperformance.top/index.php/archives/46/
-
關於spinlock
我們在知道什麼是spinlock之前，還需要知道為什麼需要這個spinlock?
spinlock本質就是鎖，提到鎖，我們就回到了多執行緒程式設計的混沌初期，為了實現多執行緒程式設計，作業系統引入了鎖。通過鎖能夠保證在多核多執行緒情況下，對臨界區資源進行保護，從而保證操作資料的一致性。
-
鎖
那麼我們來溫習下作業系統中5個知名的鎖概念，每個技術都有適合自己的應用場景，此處引入介紹不再進一步深入展開。
-
訊號量(Semaphore)
Linux中的訊號量是一種睡眠鎖。如有一個任務試圖獲得一個已被持有的訊號量時，訊號量會將其推進等待佇列，然後讓其睡眠。當持有訊號量的程序將訊號量開釋後，在等待佇列中的一個任務將被喚醒，從而便可以獲得這個訊號量。
訊號量分為二元訊號量和多元訊號量，所謂二元訊號量就是指該訊號量只有兩個狀態，要麼被佔用，要麼空閒;而多元訊號量則允許同時被N個執行緒佔有，超出N個外的佔用請求將被阻塞。訊號量是“系統級別”的，即同一個訊號量可以被不同的程序訪問。 -
互斥量 (Mutex)
和二元訊號量類似,不同的是互斥量的獲取和釋放必須是在同一個執行緒中進行的。如果一個執行緒不能去釋放一個並不是它所佔有的互斥量。而訊號量是可以由其它執行緒進行釋放的。
-
臨界區(Critical Section)
把獲取臨界區的鎖稱為進入臨界區，而把鎖的釋放稱為離開臨界區。Spinlock就是為了保護這臨界區。
-
讀寫鎖(Read-Write Lock)
如果程式大部分時間都是在讀取，使用前面的鎖時，每次讀也要申請鎖的，會導致其他執行緒就無法再對此段資料進行同步讀取。我們知道對資料進行讀取時，不存在資料同步問題的，那麼這些讀鎖就影響了程式的效能。讀寫鎖的出現就是為了解決這個問題的。
讀寫鎖，有兩種獲取方式：共享(Shared)或獨佔 (Exclusive)。如果當前讀寫鎖處於空閒狀態，那麼當多個執行緒同時以共享方式訪問該讀寫鎖時，都可以成功;而如果一個執行緒以獨佔的方式訪問該讀寫鎖，那麼它會等待所有共享訪問都結束後才可以成功。在讀寫鎖被獨佔的過程中，再次共享和獨佔請求訪問該鎖，都會進行等待狀態。 -
條件變數(Condition Variable)
條件變數相當於一種通知機制。多個執行緒可以設定等待該條件變數，一旦另外的執行緒設定了該條件變數(相當於喚醒條件變數)後，多個等待的執行緒就可以繼續執行了。
以上是作業系統相關的幾個概念，訊號量也好互斥量也罷，只是不同的手段來實現資源的保護，實際還是根據真實應用需求的來選擇。
-
Spinlock
我們來看下spinlock， spinlock叫做自旋鎖，最初針對SMP系統，實現在SMP多處理器情況下臨界區保護。
主要作用是給臨界資料加鎖，從而保護臨界資料不被同時訪問，實現多工的同步。如果臨界資料當前不可訪問，那麼就自旋直到可以訪問為止。
自旋鎖和互斥鎖存在差異的是自旋鎖不會引起呼叫者睡眠，如果自旋鎖無法獲取，那麼呼叫者一直迴圈檢測自旋鎖直到釋放。
spinlock的工作方式本身就體現了它的優缺點，優點是執行速度快，不涉及上下文切換；缺點是耗費CPU資源。
在Linux核心中，自旋鎖通常用於包含核心資料結構的操作，可以看到在許多核心資料結構中都嵌入有spinlock，這些大部分就是用於保證它自身被操作的原子性（原子操作atomic operation為"不可被中斷的一個或一系列操作"，最後其實是通過底層硬體來保證的），在操作這樣的結構體時都經歷這樣的過程：上鎖-操作-解鎖。
因為在現代處理器系統中，考慮到中斷、核心搶佔以及其他處理器的訪問，所以spinlock自旋鎖應該阻止在程式碼執行過程中出現的其他併發干擾。
-
中斷
中斷會觸發中斷例程的執行，如果中斷例程訪問了臨界區，這就可能會有大量中斷進來不斷觸發中斷例程來進入臨界區，那麼臨界區的原子性就被打破了。所以如果在中斷例程中存在訪問某個臨界區的程式碼，那麼就必須用中斷禁用spinlock保護。（不同的中斷型別（硬體中斷和軟體中斷）對應於不同版本的自旋鎖實現）
-
核心搶佔
我們知道Linux核心在2.6以後，支援核心搶佔。這種情況下進入臨界區就需要避免因核心搶佔造成的併發，使用禁用搶佔（preempt_disable()）的spinlock，結束後再開啟搶佔（preempt_enable()）。
-
多處理器的訪問
在多個物理處理器系統，肯定會有多個程序的併發訪問記憶體。這樣就需要在記憶體加一個標誌，每個需要進入臨界區的程式碼都必須檢查這個標誌，看是否有程序已經在這個臨界區中。當然每個系統都有一套自己的實現方案。
其實在核心程式碼中針對以上幾點都設計了針對的spinlock版本，開發者只要根據不同場景選擇不同版本即可。
-
核心程式碼定義
與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 -
關於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高的時候如何找到問題元凶。 -
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++ 程式。可以使程式開發者在程式執行時觀察程式的內部結構和記憶體的使用情況。 -
spinlock損耗
然後在程式碼中加上時間戳後，對比測試了一組資料。
執行緒數量從1個執行緒計算增加到40個執行緒，每個執行緒的工作內容為累加到1000千萬，40執行緒會將結果累積到4億。計算每次i增加到i+1的平均時間，就可以理解成spinlock的損耗。我們發現時間變化如下，其中橫座標為執行緒數量，縱座標為每次加法的時間損耗，單位為us。
我們發現在40執行緒下每次加法消耗的時間要比1個執行緒下每次加法消耗高出幾十倍來，雖然投入的CPU資源增加了，但是更多的是在spinlock上消耗了。
通過這樣一組實驗，對spinlock損耗有進一步的認識，並可得出這樣一個結論：當一個臨界資源被更多的執行緒共享爭用時候，在併發增加時會導致平均每次操作的時間損耗增加。
所以在一個共享資源爭用厲害的業務場景，在不優化爭用資源的情況下，一直增加負載反會讓業務響應效能更差。 -
小結
因為以上問題是基於自身模擬的問題，所以在定位思路上難免有作弊之嫌疑。不過通過了解spinlock，並深入如何使用spinlock之後，對自旋鎖本身有了更深刻的認識。後續我們看到spinlock情況的時候可以更加大膽的來找問題原因了。
-
參考連結
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
- 附錄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
- 附錄加上時間戳
#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;
}