第15课：死锁检测

事务与并发 第15课 / 共25课

📖 课程概述

死锁是并发系统中的经典问题：两个或多个事务互相等待对方持有的锁，导致所有事务都无法继续。数据库必须能够检测死锁并选择牺牲者进行回滚。本课实现等待图死锁检测、超时机制和预防策略。

🔄 死锁原理

💻 C语言实现：死锁检测器

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <stdint.h>

#define MAX_TXN      32
#define MAX_RESOURCE 64
#define MAX_EDGES    256

typedef uint32_t TxnID;

// 等待图
typedef struct {
    TxnID from;
    TxnID to;
    char  resource[MAX_RESOURCE];  // 等待的资源
} WaitEdge;

// 锁持有信息
typedef struct {
    TxnID txn_id;
    char  resource[MAX_RESOURCE];
    int   mode;  // 0=S, 1=X
} LockHold;

typedef struct {
    WaitEdge  edges[MAX_EDGES];
    int       num_edges;
    LockHold  locks[MAX_TXN * 4];
    int       num_locks;
    TxnID     wait_for[MAX_TXN];  // 每个事务在等待哪个事务
    int       deadlocks_detected;
    int       victims_chosen;
} DeadlockDetector;

DeadlockDetector* dd_create() {
    DeadlockDetector* dd = calloc(1, sizeof(DeadlockDetector));
    printf("[Deadlock] 死锁检测器初始化\n");
    return dd;
}

// 添加锁持有
void dd_lock_held(DeadlockDetector* dd, TxnID txn, const char* resource, int mode) {
    dd->locks[dd->num_locks].txn_id = txn;
    strncpy(dd->locks[dd->num_locks].resource, resource, MAX_RESOURCE - 1);
    dd->locks[dd->num_locks].mode = mode;
    dd->num_locks++;
}

// 添加等待边
void dd_add_wait(DeadlockDetector* dd, TxnID from, TxnID to, const char* resource) {
    dd->edges[dd->num_edges].from = from;
    dd->edges[dd->num_edges].to = to;
    strncpy(dd->edges[dd->num_edges].resource, resource, MAX_RESOURCE - 1);
    dd->num_edges++;
    dd->wait_for[from] = to;
    printf("  [Wait] TXN %u → TXN %u (等待 %s)\n", from, to, resource);
}

// DFS检测环
int dfs_visit(DeadlockDetector* dd, TxnID node, int* visited,
              int* on_stack, TxnID* path, int* path_len) {
    visited[node] = 1;
    on_stack[node] = 1;
    path[(*path_len)++] = node;

    for (int i = 0; i < dd->num_edges; i++) {
        if (dd->edges[i].from != node) continue;
        TxnID next = dd->edges[i].to;
        if (!visited[next]) {
            if (dfs_visit(dd, next, visited, on_stack, path, path_len))
                return 1;
        } else if (on_stack[next]) {
            // 发现环!
            path[(*path_len)++] = next;
            return 1;
        }
    }

    on_stack[node] = 0;
    (*path_len)--;
    return 0;
}

// 检测死锁
int dd_detect(DeadlockDetector* dd, TxnID** cycle, int* cycle_len) {
    int visited[MAX_TXN] = {0};
    int on_stack[MAX_TXN] = {0};
    TxnID path[MAX_TXN];
    int path_len = 0;

    // 获取所有活跃事务ID
    TxnID txns[MAX_TXN];
    int n = 0;
    for (int i = 0; i < dd->num_edges; i++) {
        int found = 0;
        for (int j = 0; j < n; j++)
            if (txns[j] == dd->edges[i].from) { found = 1; break; }
        if (!found) txns[n++] = dd->edges[i].from;
    }

    for (int i = 0; i < n; i++) {
        path_len = 0;
        if (!visited[txns[i]]) {
            if (dfs_visit(dd, txns[i], visited, on_stack, path, &path_len)) {
                // 提取环
                TxnID cycle_start = path[path_len - 1];
                int start = 0;
                for (int j = 0; j < path_len - 1; j++) {
                    if (path[j] == cycle_start) { start = j; break; }
                }
                *cycle_len = path_len - start;
                *cycle = &path[start];
                dd->deadlocks_detected++;
                return 1;
            }
        }
    }
    return 0;
}

