前言

死锁(deadlock)是并发程序里最难排查的一类问题:它不会崩溃、不会报错,只会让程序静止——CPU 占用为零,日志停在某一行,重启后又可能不复现。很多人以为死锁检测是操作系统内核的事,但在用户态自己实现一套检测机制,对定位线上问题、验证锁设计、甚至做自研调度器都非常有价值。

本文先讲清死锁的四个必要条件与资源分配图,再给出三种层次的检测思路:加锁顺序检测(静态)、超时检测(动态)、等待图环检测(运行时),最后给一份可编译运行的等待图检测实现。

一、死锁的四个必要条件

Coffman 在 1971 年总结的四个条件,必须同时满足才可能死锁,因此破坏其中任意一个就能避免死锁:

条件含义破坏手段
互斥(mutual exclusion)资源同一时刻只能被一个线程占用用原子操作/无锁结构替代锁
持有并等待(hold and wait)持有资源的同时等待其他资源一次性申请所有资源
不可抢占(no preemption)资源只能由持有者主动释放用 try_lock 失败就释放已持有的锁
循环等待(circular wait)存在线程-资源的环形等待链全局固定加锁顺序

工程上最实用、代价最低的是破坏循环等待:给所有锁编号,永远按编号从小到大申请。

// ❌ 加锁顺序相反,构成循环等待

// 线程1: lock(A); lock(B);

// 线程2: lock(B); lock(A);



// ✅ 全局统一顺序:永远先 A 后 B

void transfer(Account& from, Account& to) {

    Account& first  = (&from < &to) ? from : to;   // 按地址定序

    Account& second = (&from < &to) ? to : from;

    std::scoped_lock lock(first.mtx, second.mtx);  // scoped_lock 内部也会排序

}

但顺序法有前提:所有路径都必须遵守。一旦有人漏掉,死锁仍会发生。所以还需要检测作为兜底。

二、资源分配图与等待图

把"线程 → 等待 → 锁"抽象成有向图:


  • 资源分配图(resource allocation graph):节点是线程和资源,边有"申请边"和"分配边"。

  • 等待图(wait-for graph):把资源节点消掉,边变成"线程 A 等待线程 B 持有的锁"。等待图中存在环 ⟺ 存在死锁。


  •  

因此检测算法可以归结为一句话:在等待图里找环。

对于单实例资源(普通 mutex),等待图有环就等价于死锁。对于多实例资源(信号量),需要用银行家算法那一类矩阵方法,但工程中绝大多数死锁是互斥量造成的,等待图足够。

三、三种检测层次

3.1 静态检测:加锁顺序分析

在编译期或代码审查阶段,扫描每个函数的加锁序列,若发现 A→B 与 B→A 同时存在,就报警。工具如 Clang Thread Safety Analysis(-Wthread-safety)能标注 capability 并检查顺序。

// 用注解让编译器帮你检查

#include <mutex>

class Bank {

    std::mutex mtx_a_ __attribute__((capability("mtx_a_")));

    std::mutex mtx_b_ __attribute__((capability("mtx_b_")));

public:

    void ok() __attribute__((requires_capability(mtx_a_))) {

        std::lock_guard<std::mutex> b(mtx_b_);   // 编译器会警告顺序问题

    }

};

静态检测的优点是零运行时开销,缺点是无法覆盖多态、回调、条件分支导致的动态顺序。

3.2 动态检测:超时

最简单粗暴的运行时兜底——给锁操作加超时,超时即认为可能死锁,打印所有线程的调用栈后退出或恢复。

#include <mutex>

#include <chrono>

#include <cstdio>



std::timed_mutex mtx;



void work() {

    if (!mtx.try_lock_for(std::chrono::milliseconds(500))) {

        std::fprintf(stderr, "possible deadlock: lock timeout\n");

        // 此处可 dump 所有线程栈、上报监控

        return;

    }

    std::lock_guard<std::timed_mutex> guard(mtx, std::adopt_lock);

    // ... critical section ...

}

超时法的误报率高:慢 IO、调度延迟都可能触发。它适合做"监控告警"而不是"判定死锁"。

