死锁的现象以及原理

描述

前言

本文将从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;
}

线程

打开APP阅读更多精彩内容
声明:本文内容及配图由入驻作者撰写或者入驻合作网站授权转载。文章观点仅代表作者本人,不代表电子发烧友网立场。文章及其配图仅供工程师学习之用,如有内容侵权或者其他违规问题,请联系本站处理。 举报投诉

全部0条评论

快来发表一下你的评论吧 !

×
20
完善资料,
赚取积分