当前位置：首页 > article >正文

【MyDB】4-VersionManager 之 3-死锁及超时检测

article 2025/2/2 20:05:26

【MyDB】4-VersionManager 之 3-死锁及超时检测

死锁及超时检测
- 案例背景
- LockTable
- 锁请求与等待管理 `add`
- - vm调用add
  - putIntoList，isInList，removeFromList
- 死锁检测 hasDeadLock方法
- 资源释放与重分配
参考资料

死锁及超时检测

本章涉及代码：top/xianghua/mydb/server/vm/LockTable.java

案例背景

学习如何使用 LockTable 进行锁管理、死锁检测与解决之前，先了解一下死锁是如何发生的。假设我们有三个事务 T1、T2 和 T3，它们分别要访问两个资源 R1 和 R2。事务的执行顺序如下：

T1 锁定 R1：然后尝试锁定 R2。
T2 锁定 R2：然后尝试锁定 R1。
T3 尝试锁定 R1。

这种情况下，T1 和 T2 之间会产生死锁，而 T3 将会被阻塞在等待 R1 上。

执行过程

T1 锁定 R1
- T1 请求锁定资源 R1。
- 因为 R1 未被占用，所以 LockTable 将 R1 锁定给 T1。
- T1 继续执行，准备请求锁定 R2。
T2 锁定 R2
- T2 请求锁定资源 R2。
- 因为 R2 未被占用，所以 LockTable 将 R2 锁定给 T2。
- T2 继续执行，准备请求锁定 R1。
T1 请求锁定 R2
- T1 请求锁定 R2。
- 由于 R2 已被 T2 锁定，T1 被加入到 R2 的等待队列中。
T2 请求锁定 R1
- T2 请求锁定 R1。
- 由于 R1 已被 T1 锁定，T2 被加入到 R1 的等待队列中。
- 现在形成了 T1 → R2 → T2 → R1 → T1 的循环等待，导致死锁。
T3 尝试锁定 R1
- T3 请求锁定 R1。
- 由于 R1 已被 T1 锁定且 T2 在等待 R1，T3 被加入到 R1 的等待队列中。

LockTable

上一章提到了2PL会阻塞事务，直至持有锁的线程释放锁。这样就可能出现死锁。因此，在MyDB中需要能检测死锁。

我们可以将事务之间的这种等待关系抽象成有向边，例如 $T_j$ 在等待 $T_i$ 。这样，无数的有向边就可以形成一个图（不一定是连通图）。检测死锁的话只需要查看图中是否有环。

就MyDB而言，使用了一个LockTabel对象，通过在内存中维护这张图，而记录事务的等待关系。

注意，x2u和wait是列表，分别记录xid已经获得的资源的uid列表，和正在等待某个uid的xid列表。

public class LockTable {

    private Map<Long, List<Long>> x2u;  // 某个XID已经获得的资源的UID列表
    private Map<Long, Long> u2x;        // UID被某个XID持有
    private Map<Long, List<Long>> wait; // 正在等待UID的XID列表
    private Map<Long, Lock> waitLock;   // 正在等待资源的XID的锁
    private Map<Long, Long> waitU;      // XID正在等待的UID
    private Lock lock;

    ...
}

锁请求与等待管理 `add`

当一个事务请求获取某个资源时，LockTable 首先会检查该资源是否已被其他事务持有。如果没有被持有，资源将直接分配给请求的事务。如果资源已被占用，事务将进入等待状态，并存储在相应的等待队列中。

检查当前事务xid是否已经获得了数据uid（isInList(x2u, xid, uid)）
判断uid是否被其他xid持有，u2x.containsKey(uid)，假如未被其他xid持有，则直接获得

u2x.put(uid, xid); // Map<Long, Long> u2x; UID被某个XID持有

