目录
版本跳跃问题
测试代码
死锁
死锁的造成
死锁检测
测试代码
分析流程
VM其他实现方法
获取缓存
从图中删除事务
删除版本
本节将收尾 VM 层，介绍一下 MVCC 可能导致的版本跳跃问题，以及mysql 如何检测 2PL 导致的死锁，还介绍了一些上一章节没有介绍到的方法。
版本跳跃问题
版本跳跃问题，考虑如下的情况，假设 X 最初只有 x0 版本，T1 和 T2 都是可重复读的隔离级别：
T1 begin T2 begin R1(X) // T1读取x0 R2(X) // T2读取x0 U1(X) // T1将X更新到x1 T1 commit U2(X) // T2将X更新到x2 T2 commit
这种情况实际运行起来是没问题的，但是逻辑上不太正确。T1 将 X 从 x0 更新为了 x1，这是没错的。但是 T2 则是将 X 从 x0 更新成了 x2，跳过了 x1 版本。
读提交是允许版本跳跃的，而可重复读则是不允许版本跳跃的。解决版本跳跃的思路也很简单：如果 Ti 需要修改 X，而 X 已经被 Ti 不可见的事务 Tj 修改了，那么要求 Ti 回滚。
if(Visibility.isVersionSkip(tm, t, entry)) {
System.out.println("检查到版本跳跃，自动回滚");
t.err = Error.ConcurrentUpdateException;
internAbort(xid, true);
t.autoAborted = true;
throw t.err;
}
上一节中就总结了，Ti 不可见的 Tj，有两种情况：
- XID(Tj) > XID(Ti)
- Tj in SP(Ti)
于是版本跳跃的检查也就很简单了，取出要修改的数据 X 的最新提交版本，并检查该最新版本的创建者对当前事务是否可见：
public static boolean isVersionSkip(TransactionManager tm, Transaction t, Entry e) {
long xmax = e.getXmax();
if(t.level == 0) {
return false;
} else {
return tm.isCommitted(xmax) && (xmax > t.xid || t.isInSnapshot(xmax));
}
}
测试代码
public class skip {
public static void main(String[] args) throws Exception {
TransactionManagerImpl tm= TransactionManager.open("cun/tm");
DataManager dm=DataManager.open("cun/dm",1 << 20,tm);
VersionManager vm=VersionManager.newVersionManager(tm,dm);
byte b[]=new byte[1122];
byte b_update[]=new byte[1122];
long uid1=vm.insert(0,b);
long xid1=vm.begin(1);
System.out.println("开启事务"+xid1);
long xid2=vm.begin(1);
System.out.println("开启事务"+xid2);
//xid1将x更新到x1
vm.delete(xid1,uid1);
long uid2=vm.insert(xid1,b_update);
vm.commit(xid1);
System.out.println("======");
//xid1将x更新到x2
vm.delete(xid2,uid1);
long uid3=vm.insert(xid2,b_update);
vm.commit(xid2);
}
}
死锁
死锁的造成
死锁的造成
在mysql中，两阶段锁协议（2PL）通常包括扩张和收缩两个阶段。在扩张阶段，事务将获取锁，但不能释放任何锁。在收缩阶段，可以释放现有的锁，但不能获取新的锁，这种规定存在着死锁的风险。
例如：当T1’在扩张阶段，获取了Y的读锁，并读取了Y，此时想要去获取X的写锁，却发现T2’的读锁锁定了X，而T2’也想要获取Y的写锁。简而言之，T1’不得到X是不会释放Y的，T2’不得到Y也是不会释放X的，这便陷入了循环，便形成了死锁。
死锁检测
前面提到了 2PL 会阻塞事务，直至持有锁的线程释放锁。可以将这种等待关系抽象成有向边，例如 Tj 在等待 Ti，就可以表示为 Tj --> Ti。这样，无数有向边就可以形成一个图（不一定是连通图）。检测死锁也就简单了，只需要查看这个图中是否有环即可。
创建一个 LockTable 对象，在内存中维护这张图。维护结构如下：
public class LockTable {
private Map> x2u; // 某个XID已经获得的资源的UID列表
private Map u2x; // UID被某个XID持有
private Map> wait; // 正在等待UID的XID列表
private Map waitLock; // 正在等待资源的XID的锁
private Map waitU; // XID正在等待的UID
private Lock lock;
...
}
在每次出现等待的情况时，就尝试向图中增加一条边，并进行死锁检测。如果检测到死锁，就撤销这条边，不允许添加，并撤销该事务。
// 不需要等待则返回null，否则返回锁对象
// 会造成死锁则抛出异常
public Lock add(long xid, long uid) throws Exception {
lock.lock();
try {
//某个xid已经获得的资源的UID列表，如果在这个列表里面，则不造成死锁，也不需要等待
if(isInList(x2u, xid, uid)) {//x2u xid list
return null;
}
//这里表示有了一个新的uid，则把uid加入到u2x和x2u里面，不死锁，不等待
//u2x uid被某个xid占有
if(!u2x.containsKey(uid)) {//uid xid
u2x.put(uid, xid);
putIntoList(x2u, xid, uid);
return null;
}
//以下就是需要等待的情况
//多个事务等待一个uid的释放
waitU.put(xid, uid);//把需要等待uid的xid添加到等待列表里面
//System.out.println("waitU"+waitU);
putIntoList(wait, uid, xid);//uid list 正在等待uid的xid列表
//造成死锁
if(hasDeadLock()) {
//从等待列表里面删除
waitU.remove(xid);
removeFromList(wait, uid, xid);
throw Error.DeadlockException;
}
//没有造成死锁，但是需要等待新的xid获得锁
Lock l = new ReentrantLock();
l.lock();
waitLock.put(xid, l);
return l;
} finally {
lock.unlock();
}
}
调用 add，如果不需要等待则进行下一步，如果需要等待的话，会返回一个上了锁的 Lock 对象。调用方在获取到该对象时，需要尝试获取该对象的锁，由此实现阻塞线程的目的，如下。
Lock l = lt.add(xid, uid);
if(l != null) {
l.lock(); // 阻塞在这一步
l.unlock();
}
查找图中是否有环的算法实际上就是一个深搜，只是需要注意这个图不一定是连通图。思路就是为每个节点设置一个访问戳，都初始化为 1，随后遍历所有节点，以每个非 1 的节点作为根进行深搜，并将深搜该连通图中遇到的所有节点都设置为同一个数字，不同的连通图数字不同。这样，如果在遍历某个图时，遇到了之前遍历过的节点，说明出现了环。
实现如下：
private boolean hasDeadLock() {
xidStamp = new HashMap<>();
stamp = 1;
System.out.println("xid已经持有哪些uid x2u="+x2u);//xid已经持有哪些uid
System.out.println("uid正在被哪个xid占用 u2x="+u2x);//uid正在被哪个xid占用
for(long xid : x2u.keySet()) {//已经拿到锁的xid
Integer s = xidStamp.get(xid);
if(s != null && s > 0) {
continue;
}
stamp ++;
System.out.println("xid"+xid+"的stamp是"+s);
System.out.println("进入深搜");
if(dfs(xid)) {
return true;
}
}
return false;
}
private boolean dfs(long xid) {
Integer stp = xidStamp.get(xid);
System.out.println("xid"+xid+"的stamp是"+stp);
//遍历某个图时，遇到了之前遍历过的节点，说明出现了环。
if(stp != null && stp == stamp) {
return true;
}
if(stp != null && stp < stamp) {
System.out.println("遇到了前一个图，未成环");
return false;
}
xidStamp.put(xid, stamp);//每个已获得资源的事务一个独特的stamp
System.out.println("xidStamp找不到该xid，加入后xidStamp变为"+xidStamp);
Long uid = waitU.get(xid);//已获得资源的事务xid正在等待的uid
System.out.println("xid"+xid+"正在等待的uid是"+uid);
if(uid == null){
System.out.println("未成环，退出深搜");
return false;//xid没有需要等待的uid,无死锁
}
Long x = u2x.get(uid);//xid需要等待的uid被哪个xid占用了
System.out.println("xid"+xid+"需要的uid被"+"xid"+x+"占用了");
System.out.println("=====再次进入深搜"+"xid"+x+"====");
assert x != null;
return dfs(x);
}
测试代码
public static void main(String[] args) throws Exception {
LockTable lock=new LockTable();
lock.add(1L,3L);
lock.add(2L,4L);
lock.add(3L,5L);
lock.add(1L,4L);
// lock.add(1L,4L);
System.out.println("++++++++++++++++");
// lock.add(3L,4L);
lock.add(2L,5L);
System.out.println("++++++++++++++++");
lock.add(3L,3L);
System.out.println(hasDeadLock());
}
分析流程
为了更好的分析检测死锁的过程。
xid已经持有哪些uid x2u={1=[3], 2=[4], 3=[5]}
uid正在被哪个xid占用 u2x={3=1, 4=2, 5=3}
xid1的stamp是null
进入深搜
xid1的stamp是null
xidStamp找不到该xid，加入后xidStamp变为{1=2}
xid1正在等待的uid是4
xid1需要的uid被xid2占用了
=====再次进入深搜xid2====
xid2的stamp是null
xidStamp找不到该xid，加入后xidStamp变为{1=2, 2=2}
xid2正在等待的uid是null
未成环，退出深搜
xid3的stamp是null
进入深搜
xid3的stamp是null
xidStamp找不到该xid，加入后xidStamp变为{1=2, 2=2, 3=3}
xid3正在等待的uid是null
未成环，退出深搜
++++++++++++++++
xid已经持有哪些uid x2u={1=[3], 2=[4], 3=[5]}
uid正在被哪个xid占用 u2x={3=1, 4=2, 5=3}
xid1的stamp是null
进入深搜
xid1的stamp是null
xidStamp找不到该xid，加入后xidStamp变为{1=2}
xid1正在等待的uid是4
xid1需要的uid被xid2占用了
=====再次进入深搜xid2====
xid2的stamp是null
xidStamp找不到该xid，加入后xidStamp变为{1=2, 2=2}
xid2正在等待的uid是5
xid2需要的uid被xid3占用了
=====再次进入深搜xid3====
xid3的stamp是null
xidStamp找不到该xid，加入后xidStamp变为{1=2, 2=2, 3=2}
xid3正在等待的uid是null
未成环，退出深搜
++++++++++++++++
xid已经持有哪些uid x2u={1=[3], 2=[4], 3=[5]}
uid正在被哪个xid占用 u2x={3=1, 4=2, 5=3}
xid1的stamp是null
进入深搜
xid1的stamp是null
xidStamp找不到该xid，加入后xidStamp变为{1=2}
xid1正在等待的uid是4
xid1需要的uid被xid2占用了
=====再次进入深搜xid2====
xid2的stamp是null
xidStamp找不到该xid，加入后xidStamp变为{1=2, 2=2}
xid2正在等待的uid是5
xid2需要的uid被xid3占用了
=====再次进入深搜xid3====
xid3的stamp是null
xidStamp找不到该xid，加入后xidStamp变为{1=2, 2=2, 3=2}
xid3正在等待的uid是3
xid3需要的uid被xid1占用了
=====再次进入深搜xid1====
xid1的stamp是2
Exception in thread "main" java.lang.RuntimeException: Deadlock!
at utils.Error.(Error.java:20)
at backend.vm.LockTable.add(LockTable.java:61)
at backend.vm.test2.main(test2.java:37)
VM其他实现方法
获取缓存
VM 的实现类还被设计为 Entry 的缓存，需要继承 前面讲的通用缓存框架。实际上，就是在获取到dataItem的基础上再获取Entry。
protected Entry getForCache(long uid) throws Exception {
Entry entry = Entry.loadEntry(this, uid);
if(entry == null) {
throw Error.NullEntryException;
}
return entry;
}
public static Entry loadEntry(VersionManager vm, long uid) throws Exception {
DataItem di = ((VersionManagerImpl)vm).dm.read(uid);
return newEntry(vm, di, uid);
}
释放缓存也是如此，不再赘诉
从图中删除事务
在一个事务 commit 或者 abort 时，就可以释放所有它持有的锁，并将自身从等待图中删除。并从等待队列中选择一个xid来占用uid
解锁时，将该 Lock 对象 unlock 即可，这样其他业务线程就获取到了锁，就可以继续执行了。
public void remove(long xid) {
lock.lock();
try {
List l = x2u.get(xid);
if(l != null) {
while(l.size() > 0) {
Long uid = l.remove(0);
selectNewXID(uid);
}
}
waitU.remove(xid);
x2u.remove(xid);
waitLock.remove(xid);
} finally {
lock.unlock();
}
}
private void selectNewXID(long uid) {
u2x.remove(uid);//uid被某个xid持有
List l = wait.get(uid);//正在等待uid的xid的列表
if(l == null) return;
assert l.size() > 0;
while(l.size() > 0) {
long xid = l.remove(0);//从等待uid的xid的列表中选一个xid出来
if(!waitLock.containsKey(xid)) {//
continue;
} else {//若选出的xid获得了锁
u2x.put(uid, xid);
Lock lo = waitLock.remove(xid);
waitU.remove(xid);
lo.unlock();
break;
}
}
if(l.size() == 0) wait.remove(uid);
}
删除版本
delete() 方法看起来略为复杂：
public boolean delete(long xid, long uid) throws Exception {
lock.lock();
Transaction t = activeTransaction.get(xid);
lock.unlock();
if(t.err != null) {
throw t.err;
}
Entry entry = null;
try {
entry = super.get(uid);
} catch(Exception e) {
if(e == Error.NullEntryException) {
return false;
} else {
throw e;
}
}
try {
if(!Visibility.isVisible(tm, t, entry)) {
return false;
}
Lock l = null;
try {
l = lt.add(xid, uid);
} catch(Exception e) {
//检测到死锁
t.err = Error.ConcurrentUpdateException;
internAbort(xid, true);
t.autoAborted = true;
throw t.err;
}
//没有死锁，但是需要等待当前事务拿到锁
if(l != null) {
l.lock();
l.unlock();
}
if(entry.getXmax() == xid) {
return false;
}
//System.out.println(entry.getXmax());
//检查到版本跳跃，回滚事务
if(Visibility.isVersionSkip(tm, t, entry)) {
System.out.println("检查到版本跳跃，自动回滚");
t.err = Error.ConcurrentUpdateException;
internAbort(xid, true);
t.autoAborted = true;
throw t.err;
}
//删除事务，把当前xid设置为Xmax
entry.setXmax(xid);
return true;
} finally {
entry.release();
}
}
实际上主要是前置的三件事：一是可见性判断，二是获取资源的锁，三是版本跳跃判断。删除的操作只有一个设置 XMAX。