3.3 运行时检测:等待图环检测

这是最正统的思路。核心是在每次加锁时记录等待关系,然后判断是否成环。实现要点:


  1. 全局记录两个映射:owner[lock] = thread、waiting[thread] = lock。

  2. 加锁前,在受保护的元数据里登记"线程 T 正在等锁 L"。

  3. 沿 waiting → owner 交替跳跃,看能否回到起点(即回到 T)。

  4. 若成环,说明 T 一旦阻塞就会死锁,可以选择拒绝加锁(报错/抛异常),从而避免死锁。


  5.  

四、代码实战:一个可运行的等待图检测器

下面实现一个"会自己发现死锁并报告"的互斥量。为保持可编译,用 std::mutex 保护元数据,用 id 标识线程。

#include <mutex>

#include <thread>

#include <unordered_map>

#include <vector>

#include <string>

#include <iostream>

#include <stdexcept>



// 全局元数据:谁持有锁、谁在等锁

class DeadlockDetector {

public:

    static DeadlockDetector& instance() {

        static DeadlockDetector d;

        return d;

    }



    // 返回 true 表示会成环(即将死锁)

    bool register_wait(uint64_t tid, const void* lock) {

        std::lock_guard<std::mutex> g(mtx_);

        // 沿 waiting -> owner 链条回溯,看是否回到 tid

        uint64_t cur = tid;

        const void* cur_lock = lock;

        int guard = 0;

        while (cur_lock && guard++ < 1024) {

            auto ow = owner_.find(cur_lock);

            if (ow == owner_.end()) return false;      // 锁未被持有,不会成环

            cur = ow->second;

            if (cur == tid) return true;               // 回到自己 -> 环!

            auto wt = waiting_.find(cur);

            if (wt == waiting_.end()) return false;    // 该线程没在等别的锁

            cur_lock = wt->second;

        }

        return false;

    }



    void register_acquire(uint64_t tid, const void* lock) {

        std::lock_guard<std::mutex> g(mtx_);

        waiting_.erase(tid);

        owner_[lock] = tid;

    }



    void register_release(const void* lock) {

        std::lock_guard<std::mutex> g(mtx_);

        owner_.erase(lock);

    }



    void set_waiting(uint64_t tid, const void* lock) {

        std::lock_guard<std::mutex> g(mtx_);

        waiting_[tid] = lock;

    }



private:

    std::mutex mtx_;

    std::unordered_map<const void*, uint64_t> owner_;     // lock -> tid

    std::unordered_map<uint64_t, const void*> waiting_;   // tid  -> lock

};



// 把线程 id 归一化为 uint64

static uint64_t self_id() {

    return static_cast<uint64_t>(std::hash<std::thread::id>{}(std::this_thread::get_id()));

}



// 带死锁检测的互斥量

class CheckedMutex {

public:

    void lock() {

        const uint64_t tid = self_id();

        const void* self = this;



        DeadlockDetector::instance().set_waiting(tid, self);

        if (DeadlockDetector::instance().register_wait(tid, self)) {

            std::cerr << "DEADLOCK DETECTED: thread would block on a cycle, aborting lock\n";

            throw std::runtime_error("deadlock cycle detected");

        }

        mtx_.lock();

        DeadlockDetector::instance().register_acquire(tid, self);

    }



    void unlock() {

        DeadlockDetector::instance().register_release(this);

        mtx_.unlock();

    }



private:

    std::mutex mtx_;

};



// ---- 演示:复现经典 AB/BA 死锁 ----

CheckedMutex A, B;



void thread1() {

    try {

        std::lock_guard<CheckedMutex> a(A);

        std::this_thread::sleep_for(std::chrono::milliseconds(50));  // 放大交错概率

        std::lock_guard<CheckedMutex> b(B);                          // 这里会被检测到

        std::cout << "thread1 got both\n";

    } catch (const std::exception& e) {

        std::cout << "thread1: " << e.what() << '\n';

    }

}



