3.2.4 手写死锁检测组件

死锁常见在线程对资源的访问情形下：
比如两个线程同时请求对方 已经被占用的资源：

造成死锁，谁也不让谁。
对于多个线程来说，造成死锁表现为，线程占用其他线程的资源，构成”环“

因此，检测是否发生死锁，我们就可以通过检测有向图是否成环来判断。
来看一个死锁的例子：

pthread_mutex_t r1 = PTHREAD_MUTEX_INITIALIZER;
pthread_mutex_t r2 = PTHREAD_MUTEX_INITIALIZER;
pthread_mutex_t r3 = PTHREAD_MUTEX_INITIALIZER;
pthread_mutex_t r4 = PTHREAD_MUTEX_INITIALIZER;
pthread_mutex_t r5 = PTHREAD_MUTEX_INITIALIZER;

void *t1_cb(void *arg) 
{
    pthread_t selfid = pthread_self();
    pthread_mutex_lock(&r1);
    sleep(1);
    pthread_mutex_lock(&r2);
    pthread_mutex_unlock(&r2);

    pthread_mutex_unlock(&r1);
}

void *t2_cb(void *arg) 
{
    pthread_mutex_lock(&r2);
    sleep(1);
    pthread_mutex_lock(&r3);
    pthread_mutex_unlock(&r3);

    pthread_mutex_unlock(&r2);
}

void *t3_cb(void *arg) 
{
    pthread_mutex_lock(&r3);
    sleep(1);
    pthread_mutex_lock(&r4);
    pthread_mutex_unlock(&r4);

    pthread_mutex_unlock(&r3);
}

void *t4_cb(void *arg) 
{
    pthread_mutex_lock(&r4);
    sleep(1);
    pthread_mutex_lock(&r5);
    pthread_mutex_unlock(&r5);

    pthread_mutex_unlock(&r4);
}

void *t5_cb(void *arg) 
{
    pthread_mutex_lock(&r5);
    sleep(1);
    pthread_mutex_lock(&r1);
    pthread_mutex_unlock(&r1);

    pthread_mutex_unlock(&r5);
}

int main() 
{
    pthread_t t1, t2, t3, t4, t5;

    pthread_create(&t1, NULL, t1_cb, NULL);
    pthread_create(&t2, NULL, t2_cb, NULL);
    pthread_create(&t3, NULL, t3_cb, NULL);
    pthread_create(&t4, NULL, t4_cb, NULL);
    pthread_create(&t5, NULL, t5_cb, NULL);
    
    pthread_join(t1, NULL);
    pthread_join(t2, NULL);
    pthread_join(t3, NULL);
    pthread_join(t4, NULL);
    pthread_join(t5, NULL);

    printf("complete\n");

    return 0;
}

定义了五个线程，每个线程都持有一把互斥锁，线程一首先将r1上锁，之后sleep(1)等待线程二将r2上锁，依次，线程五将r5上锁后，等待r1被释放，然而线程1还在等待r2的释放，无法将r1释放，依次。因此互相持有锁，但都不释放锁，造成了死锁。

其实要检测是否发生死锁，我们还要从锁上下手，我们要对锁的前后时机进行掌控，比如说现在是谁持有锁，谁想占用锁？锁目前的状态等等，总之，就是要对锁的状态等信息有一个记录，从而方便我们判断目前是否造成了死锁。既然这样，我们就不能直接使用系统调用直接上锁，解锁。而需要通过hook，让上锁，解锁操作调用我们自己的函数，这样，我们就能根据线程和锁记录一些关键信息，从而构建一个有向图，之后判断这个有向图是否成环就能知道是否发生了死锁。
下面来看一下如何使用hook技术：

//hook
//define
typedef int (*pthread_mutex_lock_t)(pthread_mutex_t *mutex);
pthread_mutex_lock_t pthread_mutex_lock_f = NULL;
typedef int (*pthread_mutex_unlock_t)(pthread_mutex_t *mutex);
pthread_mutex_unlock_t pthread_mutex_unlock_f = NULL;