putIntoList(x2u, xid, uid); //Map<Long, List> x2u 某个XID已经获得的资源的UID列表
被持有则等待，向依赖等待图中添加当前等待关系，即waitU.put(xid, uid);，之后进行死锁检测hasDeadLock。如果检测到死锁，就撤销这条边，不允许添加，并撤销该事务。

    // 不需要等待则返回null，否则返回锁对象
    // 会造成死锁则抛出异常
    public Lock add(long xid, long uid) throws Exception {
        // 1. 加锁
        lock.lock();
        try {
            // 2. xid是否已经获得了uid，是则说明不需要等待，返回null
            if(isInList(x2u, xid, uid)) {
                return null;
            }
            // 3. uid是否被其他xid持有
            if(!u2x.containsKey(uid)) {
                // 3.1 uid未被其他xid持有，则直接获得
                u2x.put(uid, xid);
                putIntoList(x2u, xid, uid);
                return null;
            }
            // 3.2 uid被其他xid持有，则等待
            waitU.put(xid, uid);
            //putIntoList(wait, xid, uid);
            putIntoList(wait, uid, xid); // 这里是为了方便死锁检测，因为死锁检测是从等待队列中选择一个xid来占用uid
            if(hasDeadLock()) {
                waitU.remove(xid);
                removeFromList(wait, uid, xid);
                throw Error.DeadlockException;
            }
            Lock l = new ReentrantLock();
            l.lock();
            waitLock.put(xid, l);
            return l;

        } finally {
            lock.unlock();
        }
    }

vm调用add

在vm中，当某个事务xid需要资源uid时，就会调用add，可以看到，add方法需要等待的时候，会返回一个上了锁的Lock对象。调用方在获取到该对象时，需要尝试获取该对象的锁，由此实现阻塞线程的目的，例如：

Lock l = lt.add(xid, uid);
if(l != null) {
    l.lock();   // 阻塞在这一步
    l.unlock();
}

putIntoList，isInList，removeFromList

由于之前说过，我们再复习一下LockTable中的x2u和wait是long->List ，

private Map<Long, List> x2u; // 某个XID已经获得的资源的UID列表

private Map<Long, List> wait; // 正在等待UID的XID列表

那当我们需要知道事务是否已经获得资源uid时，由于x2u.get(xid)得到的时一个资源list，因此，需要使用inInList来判断，uid是否在xid已经获得的资源list中。

同理，当需要向x2u中添加某个xid已经获得的资源uid，也不能直接添加，而是要先获得x2u已经获得的资源的uid list，之后再向其中添加，就用到了putIntoList方法。

public class LockTable {

    private Map<Long, List<Long>> x2u;  // 某个XID已经获得的资源的UID列表
    private Map<Long, Long> u2x;        // UID被某个XID持有
    private Map<Long, List<Long>> wait; // 正在等待UID的XID列表
    private Map<Long, Lock> waitLock;   // 正在等待资源的XID的锁
    private Map<Long, Long> waitU;      // XID正在等待的UID
    private Lock lock;

    ...
}

putIntoList

    /**
     * 将uid1添加到uid0对应的列表中
     * @param listMap
     * @param id0
     * @param id1
     */
    private void putIntoList(Map<Long, List<Long>> listMap, long id0, long id1) {
        // 如果id0对应的列表不存在，则创建一个新的列表
        if(!listMap.containsKey(id0)) {
            listMap.put(id0, new ArrayList<>());
        }
        // 将id1添加到id0对应的列表中
        listMap.get(id0).add(0, id1);
    }

isInList(listMap,id0,id1)id1是否在id0对应的列表中


    /**
     * 判断id1是否在id0对应的列表中
     * @param listMap
     * @param id0
     * @param id1
     * @return
     */
    private boolean isInList(Map<Long, List<Long>> listMap, long id0, long id1) {
        List<Long> l = listMap.get(id0);
        if(l == null) return false;
        Iterator<Long> i = l.iterator();
        while(i.hasNext()) {
            long e = i.next();
            if(e == id1) {
                return true;
            }
        }
        return false;
    }

removeFromList从uid0对应的列表中删除uid1

    /**
     * 从uid0对应的列表中删除uid1
     * @param listMap
     * @param uid0
     * @param uid1
     */
    private void removeFromList(Map<Long, List<Long>> listMap, long uid0, long uid1) {
        List<Long> l = listMap.get(uid0);
        if(l == null) return;
        Iterator<Long> i = l.iterator();
        while(i.hasNext()) {
            long e = i.next();
            if(e == uid1) {
                i.remove();
                break;
            }
        }
        if(l.size() == 0) {
            listMap.remove(uid0);
        }
    }

死锁检测 hasDeadLock方法