// 选择牺牲者(简单策略: 最小事务ID)
TxnID choose_victim(DeadlockDetector* dd, TxnID* cycle, int cycle_len) {
    TxnID victim = cycle[0];
    for (int i = 1; i < cycle_len; i++) {
        if (cycle[i] < victim) victim = cycle[i];
    }
    dd->victims_chosen++;
    return victim;
}

// 解除死锁
void resolve_deadlock(DeadlockDetector* dd, TxnID victim) {
    printf("  [Deadlock] 选择牺牲者: TXN %u\n", victim);
    // 移除victim的锁和等待边
    int new_locks = 0;
    for (int i = 0; i < dd->num_locks; i++) {
        if (dd->locks[i].txn_id != victim)
            dd->locks[new_locks++] = dd->locks[i];
    }
    dd->num_locks = new_locks;

    int new_edges = 0;
    for (int i = 0; i < dd->num_edges; i++) {
        if (dd->edges[i].from != victim && dd->edges[i].to != victim)
            dd->edges[new_edges++] = dd->edges[i];
    }
    dd->num_edges = new_edges;
    dd->wait_for[victim] = 0;
}

// 打印等待图
void dd_print_graph(DeadlockDetector* dd) {
    printf("  等待图: ");
    for (int i = 0; i < dd->num_edges; i++)
        printf("TXN%u→TXN%u ", dd->edges[i].from, dd->edges[i].to);
    printf("\n");
    printf("  持有锁: ");
    for (int i = 0; i < dd->num_locks; i++)
        printf("[TXN%u:%s:%s] ", dd->locks[i].txn_id, dd->locks[i].resource,
               dd->locks[i].mode ? "X" : "S");
    printf("\n");
}

int main() {
    printf("╔══════════════════════════════════════╗\n");
    printf("║   死锁检测与恢复                     ║\n");
    printf("╚══════════════════════════════════════╝\n\n");

    DeadlockDetector* dd = dd_create();

    // 场景1: 简单两事务死锁
    printf("--- 场景1: 两事务死锁 ---\n");
    dd_lock_held(dd, 1, "R1", 1);  // TXN1持有R1(X)
    dd_lock_held(dd, 2, "R2", 1);  // TXN2持有R2(X)
    dd_add_wait(dd, 1, 2, "R2");   // TXN1等待R2
    dd_add_wait(dd, 2, 1, "R1");   // TXN2等待R1

    dd_print_graph(dd);

    TxnID* cycle;
    int cycle_len;
    if (dd_detect(dd, &cycle, &cycle_len)) {
        printf("  ⚠️ 检测到死锁! 环: ");
        for (int i = 0; i < cycle_len; i++) printf("TXN%u ", cycle[i]);
        printf("\n");
        TxnID victim = choose_victim(dd, cycle, cycle_len);
        resolve_deadlock(dd, victim);
    }

    dd_print_graph(dd);

    // 场景2: 三事务循环死锁
    printf("\n--- 场景2: 三事务循环死锁 ---\n");
    DeadlockDetector* dd2 = dd_create();
    dd2_lock_held(dd2, 1, "R1", 1);
    dd2_lock_held(dd2, 2, "R2", 1);
    dd2_lock_held(dd2, 3, "R3", 1);
    dd2_add_wait(dd2, 1, 2, "R2");
    dd2_add_wait(dd2, 2, 3, "R3");
    dd2_add_wait(dd2, 3, 1, "R1");

    dd2_print_graph(dd2);

    if (dd2_detect(dd2, &cycle, &cycle_len)) {
        printf("  ⚠️ 检测到死锁! 环: ");
        for (int i = 0; i < cycle_len; i++) printf("TXN%u ", cycle[i]);
        printf("\n");
        TxnID victim = choose_victim(dd2, cycle, cycle_len);
        resolve_deadlock(dd2, victim);
    }
    dd2_print_graph(dd2);

    // 再次检测
    if (!dd2_detect(dd2, &cycle, &cycle_len)) {
        printf("  ✅ 死锁已解除\n");
    }

    printf("\n=== 统计 ===\n");
    printf("场景1: 检测%d次, 牺牲%d个事务\n", dd->deadlocks_detected, dd->victims_chosen);
    printf("场景2: 检测%d次, 牺牲%d个事务\n", dd2->deadlocks_detected, dd2->victims_chosen);

    printf("\n✅ 死锁检测器运行完成\n");
    return 0;
}