void thread2() {

    try {

        std::lock_guard<CheckedMutex> b(B);

        std::this_thread::sleep_for(std::chrono::milliseconds(50));

        std::lock_guard<CheckedMutex> a(A);                          // 这里会被检测到

        std::cout << "thread2 got both\n";

    } catch (const std::exception& e) {

        std::cout << "thread2: " << e.what() << '\n';

    }

}



int main() {

    std::thread t1(thread1), t2(thread2);

    t1.join(); t2.join();

    std::cout << "program survived (no real deadlock)\n";

}

不检测时,这段代码几乎必然死锁、程序永远挂起;加上检测后,其中一个线程会在真正阻塞前抛异常退出,另一个拿到两把锁继续跑完,程序正常结束。

实现细节与局限

关注点说明
检测时机加锁前检测。等 lock() 真阻塞了就晚了
元数据锁检测器自身也用了 std::mutex,必须只做内存操作,不能在里面做 IO
误报只对"同一时刻真实持有"的边建图,try_lock 成功的情况不会登记等待
性能每次加锁多一次图遍历,O(链长),适合调试/灰度,不适合极致性能路径
多实例资源信号量、shared_mutex 需扩展为计数模型

其他值得知道的现实手段:


  • gdb 附加:thread apply all bt 看所有线程栈,若多个线程都停在 pthread_mutex_lock 且互相等待,基本可确认死锁。

  • TSan:-fsanitize=thread 能在测试阶段捕获锁顺序反转(lock-order-inversion),这是最推荐的日常防线。

  • try_lock + 回退:加锁失败时释放已持有的锁、退避后重试,用"活锁风险"换"死锁免疫"。


  •  

常见坑点

坑 1:在检测器内部再加锁导致自死锁。 检测器的 register_wait 自己拿 mtx_,如果调用方在持有 mtx_ 时又触发检测,就会自锁。检测路径必须与业务路径分离,且检测元数据锁永远只能短暂持有。

坑 2:把 try_lock 的失败当作死锁。 try_lock 返回 false 只是"当前被占用",不代表死锁。用它做超时监控可以,做死锁判定会大量误报。

坑 3:忘记在异常路径上更新元数据。 如果 mtx_.lock() 抛异常(或线程被取消)而 waiting_ 没清理,检测器会保留一条幽灵等待边,后续误报。所以 set_waiting 登记后,失败路径必须清理。

坑 4:std::recursive_mutex 让检测失效。 同一线程重复加同一把锁在等待图里会形成自环,但那不是死锁。检测器必须知道锁是否可重入。

坑 5:静态变量的初始化顺序。 检测器用 static 单例,若在全局对象构造期(main 之前)就加锁,可能遇到"检测器还没构造"的问题——改用函数内 static(magic static)可解决。

坑 6:只检测不修复。 检测到环后只是 std::cerr 打印,然后继续阻塞,程序依然挂。必须让检测路径拒绝加锁(返回错误/抛异常),把死锁转化为可处理的错误。

坑 7:跨进程死锁不适用。 本文的等待图只在单进程内有效。跨进程(如共享内存里的锁、数据库行锁)需要分布式死锁检测,通常靠超时 + 事务回滚。

总结


  • 死锁成立的四个必要条件中,破坏循环等待(统一加锁顺序)是工程上最划算的手段。

  • 检测的本质是在等待图里找环:沿"线程等锁 → 锁被谁持有"交替回溯,回到起点即成环。

  • 三种层次各有取舍:静态分析零开销但不全;超时法简单但误报高;等待图检测最准确,代价是每次加锁多一次遍历。

  • 检测必须在加锁前做,并且检测到环要拒绝加锁而不是继续阻塞,否则只是"报告了死锁"而已。

  • 日常最实用的防线是 TSan + 统一锁序,自研检测器适合调试期与自研运行时。


  •  

Logo

openEuler 是由开放原子开源基金会孵化的全场景开源操作系统项目,面向数字基础设施四大核心场景(服务器、云计算、边缘计算、嵌入式),全面支持 ARM、x86、RISC-V、loongArch、PowerPC、SW-64 等多样性计算架构

更多推荐