为什么企业网站不是开源系统湖南长沙疫情最新情况

死锁及超时检测

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

案例背景

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

1. T1 锁定 R1：然后尝试锁定 R2。
2. T2 锁定 R2：然后尝试锁定 R1。
3. T3 尝试锁定 R1。

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

执行过程

1. T1 锁定 R1
   • T1 请求锁定资源 R1。
   • 因为 R1 未被占用，所以 LockTable 将 R1 锁定给 T1。
   • T1 继续执行，准备请求锁定 R2。
2. T2 锁定 R2
   • T2 请求锁定资源 R2。
   • 因为 R2 未被占用，所以 LockTable 将 R2 锁定给 T2。
   • T2 继续执行，准备请求锁定 R1。
3. T1 请求锁定 R2
   • T1 请求锁定 R2。
   • 由于 R2 已被 T2 锁定，T1 被加入到 R2 的等待队列中。
4. T2 请求锁定 R1
   • T2 请求锁定 R1。
   • 由于 R1 已被 T1 锁定，T2 被加入到 R1 的等待队列中。
   • 现在形成了 T1 → R2 → T2 → R1 → T1 的循环等待，导致死锁。
5. 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正在等待的UIDprivate Lock lock;...
}

锁请求与等待管理 add

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

1. 检查当前事务xid是否已经获得了数据uid（isInList(x2u, xid, uid)）

2. 判断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列表

3. 被持有则等待，向依赖等待图中添加当前等待关系，即waitU.put(xid, uid);，之后进行死锁检测hasDeadLock。如果检测到死锁，就撤销这条边，不允许添加，并撤销该事务。

    // 不需要等待则返回null，否则返回锁对象// 会造成死锁则抛出异常public Lock add(long xid, long uid) throws Exception {// 1. 加锁lock.lock();try {// 2. xid是否已经获得了uid，是则说明不需要等待，返回nullif(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来占用uidif(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正在等待的UIDprivate 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对应的stampprivate int stamp; // 当前stampprivate 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等待的资源uidif(uid == null) return false; // 没有等待的资源Long x = u2x.get(uid); //找到占用资源uid的事务xassert 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所占用的资源列表uidsList<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来占用uidprivate void selectNewXID(long uid) {// u2x:uid被某个xid持有// 1. 从u2x中释放uid，标记uid不再被xid占用u2x.remove(uid);// 2. 获取该uid的等待队列xidsList<Long> xids = wait.get(uid); // wait:uid被哪些xid等待if(xids == null) return;assert xids.size() > 0;// 3. 从等待队列中选择一个xid来占用uidwhile(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);}

参考资料

死锁及超时检测 | EasyDB (blockcloth.cn)

MYDB 7. 死锁检测与 VM 的实现 | 信也のブログ (shinya.click)