🐍 Python实现：死锁模拟与预防

"""
死锁模拟: 不同预防策略的对比
"""
import random, time
from collections import defaultdict

class WaitGraph:
    def __init__(self):
        self.edges = []  # (from, to, resource)
        self.locks = {}  # resource → (txn, mode)

    def lock(self, txn, resource, mode):
        if resource in self.locks and self.locks[resource][0] != txn:
            holder = self.locks[resource][0]
            self.edges.append((txn, holder, resource))
            return False  # 等待
        self.locks[resource] = (txn, mode)
        return True  # 获得锁

    def unlock_all(self, txn):
        to_remove = [r for r, (t, _) in self.locks.items() if t == txn]
        for r in to_remove:
            del self.locks[r]
        self.edges = [(f, t, r) for f, t, r in self.edges if f != txn and t != txn]

    def detect_cycle(self):
        """BFS检测环"""
        graph = defaultdict(list)
        for f, t, r in self.edges:
            graph[f].append(t)
        for start in graph:
            visited = set()
            queue = [(start, [start])]
            while queue:
                node, path = queue.pop(0)
                for next_node in graph[node]:
                    if next_node == start and len(path) > 1:
                        return path
                    if next_node not in visited:
                        visited.add(next_node)
                        queue.append((next_node, path + [next_node]))
        return None

class DeadlockSimulator:
    """模拟不同策略下的死锁频率"""
    def __init__(self, num_txns=10, num_resources=20):
        self.num_txns = num_txns
        self.num_resources = num_resources

    def simulate(self, strategy="none", num_rounds=1000):
        deadlocks = 0
        successes = 0
        for _ in range(num_rounds):
            wg = WaitGraph()
            txns = list(range(1, self.num_txns + 1))
            random.shuffle(txns)

            for txn in txns:
                # 每个事务随机获取2个资源
                if strategy == "ordered":
                    # 预防策略: 按资源ID顺序获取
                    resources = sorted(random.sample(range(self.num_resources), 2))
                else:
                    resources = random.sample(range(self.num_resources), 2)

                got_all = True
                for r in resources:
                    if not wg.lock(txn, f"R{r}", "X"):
                        got_all = False
                        break

                if got_all:
                    successes += 1
                else:
                    cycle = wg.detect_cycle()
                    if cycle:
                        deadlocks += 1
                        # 回滚一个事务
                        victim = min(cycle)
                        wg.unlock_all(victim)

        return {"successes": successes, "deadlocks": deadlocks,
                "deadlock_rate": deadlocks / num_rounds * 100}

sim = DeadlockSimulator(num_txns=10, num_resources=20)
print("=== 死锁预防策略对比 ===")
print(f"{'策略':>15} | {'成功':>6} | {'死锁':>6} | {'死锁率':>8}")
print("-" * 50)
for strategy, name in [("none", "无预防"), ("ordered", "有序获取")]:
    r = sim.simulate(strategy=strategy)
    print(f"{name:>15} | {r['successes']:>6} | {r['deadlocks']:>6} | {r['deadlock_rate']:>6.1f}%")

print("\n✅ 死锁模拟完成")

🔑 关键概念总结