從策略和實踐，帶你掌握死鎖檢測
本文分享自華為雲社群《掌握死鎖檢測：策略和最佳實踐》，作者： Lion Long。
一、背景：死鎖產生原因
死鎖，是指多個執行緒或者程序在執行過程中因爭奪資源而造成的一種僵局，當程序或者執行緒處於這種僵持狀態，若無外力作用，它們將無法再向前推進。
如下圖所示，執行緒 A 想獲取執行緒 B 的鎖，執行緒 B 想獲取執行緒 C 的鎖，執行緒 C 想獲取執行緒 D 的鎖，執行緒 D 想獲取執行緒 A 的鎖，從而構建了一個資源獲取環。
如果有兩個及以上的CPU佔用率達到100%時，極可能是程式進入死鎖狀態。
死鎖的存在是因為有資源獲取環的存在，所以只要能檢測出資源獲取環，就等同於檢測出死鎖的存在。
1.1、構建一個死鎖
#include <stdio.h> #include <stdlib.h> #include <unistd.h> #include <pthread.h> pthread_mutex_t mutex1 = PTHREAD_MUTEX_INITIALIZER; pthread_mutex_t mutex2 = PTHREAD_MUTEX_INITIALIZER; pthread_mutex_t mutex3 = PTHREAD_MUTEX_INITIALIZER; pthread_mutex_t mutex4 = PTHREAD_MUTEX_INITIALIZER; void *thread_funcA(void *arg) { pthread_mutex_lock(&mutex1); sleep(1); pthread_mutex_lock(&mutex2); printf("funcA --> \n"); pthread_mutex_unlock(&mutex2); pthread_mutex_unlock(&mutex1); } void *thread_funcB(void *arg) { pthread_mutex_lock(&mutex2); sleep(1); pthread_mutex_lock(&mutex3); printf("funcB --> \n"); pthread_mutex_unlock(&mutex3); pthread_mutex_unlock(&mutex2); } void *thread_funcC(void *arg) { pthread_mutex_lock(&mutex3); sleep(1); pthread_mutex_lock(&mutex4); printf("funcC --> \n"); pthread_mutex_unlock(&mutex4); pthread_mutex_unlock(&mutex3); } void *thread_funcD(void *arg) { pthread_mutex_lock(&mutex4); sleep(1); pthread_mutex_lock(&mutex1); printf("funcD --> \n"); pthread_mutex_unlock(&mutex1); pthread_mutex_unlock(&mutex4); } int main() { pthread_t tid[4] = { 0 }; pthread_create(&tid[0], NULL, thread_funcA, NULL); pthread_create(&tid[1], NULL, thread_funcB, NULL); pthread_create(&tid[2], NULL, thread_funcC, NULL); pthread_create(&tid[3], NULL, thread_funcD, NULL); pthread_join(tid[0], NULL); pthread_join(tid[1], NULL); pthread_join(tid[2], NULL); pthread_join(tid[3], NULL); return 0; }
二、使用hook檢測死鎖
hook使用場景：
（1）實現自己的協定棧，通過hook posix api。
2.1、dlsym()函數
獲取共用物件或可執行檔案中符號的地址。
函數原型：
#include <dlfcn.h> void *dlsym(void *handle, const char *symbol); #define _GNU_SOURCE #include <dlfcn.h> void *dlvsym(void *handle, char *symbol, char *version); // Link with -ldl.
描述：
函數dlsym()接受dlopen()返回的動態載入共用物件的「控制程式碼」以及以空結尾的符號名，並返回該符號載入到記憶體中的地址。如果在指定物件或載入物件時dlopen()自動載入的任何共用物件中找不到該符號，dlsym()將返回NULL。（dlsym()執行的搜尋是通過這些共用物件的依賴關係樹進行的廣度優先搜尋。）
由於符號的值實際上可能是NULL（因此，dlsym()的NULL返回值不必指示錯誤），因此測試錯誤的正確方法是呼叫dlerror()以清除任何舊的錯誤條件，然後呼叫dlsym。
handle中可以指定兩個特殊的偽控制程式碼：
| 程式碼 | 含義 |
|---|---|
| RTLD_DEFAULT | 使用預設共用物件搜尋順序查詢所需符號的第一個匹配項。搜尋將包括可執行檔案及其依賴項中的全域性符號，以及使用RTLD_GLOBAL標誌動態載入的共用物件中的符號。 |
| RTLD_NEXT | 在當前物件之後，按搜尋順序查詢所需符號的下一個匹配項。這允許在另一個共用物件中為函數提供包裝，例如，預載入共用物件中的函數定義可以查詢並呼叫另一共用物件中提供的「真實」函數（或者在預載入有多層的情況下，函數的「下一個」定義）。 |
函數dlvsym（）的作用與dlsym（）相同，但使用版本字串作為附加引數。
返回值：
成功時，這些函數返回與符號關聯的地址。
失敗時，返回NULL；可以使用dlerror()診斷錯誤的原因。
2.2、pthread_self()函數
獲取呼叫執行緒的ID。
函數原型：
#include <pthread.h> pthread_t pthread_self(void); // Compile and link with -pthread.
說明：
函數的作用是返回撥用執行緒的ID。這與建立此執行緒的pthread_create()呼叫中*thread中返回的值相同。
返回值：
此函數始終成功，返回撥用執行緒的ID。
2.3、實現步驟
（1）構建函數指標
（2）定義與目標函數一樣的型別
typedef int(*pthread_mutex_lock_t)(pthread_mutex_t *mutex); typedef int(*pthread_mutex_unlock_t)(pthread_mutex_t *mutex); pthread_mutex_lock_t pthread_mutex_lock_f; pthread_mutex_unlock_t pthread_mutex_unlock_f;
（3）具體函數實現，函數名與目標函數名一致
int pthread_mutex_lock(pthread_mutex_t *mutex) { pthread_t selfid = pthread_self(); printf("pthread_mutex_lock: %ld, %p\n", selfid, mutex); // ... return 0; } int pthread_mutex_unlock(pthread_mutex_t *mutex) { pthread_t selfid = pthread_self(); printf("pthread_mutex_unlock: %ld, %p\n", selfid, mutex); // ... return 0; }
（4）呼叫dlsym()函數，即勾點。
int init_hook() { pthread_mutex_lock_f = dlsym(RTLD_NEXT, "pthread_mutex_lock"); pthread_mutex_unlock_f = dlsym(RTLD_NEXT, "pthread_mutex_unlock"); // ... return 0; }
2.4、範例程式碼
#define _GNU_SOURCE #include <dlfcn.h> #include <stdio.h> #include <stdlib.h> #include <unistd.h> #include <pthread.h> typedef int(*pthread_mutex_lock_t)(pthread_mutex_t *mutex); typedef int(*pthread_mutex_unlock_t)(pthread_mutex_t *mutex); pthread_mutex_lock_t pthread_mutex_lock_f; pthread_mutex_unlock_t pthread_mutex_unlock_f; int pthread_mutex_lock(pthread_mutex_t *mutex) { pthread_t selfid = pthread_self(); pthread_mutex_lock_f(mutex); printf("pthread_mutex_lock: %ld, %p\n", selfid, mutex); return 0; } int pthread_mutex_unlock(pthread_mutex_t *mutex) { pthread_t selfid = pthread_self(); pthread_mutex_unlock_f(mutex); printf("pthread_mutex_unlock: %ld, %p\n", selfid, mutex); return 0; } int init_hook() { pthread_mutex_lock_f = dlsym(RTLD_NEXT, "pthread_mutex_lock"); pthread_mutex_unlock_f = dlsym(RTLD_NEXT, "pthread_mutex_unlock"); return 0; } #if 1 // debug pthread_mutex_t mutex1 = PTHREAD_MUTEX_INITIALIZER; pthread_mutex_t mutex2 = PTHREAD_MUTEX_INITIALIZER; pthread_mutex_t mutex3 = PTHREAD_MUTEX_INITIALIZER; pthread_mutex_t mutex4 = PTHREAD_MUTEX_INITIALIZER; void *thread_funcA(void *arg) { pthread_mutex_lock(&mutex1); sleep(1); pthread_mutex_lock(&mutex2); printf("funcA --> \n"); pthread_mutex_unlock(&mutex2); pthread_mutex_unlock(&mutex1); } void *thread_funcB(void *arg) { pthread_mutex_lock(&mutex2); sleep(1); pthread_mutex_lock(&mutex3); printf("funcB --> \n"); pthread_mutex_unlock(&mutex3); pthread_mutex_unlock(&mutex2); } void *thread_funcC(void *arg) { pthread_mutex_lock(&mutex3); sleep(1); pthread_mutex_lock(&mutex4); printf("funcC --> \n"); pthread_mutex_unlock(&mutex4); pthread_mutex_unlock(&mutex3); } void *thread_funcD(void *arg) { pthread_mutex_lock(&mutex4); sleep(1); pthread_mutex_lock(&mutex1); printf("funcD --> \n"); pthread_mutex_unlock(&mutex1); pthread_mutex_unlock(&mutex4); } int main() { init_hook(); pthread_t tid[4] = { 0 }; pthread_create(&tid[0], NULL, thread_funcA, NULL); pthread_create(&tid[1], NULL, thread_funcB, NULL); pthread_create(&tid[2], NULL, thread_funcC, NULL); pthread_create(&tid[3], NULL, thread_funcD, NULL); pthread_join(tid[0], NULL); pthread_join(tid[1], NULL); pthread_join(tid[2], NULL); pthread_join(tid[3], NULL); return 0; } #endif
缺點：這種方式在少量鎖情況下還可以分析，在大量鎖使用的情況，分析過程極為困難。
三、使用圖演演算法檢測死鎖
死鎖檢測可以利用圖演演算法，檢測有向圖是否有環。
3.1、圖的構建
（1）矩陣
| 指向 1 | 指向 2 | 指向 3 | 指向 … | |
|---|---|---|---|---|
| 節點 1 | ||||
| 節點 2 | ||||
| 節點 3 | ||||
| 節點 … |
（2）鄰接表
資料結構原理示意圖：
「圖」連線：
3.2、圖的使用
先新增節點再新增邊。
（1）每建立一個執行緒，新增一個節點；注意，不是執行緒建立的時候就要加節點（有些執行緒不會用到鎖），而是執行緒呼叫鎖（以互斥鎖為例，pthread_mutex_lock() ）的時候才新增節點。
（2）執行緒加鎖（以互斥鎖為例，pthread_mutex_lock() ）的時候，並且檢測到鎖已經佔用，則新增一條邊。
（3）移除邊，呼叫鎖（以互斥鎖為例，pthread_mutex_lock() ）前，如果此時鎖沒有被佔用，並且該邊存在，則移除邊。
（4）移除節點是在解鎖之後。
三個原語操作：
（1）加鎖之前的操作，lock_before()；
（2）加鎖之後的操作，lock_after()；
（3）解鎖之後的操作，unlock_after()；
3.3、範例程式碼
程式碼比較長，為了避免篇幅較長，不利於閱讀，這裡沒有貼上。如果需要，可以聯絡博主，或者關注微信公眾號 《Lion 萊恩呀》 獲取。
總結
死鎖的產生是因為多執行緒之間存在交叉申請鎖的情況，因爭奪資源而造成的一種僵局。
hook使用：
（1）定義與目標函數一樣的型別；
（2）具體函數實現，函數名與目標函數名一致；
（3）呼叫dlsym()函數，初始化hook。
死鎖檢測可以使用圖演演算法，通過檢測有向圖是否有環判斷是否有死鎖。