前言

本文將從0到1寫一個死鎖檢測組件。源碼：deadlock_success.c

組件如何放入自己的項目里？把代碼末兩個Debug部分刪除，在你的項目里添加下面兩句代碼即可使用死鎖檢測組件。

init_hook();
start_check();

1. 死鎖的現象以及原理

1.1 復現最簡單的死鎖

線程A占有鎖1，線程B占有鎖2；此時線程A想要獲取鎖2，但是鎖2已經被線程B占有， 此時線程A會休眠等待線程B釋放鎖2后，再去獲得鎖2。可以看到下面的場景，線程B想要獲取鎖1，結果線程B也休眠去了。這就導致死鎖，鎖1和鎖2永遠得不到釋放，因為線程A和線程B都在等待另一個鎖的釋放。這種僵持的狀態，就稱為死鎖。

正如下面代碼所示，這樣就引發了死鎖

void *thread_rountine_1(void *args) {
pthread_t selfid = pthread_self();
printf("thread_routine 1 : %ld n", selfid);
pthread_mutex_lock(&mutex_1);
sleep(1);
pthread_mutex_lock(&mutex_2);
pthread_mutex_unlock(&mutex_2);
pthread_mutex_unlock(&mutex_1);
return (void *) (0);
}
void *thread_rountine_2(void *args) {
pthread_t selfid = pthread_self(); //
printf("thread_routine 2 : %ld n", selfid);
pthread_mutex_lock(&mutex_2);
sleep(1);
pthread_mutex_lock(&mutex_1);
pthread_mutex_unlock(&mutex_1);
pthread_mutex_unlock(&mutex_2);
return (void *) (0);
}

1.2 從死鎖中找出檢測死鎖的規律

我們來看看下面這張圖，線程A想要獲取線程B的資源，線程B想要獲取線程C的資源，線程C想要獲取線程D的資源，線程D想要獲取線程A的資源，這其實就構成了一個有向圖的環路

來看看前面介紹的最簡單的死鎖，發現其本直也是構成了一個有向圖的環路

來看看非死鎖的場景，只要線程D釋放了mutex4，那么線程C就能獲得鎖，隨后線程C釋放mutex3和4，那么線程B…可以發現，這個非死鎖的場景，它是一個有向圖,但這個圖沒有構成環路。

過上面三個場景的分析，我們其實就可以把死鎖的問題，轉換為 有向圖的環路檢測。在線程進行加鎖前，我們去判斷一下所有的線程有沒有構成環路，如果有，則說明現在很有可能會進入死鎖。

2. 檢測死鎖的前置條件

2.1 有向圖的邊怎么來？

我們現在已經知道了死鎖的問題，就轉換為 有向圖的環路檢測。那么這個有向圖怎么構建？在我們對mutex1加鎖的時候，我們怎么知道是線程A占有mutex1，在對mutex2加鎖的時候，怎么知道它已經被線程B占有了？我們無法知道鎖是屬于哪個線程的。既然連鎖都不知道屬于哪個線程，哪有如何構建出有向圖呢？換言之，我們需要解決：知道當前鎖被哪個線程占用。我們不知道的原因很簡單，就是mutex和pthread_id沒有一個對應關系。

//鎖與線程的信息struct pair_t {    unsigned long int th_id;    enum Type type;    unsigned long int lock_id;    int degress;};

我們可以做出一個數據結構，在加鎖之前，判斷這個鎖有沒有被別的線程使用，如果沒有，在加鎖之后我們將這個鎖與本線程綁定，做一個pair，然后把這個pair存起來。比如說線程線程A和mutex1綁定，線程B和mutex2綁定了。當線程A再次去嘗試對mutex2加鎖之前，先判斷mutex2是否名花有主？如果有，那有向圖的邊不就來了嗎？不知道讀者有沒有注意到，這一段話都建立在加鎖之前判斷 鎖 是否名花有主。

有一個非常簡單粗暴的方法，在加鎖之前調用一個函數，加鎖之后調用一個函數。讀者可以想一下，本文是要實現一個組件，所謂組件，給別人也能用，難道在一個項目里面，想要檢測一下死鎖，去把lock上下全部加兩個函數？這顯然不符合我們組件的設想，我們希望不改變別人的代碼，就能實現檢測。

//鎖與線程的信息
struct pair_t {
    unsigned long int th_id;
    enum Type type;

    unsigned long int lock_id;
    int degress;
};

要想實現上面的需求，我們可以使用hook。

2.2 hook—>dlsym

