當數據被多線程并發訪問(讀/寫)時,需要對數據加鎖。linux 內核中常用的鎖有兩類:自旋鎖和互斥體。在使用鎖的時候,最常見的 bug 是死鎖問題,死鎖問題很多時候比較難定位,并且影響較大。本文先會介紹兩種引起死鎖的原因,對比自旋鎖和互斥體的區別,最后記錄一下可以遞歸調用的鎖。本文通過內核模塊來展示鎖的使用。
鎖保護的是什么 ?
鎖保護的是數據,不是代碼。數據在代碼中要么是一個變量,要么是一個數組,一個鏈表,紅黑樹等。如下代碼所示,有一個全局靜態變量 counter,假如有兩個線程分別調用 inc() 和 dec() 函數對 counter 做遞增和遞減運算。這樣在兩個線程對 counter 進行修改之前就需要加鎖,修改之后解鎖。鎖保護的是 counter 這個數據,而不是 counter++ 或者 counter-- 這兩行代碼。
static int counter = 0;
void inc() {lock();counter++;unlock();
}void dec() {lock();counter--;unlock();
}
臨界區
臨界區指的是代碼段,從加鎖到解鎖的這段代碼就稱為臨界區。終于有一個概念對應的是代碼!上邊的代碼,counter++ 和 counter-- 這兩行代碼就是臨界區。
并發的任務有哪些
最常見的多任務是多個線程,多個線程訪問同一個資源,需要加鎖。除了線程,任務形式還包括中斷和軟中斷。線程和線程之間,線程和中斷之間,線程和軟軟中斷之間,中斷和中斷之間,分別有不同的鎖可以選用。
鎖對性能的影響
鎖對性能的影響要看臨界區所占的線程任務的比例以及并發度。如果并發度比較大,并且每個線程的主要邏輯都在臨界區里,這種情況下,鎖對性能的影響是比較大的,多線程并發的表現接近于多線程串行的性能。相反,如果并發度不高,并且臨界區所占邏輯比例比較小的話,那么對性能影響就會小一些。
在性能分析時,臨界區往往會成為程序的熱點。
延遲加鎖和兩次判斷
在工作中使用隊列,如果隊列有多個消費者,消費者在消費的時候往往要先判斷隊列是不是有數據,如果有數據則消費;否則直接返回。這種場景,往往有兩種加鎖的方式:
方式 1:
首先加鎖,然后判斷隊列中是不是有數據,有數據則消費,否則返回。
方式 2:
先不加鎖,而是先判斷隊列是不是有數據,如果有數據才加鎖。加鎖之后還要再進行一次判斷,如果有數據則消費,否則返回。
方式 1 少一次判斷,適用于隊列中經常有數據的場景,因為如果隊列中經常沒有數據,那么加鎖和解鎖的操作完全是浪費。方式 2 多一次判斷,適用于隊列經常沒有數據的場景,因為沒有數據就直接返回了,減少了加鎖和解鎖的操作。
// 方式 1
lock();if (!queue_empty()) {consume();}
unlock();// 方式 2
if (!queue_empty()) {lock();if (!queue_empty()) {consume();}unlock();
}
?
1 死鎖
死鎖是嚴重的 bug,造成死鎖的原因有兩個:自死鎖和 ABBA 鎖。
1.1 自死鎖
自死鎖,說的是一個線程已經獲取到了這個鎖,然后嘗試再次獲取同一個鎖。因為這個鎖已經被占用,第二次獲取的時候就要一直等。
如下內核模塊中,創建了一個內核線程,線程名是 lockThread,線程的入口函數是 thread_func,在這個函數中打印出了線程 id。定義了一個自旋鎖 lock,在線程中連續兩次調用 spin_lock() 來嘗試拿到鎖。第一次 spin_lock() 可以成功拿到鎖,第二次 spin_lock() 拿不到鎖,一直死等。
#include <linux/init.h>
#include <linux/module.h>
#include <linux/kernel.h>
#include <linux/kthread.h>
#include <linux/delay.h>
#include <linux/spinlock.h>MODULE_LICENSE("GPL");
MODULE_AUTHOR("Your Name");
MODULE_DESCRIPTION("Spinlock Example");spinlock_t lock;
static struct task_struct *thread;static int thread_func(void *data) {printk("tid = %d\n", current->pid);spin_lock(&lock);spin_lock(&lock);return 0;
}static int __init spinlock_example_init(void) {printk(KERN_INFO "Spinlock Example: Module init\n");spin_lock_init(&lock);thread = kthread_run(thread_func, NULL, "lockThread");return 0;
}static void __exit spinlock_example_exit(void) {printk(KERN_INFO "Spinlock Example: Module exit\n");kthread_stop(thread);
}module_init(spinlock_example_init);
module_exit(spinlock_example_exit);
如下是內核模塊運行情況,線程 id 是 4151。自旋鎖死鎖問題可以被 soft lockup 機制檢測出來并打印相關的信息。
soft lockup 異常檢測,參考如下鏈接:
[linux][異常檢測] hung task, soft lockup, hard lockup, workqueue stall
通過 top 命令可以看到線程 4151 的 cpu 占用情況。可以看到 cpu 占用率達到 100%,這也是自旋鎖的一個特點,使用自旋鎖在等待鎖的時候是一直自旋,也就是一直死循環,一直占著 cpu。
1.2 ABBA 鎖
讀 《明朝那些事》的時候看到了一個例子。朱元璋洪武二十年派兵討伐北元。當時明軍的實力相對于北元有絕對的優勢,當明軍找到北元的時候并沒有進攻,而是采取了勸降的策略。投降儀式在一個飯局進行,明軍設盛宴款待北元的納哈出。
死鎖的雙方是明軍的藍玉和北元的納哈出。下邊這段話是原文:
?????? 就在一切都順利進行的時候,藍玉的一個舉動徹底打破了這種和諧的氣氛。
?????? 當時納哈出正在向藍玉敬酒,大概也說了一些不喝酒就不夠兄弟的話,藍玉看納哈出的衣服破舊,便脫下了自己身上的衣服,要納哈出穿上。
?????? 應該說這是一個友好的舉動,但納哈出拒絕了,為什么呢 ?這就是藍玉的疏忽了,他沒有想到,自己和納哈出不是同一個民族,雙方的衣著習慣是不同的,雖然藍玉是好意,但在納哈出看來,這似乎是勝利者對失敗者的一種強求和恩賜。
?????? 藍玉以為對方客氣,便反復要求納哈出穿上,并表示納哈出不穿,他就不喝酒,而納哈出則順水推舟的表示,藍玉不喝,他就不穿這件衣服。
藍玉和納哈出這樣僵持下去,那么藍玉永遠不會喝納哈出的酒,納哈出也永遠不會穿藍玉的衣服。
有兩個鎖,鎖 A 和 鎖 B。有兩個線程,線程 1 和線程 2。線程 1 獲取到了鎖 A,與此同時線程 2 獲取到了鎖 B,然后線程 1 嘗試獲取鎖 B,線程 2 嘗試獲取鎖 A。因為鎖 A 被線程 1 持有,所以線程 2 無法獲取到鎖 A,只能一直等待;鎖 B 被線程 2 持有,所以線程 1 也只能一直等待。這樣兩個線程都獲取不到鎖,只能死等。這就是死鎖。
為避免死鎖,在開發中應該遵守一些簡單的規則:
① 不要重復請求同一個鎖,防止自死鎖
② 按順序加鎖
如果在代碼中要用多個鎖,那么要按相同的順序來加鎖,這樣可以避免 ABBA 鎖。如上面這張圖,線程 1 和線程 2 首先都獲取鎖 A 或者都獲取鎖 B,這樣就不會產生死鎖了
③ 防止發生饑餓,保證臨界區的代碼是可以執行結束的,而不是一直執行,導致鎖不會被釋放
④ 按順序釋放鎖,以加鎖的逆順序來釋放鎖,比如加鎖的時候先加鎖 A 再加鎖 B,那么釋放鎖的時候盡量先釋放鎖 B,再釋放鎖 A
2 自旋鎖與互斥體
2.1 自旋鎖
2.1.1 自旋鎖
spinlock_t
自旋鎖的數據類型,使用 spinlock_t 可以聲明一個自旋鎖
spin_lock_init()
初始化自旋鎖
spin_lock(), spin_unlock()
自旋鎖加鎖,自旋鎖解鎖。數據被多線程共享時,可以使用這兩個函數加鎖和解鎖。
spin_lok_irq(),spin_unlock_irq()
關閉本地中斷,加鎖;解鎖并開啟本地中斷。當數據在線程和中斷之間共享時,需要使用這兩個函數來加鎖和解鎖。因為中斷可以打斷線程的執行,試想,如果在線程中加鎖的時候沒有關中斷,那么在臨界區的時候,線程可能被中斷打斷,這個時候中斷處理程序想要獲取鎖,只能死等,因為這個時候鎖被線程拿著,還沒有釋放,這樣就產生了死鎖。
spin_lock_irqsave() 和 spin_unlock_irqrestore() 這兩個函數也是關中斷加鎖和解鎖開中斷。不同的是,加鎖的時候會保存當前的中斷狀態,解鎖的時候會將中斷恢復到加鎖時的狀態。
spin_lock_bh() 和 spin_unlock_bh()
關下半部加鎖,解鎖并開下半部。適用于線程和軟中斷并發的這種場景,因為軟中斷也可以打斷線程的執行,所以線程中在加鎖的同時需要關閉下半部。
并發場景 | 鎖的選用 |
線程和線程 | 普通的加鎖方式 沒必要關中斷,也沒必要關軟中斷 |
線程和中斷 | 關中斷加鎖 在線程中需要關中斷加鎖,在中斷中可以使用普通加鎖 |
線程和軟中斷 | 關下半部加鎖 在線程中可以關軟中斷加鎖,在軟中斷中可以使用普通加鎖 |
中斷和中斷 | 在 linux 下,中斷不會搶中斷,所以可以使用普通的加鎖方式 |
中斷和軟中斷 | 關中斷加鎖 |
軟中斷和軟中斷 | 普通加鎖 |
下半部并發的情況,主要考慮軟中斷和 tasklet。
對于軟中斷來說,如果數據只在軟中斷之間共享,那么只需要普通加鎖就可以了。因為在一個 cpu 上,一個正在運行的軟中斷不會被另一個軟中斷打斷,所以沒有必要關下半部。線程和中斷并發的時候,之所以需要在線程中關中斷加鎖,是因為中斷可以打斷線程;同樣的,關下半部加鎖,也是因為下半部可以搶占線程。
tasklet 與軟中斷不同。同一個軟中斷可以在多個 cpu 上并發執行,同一個 tasklet 只會在一個 cpu 上執行,所以如果數據只在一個 tasklet 中訪問,那么不需要加鎖。如果數據在不同的 tasklet 中共享,那么就需要加鎖,普通加鎖就可以,因為一個 cpu 上正在執行一個 tasklet 的時候,不會被另一個 tasklet 打斷。
tasklet
軟中斷
2.1.2 讀寫自旋鎖
有時候,對數據的訪問可以明確的分為兩個場景,讀和寫。比如,對一個鏈表可能既要更新,又要查詢。當更新鏈表時,那么更新操作是互斥的,在同一個時刻,只能有一個更新者,并且在更新的時候不能讀。如果此時沒有更新,而只有讀者,那么多個讀者是可以同時進行的。
讀寫自旋鎖的互斥規則:
(1)寫和寫互斥
(2)寫和讀互斥
(3)讀和讀不互斥
自旋鎖是完全互斥。在一些場景下,比如讀多寫少的場景,相對于使用自旋鎖,讀寫鎖可以帶來性能上的優化。
讀寫自旋鎖,在等待鎖的時候也是自旋,一直占著 cpu,這點與自旋鎖是類似的。
與自旋鎖不同的是,讀寫自旋鎖是可以遞歸調用的,也就是說一個線程調用一次之后,還可以再次調用。當然,實際使用中很少有這么用的,對于一個線程來說,重復獲取一個讀自旋鎖,毫無意義。
如下內核模塊有兩個線程,一個線程中獲取讀鎖,一個線程中獲取寫鎖。安裝內核模塊之后的打印結果截圖貼在了下邊。從打印信息可以看出來兩點:讀鎖可以多次獲取;讀鎖和寫鎖是互斥的,只有兩次釋放讀鎖之后,寫鎖才可以被獲取到。
#include <linux/init.h>
#include <linux/module.h>
#include <linux/kernel.h>
#include <linux/kthread.h>
#include <linux/delay.h>
#include <linux/spinlock.h>MODULE_LICENSE("GPL");
MODULE_AUTHOR("Your Name");
MODULE_DESCRIPTION("Spinlock Example");rwlock_t lock;static struct task_struct *rthread;
static struct task_struct *wthread;static int rthread_func(void *data) {printk("rthread tid = %d\n", current->pid);read_lock(&lock);printk("first got read lock\n");read_lock(&lock);printk("second got read lock\n");read_unlock(&lock);printk("first release read lock\n");mdelay(2000);read_unlock(&lock);printk("second release read lock\n");return 0;
}static int wthread_func(void *data) {printk("wthread tid = %d\n", current->pid);printk("getting write lock\n");mdelay(2000);write_lock(&lock);printk("got write lock\n");write_unlock(&lock);return 0;
}static int __init spinlock_example_init(void) {printk(KERN_INFO "Spinlock Example: Module init\n");rwlock_init(&lock);rthread = kthread_run(rthread_func, NULL, "rThread");wthread = kthread_run(wthread_func, NULL, "wThread");return 0;
}static void __exit spinlock_example_exit(void) {printk(KERN_INFO "Spinlock Example: Module exit\n");kthread_stop(rthread);kthread_stop(wthread);
}module_init(spinlock_example_init);
module_exit(spinlock_example_exit);
2.2 互斥體
互斥體,從名字就可以看出來能夠起到互斥的作用。互斥體和自旋鎖是兩種典型的同步機制,分別有各自的使用場景。一個事物往往會分為兩個方面,有兩種實現方式,這兩種方式有各自的使用場景,比如通信模型中的傳輸層,有 udp 和 tcp:tcp 是面向連接的可靠的字節流;而 udp 正好相反,沒有連接,不可靠,數據報。udp 和 tcp 分別有各自的使用場景。
互斥體在 linux 內核中是一個 struct mutex 結構體。如下內核模塊中,聲明了一個互斥體 struct mutex mtx。創建了兩個內核線程 thread1 和 thread2,在 thread1 中首先獲取了 mtx,然后嘗試再次獲取 mtx,這種情況下會產生死鎖,因為 mutex 不能遞歸調用;thread2 中首先 delay 了 2s,之所以 delay,是為了先讓 thread1 獲取鎖,這樣 thread2 嘗試獲取鎖就會獲取失敗,一直等待。
#include <linux/init.h>
#include <linux/module.h>
#include <linux/kernel.h>
#include <linux/kthread.h>
#include <linux/delay.h>
#include <linux/spinlock.h>
#include <linux/mutex.h>MODULE_LICENSE("GPL");
MODULE_AUTHOR("Your Name");
MODULE_DESCRIPTION("Spinlock Example");struct mutex mtx;static struct task_struct *thread1;
static struct task_struct *thread2;static int thread1_func(void *data) {printk("thread1 tid = %d\n", current->pid);mutex_lock(&mtx);printk("first got mutex\n");mutex_lock(&mtx);return 0;
}static int thread2_func(void *data) {printk("wthread tid = %d\n", current->pid);printk("getting mutex\n");mdelay(2000);mutex_lock(&mtx);return 0;
}static int __init spinlock_example_init(void) {printk(KERN_INFO "Spinlock Example: Module init\n");mutex_init(&mtx);thread1 = kthread_run(thread1_func, NULL, "Thread1");thread2 = kthread_run(thread2_func, NULL, "Thread2");return 0;
}static void __exit spinlock_example_exit(void) {printk(KERN_INFO "Spinlock Example: Module exit\n");kthread_stop(thread1);kthread_stop(thread2);
}module_init(spinlock_example_init);
module_exit(spinlock_example_exit);
使用 dmesg 可以查看內核模塊的打印信息,兩個線程的線程號分別是 2937 和 2938。
使用 top 分別查看兩個線程的狀態,可以看到兩個線程的狀態均是 D 狀態,并且 cpu 使用率為 0%。這是 mutex 和 spinlock 之間的主要區別,spinlock 在等待瑣時忙等,還在占著 cpu,mutex 等待鎖時睡眠。
D 狀態是的說明可以參考下邊的博客。
linux 中進程的 D 狀態和 Z 狀態
如果線程長時間處于 D 狀態,那么內核也有檢測機制,可以檢測出來 D 狀態異常并打印相關信息。
2.3 區別
2.3.1 等鎖的方式不一樣
如上所述,自旋鎖在等待鎖時是忙等,一直占著 cpu,線程狀態是 R,也就是運行態;互斥體在等鎖時,線程是 D 狀態,線程睡眠,不占用 cpu。
?
2.3.2 使用場景不一樣
根據自旋鎖和互斥體的區別,兩者有不同的適用場景。
只能使用自旋鎖的場景
中斷上下文。在中斷上下文中使用鎖,只能使用自旋鎖,因為互斥體在等鎖的時候會睡眠,睡眠就會引起任務調度。而在中斷上下文中,是不能調度的,因為從 linux 內核的語義來說,中斷本身就是最緊迫的,優先級最高的任務,需要快速完成的任務,所以中斷中不能發生調度。linux 內核并不保存中斷的上下文,在中斷中發生調度的話,就永遠無法調度回來,找不到回家的路。
在臨界區需要睡眠,只能使用互斥體
如果在臨界區需要睡眠,那么只能使用互斥體。因為睡眠會引起調度,如果在持有自旋鎖的時候睡眠,并且新調度的任務也要獲取同一個自旋鎖,那么就可能產生死鎖。為什么持有互斥體,不會產生死鎖呢,因為等互斥體的時候線程會睡眠,睡眠的話,原來持有互斥體的線程就會有機會運行,運行完畢,釋放互斥體之后,后來的線程就可以運行。而獲取自旋鎖的話,等鎖的線程會一直占著 cpu,已經持有鎖的線程可能得不到運行,也就不會釋放自旋鎖,這樣就會死鎖。之所以說可能死鎖,是因為這是前后兩個線程在同一個 cpu 核上運行的情況,如果兩個線程在不同的 cpu 核上運行,那么已經持有鎖的線程就能繼續運行,運行完畢釋放鎖。為了避免可能的死鎖問題,在自旋鎖臨界區,不能睡眠。
其它選擇
低開銷加鎖,臨界區很短。建議使用自旋鎖,因為臨界區短,所以等鎖時間也會比較短,忙等一會可以接收。因為等互斥體的時候會觸發調度,如果臨界區很短的話,進程調度的時間會大于等鎖本身需要消耗的時間,還不如把時間花在等鎖上。
短期鎖定,優先使用自旋鎖;長期加鎖,優先選用互斥體。
3 哪種鎖可以遞歸調用
由上邊的分析可以知道,讀自旋鎖可以遞歸調用。
普通自旋鎖,寫自旋鎖,互斥體都不可以遞歸調用。