为了避免死锁，LockTable 实现了基于深度优先搜索（DFS）的死锁检测机制。通过遍历事务依赖图，系统可以检测到是否存在循环依赖，从而识别死锁。

需要注意这个图不一定是连通图。思路就是为每个节点设置一个访问戳xidStamp，都初始化为 -1，随后遍历所有节点，以每个非 -1 的节点作为根进行深搜，并将深搜该连通图中遇到的所有节点都设置为同一个数字，不同的连通图数字不同。这样，如果在遍历某个图时，遇到了之前遍历过的节点，说明出现了环。

在这里插入图片描述

    private Map<Long, Integer> xidStamp; // XID对应的stamp
    private int stamp; // 当前stamp

    private boolean hasDeadLock() {
        xidStamp = new HashMap<>();
        stamp = 1;
        for(long xid : x2u.keySet()) {
            Integer s = xidStamp.get(xid);
            if(s != null && s > 0) {
                continue;
            }
            stamp ++;
            if(dfs(xid)) {
                return true;
            }
        }
        return false;
    }

    private boolean dfs(long xid) {
        Integer stp = xidStamp.get(xid);
        if(stp != null && stp == stamp) {
            return true;
        }
        if(stp != null && stp < stamp) {
            return false;
        }
        xidStamp.put(xid, stamp);

        Long uid = waitU.get(xid); // 得到事务xid等待的资源uid
        if(uid == null) return false; // 没有等待的资源
        Long x = u2x.get(uid); //找到占用资源uid的事务x
        assert x != null;
        return dfs(x); // 继续搜索事务x
    }

可以在LockTabelTest中运行testLockTable以检测死锁的发生

    public void testLockTable() {
        LockTable lt = new LockTable();
        try {
            lt.add(1, 1);
        } catch (Exception e) {
            Panic.panic(e);
        }

        try {
            lt.add(2, 2);
        } catch (Exception e) {
            Panic.panic(e);
        }

        try {
            lt.add(2, 1);
        } catch (Exception e) {
            Panic.panic(e);
        }
        try {
            lt.add(1, 2);
        } catch (Exception e) {
            Panic.panic(e);
        }
        
        assertThrows(RuntimeException.class, ()->lt.add(1, 2));
    }

资源释放与重分配

在一个事务 commit 或者 abort 时，就可以释放所有它持有的锁，并将自身从等待图中删除。然后将资源锁重新分配给等待队列中的其他事务。

while 循环释放掉了这个线程所有持有的资源的锁，这些资源可以被等待的线程所获取：

    public void remove(long xid) {
        lock.lock();
        try {
            // 1. 释放xid所占用的资源列表uids
            List<Long> uids = x2u.get(xid);
            if(uids != null) {
                while(uids.size() > 0) {
                    Long uid = uids.remove(0);
                    selectNewXID(uid); //   从等待队列中选择一个xid来占用uid
                }
            }
            // 2. 在waitU(正在等待资源的xid)，x2u(某个XID已经获得的资源的UID列表),waitLock中删除xid:
            waitU.remove(xid);
            x2u.remove(xid);
            waitLock.remove(xid);
        } finally {
            lock.unlock();
        }
    }

selectNewXID

从 List 开头开始尝试解锁，还是个公平锁。解锁时，将该 Lock 对象 unlock 即可，这样业务线程就获取到了锁，就可以继续执行了。

    // 从等待队列中选择一个xid来占用uid
    private void selectNewXID(long uid) {
        // u2x:uid被某个xid持有
        // 1. 从u2x中释放uid，标记uid不再被xid占用
        u2x.remove(uid);

        // 2. 获取该uid的等待队列xids
        List<Long> xids = wait.get(uid); // wait:uid被哪些xid等待
        if(xids == null) return;
        assert xids.size() > 0;

        // 3. 从等待队列中选择一个xid来占用uid
        while(xids.size() > 0) {
            long xid = xids.remove(0);
            if(!waitLock.containsKey(xid)) {
                continue;
            } else {
                u2x.put(uid, xid);
                Lock lo = waitLock.remove(xid);
                waitU.remove(xid);
                lo.unlock();
                break;
            }
        }

        if(xids.size() == 0) wait.remove(uid);
    }