hook是什么意思？鉤子，簡單來說，我們使用hook，可以把系統或第三方庫提供的函數，替換成我們寫的同名函數，而第三方庫的函數則被我們改名，在我們寫的同名函數里，可以去調用第三方庫原來的函數。

正如下面代碼所示，系統提供的pthread_mutex_lock被改名為pthread_mutex_lock_f。那么我們就可以使用pthread_mutex_lock來當作函數名稱，如此一來，在別的項目里面，我們通過hook就可以進行死鎖檢測，而不需要去改代碼了。

hook提供了兩個接口；1. dlsym()是針對系統的，系統原始的api。2. dlopen()是針對第三方的庫。

/* ******* ******************Hook****************** ******* */

typedef int (*pthread_mutex_lock_t)(pthread_mutex_t *mutex);

pthread_mutex_lock_t pthread_mutex_lock_f;

typedef int (*pthread_mutex_unlock_t)(pthread_mutex_t *mutex);

pthread_mutex_unlock_t pthread_mutex_unlock_f;

static int init_hook() {
    pthread_mutex_lock_f = dlsym(RTLD_NEXT, "pthread_mutex_lock");
    pthread_mutex_unlock_f = dlsym(RTLD_NEXT, "pthread_mutex_unlock");
}


int pthread_mutex_lock(pthread_mutex_t *mutex) {
    pthread_t self_id = pthread_self(); //
    lock_before(self_id, (unsigned long int) mutex);
    pthread_mutex_lock_f(mutex);
    lock_after(self_id, (unsigned long int) mutex);
}

在進程的虛擬內存空間里面，有一塊代碼段 ，上面代碼中，pthread_mutex_lock_f是一個函數指針，實際上，就是把pthread_mutex_lock_f指向代碼段里系統函數的入口地址 ，以此來實現偷天換日。

還需要注意一點，這個#define _GNU_SOURCE要寫在前面，因為這個就相當于一個開關，在下面的.h文件里面，有#ifdef _GNU_SOURCE的地方。在gcc編譯的時候后面加上 -ldl。

#define _GNU_SOURCE
#include < dlfcn.h >

3. 有向圖

3.1 有向圖的數據結構

下面來看一下結構體的含義

ertex_list的每一項，都是一個頂點，后面鏈表里面存的，都是邊的另一個點。

vlock_list的每一項，存的都是鎖與線程的信息

/* ******* ******************Digraph****************** ******* */

enum Type {
    PROCESS, RESOURCE
};
//鎖與線程的信息
struct pair_t {
    unsigned long int th_id;
    enum Type type;

    unsigned long int lock_id;
    int degress;
};
//頂點
struct vertex_t {
    struct pair_t pair;
    struct vertex_t *next;
};

struct task_graph {
    struct vertex_t vertex_list[MAX];
    int vertex_num;

    struct pair_t lock_list[MAX];
    int lock_num;

    pthread_mutex_t mutex;

    int path[MAX + 1];
    int visited[MAX];
    int k;
    int deadlock;
};

struct task_graph *tg = NULL;


//創建一個vertex
struct vertex_t *create_vertex(struct pair_t pair) {
    struct vertex_t *tex = (struct vertex_t *) malloc(sizeof(struct vertex_t));

    tex- >pair = pair;
    tex- >next = NULL;
    return tex;
}

//查找vertex在list里面的下標
int search_vertex(struct pair_t pair) {
    int i = 0;

    for (i = 0; i < tg- >vertex_num; i++) {
        if (tg- >vertex_list[i].pair.type == pair.type && tg- >vertex_list[i].pair.th_id == pair.th_id) {
            return i;
        }
    }

    return -1;
}

//把vertex添加到vertex_list里面
void add_vertex(struct pair_t pair) {
    if (search_vertex(pair) == -1) {
        tg- >vertex_list[tg- >vertex_num].pair = pair;
        tg- >vertex_list[tg- >vertex_num].next = NULL;
        tg- >vertex_num++;
    }
}

//添加邊,把v添加到u的鏈表里
int add_edge(struct pair_t u, struct pair_t v) {
    add_vertex(u);
    add_vertex(v);

    struct vertex_t *cnt = &(tg- >vertex_list[search_vertex(u)]);

    while (cnt- >next != NULL) {
        cnt = cnt- >next;
    }

    cnt- >next = create_vertex(v);
}