//implement
int pthread_mutex_lock(pthread_mutex_t *mutex) {
    pthread_t selfid = pthread_self();

    //lock_before();
    pthread_mutex_lock_f(mutex);
    //lock_after();
}

int pthread_mutex_unlock(pthread_mutex_t *mutex) {
    pthread_t selfid = pthread_self();
    
    pthread_mutex_unlock_f(mutex);
    //unlock_after();
}

//init
void init_hook(void) {
    if (!pthread_mutex_lock_f)
        pthread_mutex_lock_f = dlsym(RTLD_NEXT, "pthread_mutex_lock");

    if (!pthread_mutex_unlock_f)
        pthread_mutex_unlock_f = dlsym(RTLD_NEXT, "pthread_mutex_unlock");
}

其实就是覆写pthread_mutex_lock和pthread_mutex_unlock两个函数，使用时首先调用覆写的，之后如果想真正调用系统调用，使用pthread_mutex_lock_f(mutex)即可。
这样，我们就能对上锁，解锁操作有更细力度的掌握，比如在上锁前可以执行lock_before上锁成功后执行lock_after，解锁成功后执行unlock_after。
在上锁时lock_before构建有向图的边：

void lock_before(pthread_t tid, uint64_t lockaddr) {
    int idx = 0;
    for (idx = 0; idx < tg->lockidx; idx++) {
        if (tg->locklist[idx].lock_id == lockaddr) {
            struct source_type from;
            from.id = tid;
            from.type = PROCESS;
            add_vertex(from);

            struct source_type to;
            to.id = tg->locklist[idx].id;
            to.type = PROCESS;
            add_vertex(to);

            tg->locklist[idx].degress++;
            //如果之前没构成边，就添加边
            if (!verify_edge(from, to))
                add_edge(from, to);
        }
    }
}

lock_after中判断这个锁，从而修改边：

void lock_after(pthread_t tid, uint64_t lockaddr) {
    int idx = 0;
    //如果这个锁之前没被使用，说明没人上锁，那么现在上锁成功。
    //如果这个锁没有被上锁，搜索空位置放入
    if (-1 == (idx = search_lock(lockaddr))) {
        int eidx = search_empty_lock(lockaddr);

        tg->locklist[eidx].id = tid;
        tg->locklist[eidx].lock_id = lockaddr;

        tg->lockidx++;
    } else {
        //如果这个锁之前被用过，说明有人上过锁，那么现在上锁失败。
        struct source_type from;
        from.id = tid;
        from.type = PROCESS;
        add_vertex(from);

        struct source_type to;
        to.id = tg->locklist[idx].id;
        to.type = PROCESS;
        add_vertex(to);

        tg->locklist[idx].degress--;
        //如果之前构成边了，就删除边
        if (verify_edge(from, to))
            remove_edge(from, to);
        
        tg->locklist[idx].id = tid;
    }
}

之后，启动一个新的线程，不断去判断目前的有向图是否有环，从而判断当前线程是否形成死锁：

void check_dead_lock(void)
{
    int i = 0;
    deadlock = 0;
    for (i = 0; i < tg->num; i++) {
        if (deadlock == 1) break;
        search_for_cycle(i);
    }
    if (deadlock == 0) {
        printf("no deadlock\n");
    }
}

static void *thread_routine(void *args)
{
    while (1) {
        sleep(5);
        check_dead_lock();
    }
}

void start_check(void)
{
    tg = (struct task_graph*)malloc(sizeof(struct task_graph));
    tg->num = 0;
    tg->lockidx = 0;

    pthread_t tid;
    pthread_create(&tid, NULL, thread_routine, NULL);
}

至此，我们就可以完成死锁的检测。

文章参考于<零声教育>的C/C++linux服务期高级架构系统教程学习:https://ke.qq.com/course/417774?flowToken=1020253