//檢查邊是否存在
int verify_edge(struct pair_t u, struct pair_t v) {
    if (tg- >vertex_num == 0) return 0;

    int idx = search_vertex(u);
    if (idx == -1) {
        return 0;
    }

    struct vertex_t *cnt = &(tg- >vertex_list[idx]);

    while (cnt != NULL) {
        if (cnt- >pair.th_id == v.th_id) {
            return 1;
        }
        cnt = cnt- >next;
    }
    return 0;
}

//刪除邊
int remove_edge(struct pair_t u, struct pair_t v) {

    int idx_u = search_vertex(u);
    int idx_v = search_vertex(v);

    if (idx_u != -1 && idx_v != -1) {

        struct vertex_t *cnt = &tg- >vertex_list[idx_u];
        struct vertex_t *remove;

        while (cnt- >next != NULL) {
            if (cnt- >next- >pair.th_id == v.th_id) {
                remove = cnt- >next;
                cnt- >next = cnt- >next- >next;
                free(remove);
                break;
            }
            cnt = cnt- >next;
        }
    }
}

3.2 dfs判斷環的方法

現在邊也處理好了，鎖與線程的關系也處理好了，那么我們如何去判斷有沒有環呢？我們使用DFS來判斷。

/* ******* ******************check cycle****************** ******* */

//打印
void print_deadlock(void) {
    int i = 0;
    printf("deadlock : ");
    for (i = 0; i < tg- >k - 1; i++) {
        printf("%ld -- > ", tg- >vertex_list[tg- >path[i]].pair.th_id);
    }
    printf("%ldn", tg- >vertex_list[tg- >path[i]].pair.th_id);
}

void print_locklist(void) {
    int i = 0;

    printf("-----------print_locklist----------n");
    for (i = 0; i < tg- >lock_num; i++) {
        printf("threadid : %ld, lockid: %ldn", tg- >lock_list[i].th_id, tg- >lock_list[i].lock_id);
    }
    printf("-----------------------------------n");
}

int DFS(int idx) {
    struct vertex_t *ver = &tg- >vertex_list[idx];
    if (tg- >visited[idx] == 1) {
        tg- >path[tg- >k++] = idx;
        print_deadlock();
        tg- >deadlock = 1;
        return 0;
    }

    tg- >visited[idx] = 1;
    tg- >path[tg- >k++] = idx;

    while (ver- >next != NULL) {
        DFS(search_vertex(ver- >next- >pair));
        tg- >k--;
        ver = ver- >next;
    }

    return 1;
}

//判斷某個頂點是否成環
int search_for_cycle(int idx) {
    struct vertex_t *ver = &tg- >vertex_list[idx];
    tg- >visited[idx] = 1;
    tg- >k = 0;
    tg- >path[tg- >k++] = idx;

    while (ver- >next != NULL) {
        int i = 0;
        for (i = 0; i < tg- >vertex_num; i++) {
            if (i == idx) continue;
            tg- >visited[i] = 0;
        }

        for (i = 1; i <= MAX; i++) {
            tg- >path[i] = -1;
        }
        tg- >k = 1;

        DFS(search_vertex(ver- >next- >pair));
        ver = ver- >next;
    }
}

//檢查是否死鎖
void check_dead_lock(void) {
    printf("-----------check deadlock----------n");

    int i;
    tg- >deadlock = 0;
    for (i = 0; i < tg- >vertex_num; i++) {
        if (tg- >deadlock == 1) {
            break;
        }
        //從每個點都出發一遍
        search_for_cycle(i);
    }
    if (tg- >deadlock == 0) {
        printf("no deadlockn");
    }

    printf("----------------------------------n");
}

3.3 簡單測試一下

可以看到我們的結果與預期一致，說明我們的有向圖與判斷環完成了，那么下面我們就應該去寫上鎖前后的函數了。

/* ******* ******************Debug 2****************** ******* */


int main() {
    tg = (struct task_graph *) malloc(sizeof(struct task_graph));
    tg- >vertex_num = 0;

    struct pair_t v1;
    v1.th_id = 1;
    v1.type = PROCESS;
    add_vertex(v1);

    struct pair_t v2;
    v2.th_id = 2;
    v2.type = PROCESS;
    add_vertex(v2);

    struct pair_t v3;
    v3.th_id = 3;
    v3.type = PROCESS;
    add_vertex(v3);

    struct pair_t v4;
    v4.th_id = 4;
    v4.type = PROCESS;
    add_vertex(v4);


    struct pair_t v5;
    v5.th_id = 5;
    v5.type = PROCESS;
    add_vertex(v5);


    add_edge(v1, v2);
    add_edge(v2, v3);
    add_edge(v3, v4);
    add_edge(v4, v5);
    add_edge(v3, v1);
    add_edge(v5, v1);

    check_dead_lock();
//    search_for_cycle(search_vertex(v1));
}

root@wxf:/tmp/tmp.d4vz2dOyJP# gcc -o deadlock_success deadlock_success.c -lpthread -ldl
root@wxf:/tmp/tmp.d4vz2dOyJP# ./deadlock_success 
-----------check deadlock----------
deadlock : 1 -- > 2 -- > 3 -- > 4 -- > 5 -- > 1
deadlock : 1 -- > 2 -- > 3 -- > 1
----------------------------------
root@wxf:/tmp/tmp.d4vz2dOyJP#

4. 三個原語操作

現在有向圖和hook都有了，那么我們如何把死鎖檢測出來？換言之，我們怎么使用pthread_mutex_lock和pthread_mutex_unlock構建有向圖？

在調用系統提供的lock以前，我們需要檢測這個鎖有沒有被別的線程占用，如果被占用，那么我們就需要往圖里面加一條邊。

如果沒有被占用，那么我們就往里面走。也就是說加鎖完，調用系統提供的lock之后， 我們需要告訴后面的線程，這個鎖被我占用了，即添加一項pair，供別人lock之前去檢測。如果被占用了，然后鎖被釋放，本線程獲取到了這個以前被占用的鎖，那么我們lock之后，需要把原來添加的一條邊刪除掉，因為這個鎖已經屬于自己了，并且將鎖對應的pair中的th_id改成自己。

在調用系統提供的unlock之后，解鎖了一個鎖之后，我們去看看還有沒有渴望得到這個鎖的，如果沒有，則將鎖對應的pair置空，如果有，則不管pair。

注意：下面三個函數，我對三個函數都加鎖了，這里是我的偷懶操作，鎖的粒度較大。如果想優化，應該放到serch函數里面，我這里懶得去改了。

int pthread_mutex_lock(pthread_mutex_t *mutex) {
    pthread_t self_id = pthread_self(); 

    lock_before(self_id, (unsigned long int) mutex);
    pthread_mutex_lock_f(mutex);
    lock_after(self_id, (unsigned long int) mutex);
}

int pthread_mutex_unlock(pthread_mutex_t *mutex) {
    pthread_t self_id = pthread_self();

    pthread_mutex_unlock_f(mutex);
    unlock_after(self_id, (unsigned long int) mutex);
}

4.1 lock_before

我們現在把加鎖理解為談戀愛確認關系。在確認關系之前，我們要去看一下這個女生有沒有男朋友，如果她沒有男朋友，妙哉！那么我們就直接確認關系(lock)吧！如果她有男朋友，那現在還不能和她談戀愛，我們先與她曖昧曖昧(add_edge)，等著她分手。

void lock_before(unsigned long int thread_id, unsigned long int lock) {
    pthread_mutex_lock_f(&tg- >mutex);

    int idx = search_lock(lock);
//    printf("[lock_before] self_id:%lu lock:%lu lock idx:%d n", thread_id, lock, idx);
    //如果該鎖是第一次則什么都不做
    if (idx != -1) {
        //u是想要加鎖的線程
        struct pair_t u;
        u.th_id = thread_id;
        u.type = PROCESS;
        //把vertex添加到vertex_list里面
        add_vertex(u);
        //v是鎖原來的線程
        struct pair_t v;
        v.th_id = tg- >lock_list[idx].th_id;
        tg- >lock_list[idx].degress++;
        v.type = PROCESS;
        add_vertex(v);

        if (!verify_edge(u, v)) {
            add_edge(u, v); // 把v加入到vertex_list的u的鏈表中
        }
    }

    pthread_mutex_unlock_f(&tg- >mutex);
}

4.2 lock_after

現在我們加鎖完了，也就是談戀愛確認關系了之后，如果我們是她的初戀，那么我們要向全世界宣布(tg->lock_list[empty_lock_idx])：她，是我的女人！如果不是初戀，她被別人宣布過了，那我們就別搞這么浪漫了，把她給我們的備注改成男朋友就好了(tg->lock_list[idx].th_id = thread_id;)，并且我們也不需要曖昧聊天了(remove_edge)，因為她已經是我們女朋友了。

void lock_after(unsigned long int thread_id, unsigned long int lock) {
    pthread_mutex_lock_f(&tg- >mutex);

    int idx = search_lock(lock);
//    printf("[lock_after ] self_id:%lu lock:%lu ", thread_id, lock);

    if (idx == -1) {  // 第一次加鎖,找一個空位lock_list，設置th_id和lock
        int empty_lock_idx = search_empty_lock(lock);
        tg- >lock_list[empty_lock_idx].th_id = thread_id;
        tg- >lock_list[empty_lock_idx].lock_id = lock;
//        printf("分配lock_list位置 idx:%d n", empty_lock_idx);
        if (empty_lock_idx >= tg- >lock_num) {
            inc(&tg- >lock_num, 1);
        }
    }
    else {
        //u是想要加鎖的線程
        struct pair_t u;
        u.th_id = thread_id;
        u.type = PROCESS;
        //v是鎖原來的線程
        struct pair_t v;
        v.th_id = tg- >lock_list[idx].th_id;
        tg- >lock_list[idx].degress--;
        v.type = PROCESS;
        //刪除邊
        if (verify_edge(u, v)) {
            remove_edge(u, v);
        }
        //設為本線程
        tg- >lock_list[idx].th_id = thread_id;

//        printf("獲得 lock idx:%d n", idx);
    }

    pthread_mutex_unlock_f(&tg- >mutex);
}

4.3 unlock_after

unlock就相當于分手，如果她沒有備胎，那么她就恢復單身(pair置空)，如果她有備胎，那就隨她吧~

void unlock_after(unsigned long int thread_id, unsigned long int lock) {
    pthread_mutex_lock_f(&tg- >mutex);

    int idx = search_lock(lock);
    //如果入度為0，說明沒有別的線程指向該鎖，則把這個idx位置置空
    if (tg- >lock_list[idx].degress == 0) {
        tg- >lock_list[idx].th_id = 0;
        tg- >lock_list[idx].lock_id = 0;
    }

    pthread_mutex_unlock_f(&tg- >mutex);
}

5. 死鎖檢測線程的測試

下面我們來測試這個場景。完整代碼在目錄前言中。

/* ******* ******************Debug 1****************** ******* */
pthread_mutex_t mutex_1 = PTHREAD_MUTEX_INITIALIZER;
pthread_mutex_t mutex_2 = PTHREAD_MUTEX_INITIALIZER;
pthread_mutex_t mutex_3 = PTHREAD_MUTEX_INITIALIZER;
pthread_mutex_t mutex_4 = PTHREAD_MUTEX_INITIALIZER;

void *thread_rountine_1(void *args) {
    pthread_t selfid = pthread_self(); //

    printf("thread_routine 1 : %ld n", selfid);

    pthread_mutex_lock(&mutex_1);
    sleep(1);
    pthread_mutex_lock(&mutex_2);

    pthread_mutex_unlock(&mutex_2);
    pthread_mutex_unlock(&mutex_1);

    return (void *) (0);
}

void *thread_rountine_2(void *args) {
    pthread_t selfid = pthread_self(); //

    printf("thread_routine 2 : %ld n", selfid);

    pthread_mutex_lock(&mutex_2);
    sleep(1);
    pthread_mutex_lock(&mutex_3);

    pthread_mutex_unlock(&mutex_3);
    pthread_mutex_unlock(&mutex_2);

    return (void *) (0);
}

void *thread_rountine_3(void *args) {
    pthread_t selfid = pthread_self(); //

    printf("thread_routine 3 : %ld n", selfid);

    pthread_mutex_lock(&mutex_3);
    sleep(1);
    pthread_mutex_lock(&mutex_4);

    pthread_mutex_unlock(&mutex_4);
    pthread_mutex_unlock(&mutex_3);

    return (void *) (0);
}

void *thread_rountine_4(void *args) {
    pthread_t selfid = pthread_self(); //

    printf("thread_routine 4 : %ld n", selfid);

    pthread_mutex_lock(&mutex_4);
    sleep(1);
    pthread_mutex_lock(&mutex_1);

    pthread_mutex_unlock(&mutex_1);
    pthread_mutex_unlock(&mutex_4);

    return (void *) (0);
}


int main() {
    init_hook();
    start_check();

    printf("start_checkn");

    pthread_t tid1, tid2, tid3, tid4;
    pthread_create(&tid1, NULL, thread_rountine_1, NULL);
    pthread_create(&tid2, NULL, thread_rountine_2, NULL);
    pthread_create(&tid3, NULL, thread_rountine_3, NULL);
    pthread_create(&tid4, NULL, thread_rountine_4, NULL);

    pthread_join(tid1, NULL);
    pthread_join(tid2, NULL);
    pthread_join(tid3, NULL);
    pthread_join(tid4, NULL);

    return 0;
}