开源分布式数据库PolarDB-X源码解读——PolarDB-X源码解读（六）：分布式死锁检测

一、死锁检测任务的生命周期

本文将主要解读PolarDB-X中分布式死锁检测功能的源码。阅读前建议先了解我们分布式死锁检测原理的相关文章：

 PolarDB-X分布式MDL死锁检测。

死锁检测功能属于事务模块的功能，死锁检测任务则挂载在事务管理器TransactionManager中。在TransactionManager初始化时（参考PolarDB-X CN启动流程，代码入口在MatrixConfigHolder#doInit中调用了对应transactionManager#init方法），TransactionManager会初始化一个死锁检测的定时任务，该任务每个计算节点（CN）都有且只有一个，每隔一定时间进行一次死锁检测（默认是1秒）。

接下来看一下这个死锁检测的任务里面具体做了什么事情。

二、代码主体逻辑：DeadlockDetectionTask

死锁检测任务的代码主要在DeadlockDetectionTask里，任务每次被调度时，执行的代码入口是其中的run方法。

该方法首先会判断一下当前CN是否为leader，只有leader节点才会执行死锁检测任务。

if (!hasLeadership()) {
    return;
}

一次死锁检测主要涉及三个步骤。

 第一步，先获取所有分布式事务的信息。

// Get all global transaction information
final TrxLookupSet lookupSet = fetchTransInfo();

这一步需要sync到所有CN节点，以获取所有分布式事务的信息，fetchTransInfo方法里主要是调用了FetchTransForDeadlockDetectionSyncAction#sync这个sync的方法。为了避免返回过多的事务信息，这里面只会返回从开始到现在大于1秒的分布式事务。

final long beforeTimeMillis = System.currentTimeMillis() - 1000L;
final long beforeTxid = IdGenerator.assembleId(beforeTimeMillis, 0, 0);
for (ITransaction tran : transactions) {
    if (!tran.isDistributed()) {
        continue;
    }
    // Do deadlock detection only for transactions that take longer than 1s.
    if (tran.getId() >= beforeTxid) {
        continue;
    }
    // Get information from this tran.
    ......
}

返回的TrxLookupSet记录了所有事务的信息，主要包括事务id，事务所在前端连接的id，事务涉及的分片，分片上分支事务（MySQL上的事务）的连接id等。以上信息均会用在后续的死锁检测中。  

 第二步，获取每个数据节点（DN）的锁等待信息，并结合上一步的分布式事务信息，更新全局的事务等待关系图。

首先会获取所有DN的数据源。下面Map返回的数据中，key是DN的id，value是一个列表，存放了所有schema对应的数据源。我们可以使用任意一个数据源来访问DN，同时我们也会使用到这些数据源中存放的物理分片名和分支事务的连接id来确定对应的分布式事务。注意，只使用分支事务的连接id无法确定对应的分布式事务，因为不同的DN上能同时存在相同的连接id。

// Get all group data sources, and group by DN's ID (host:port)
final Map<String, List<TGroupDataSource>> instId2GroupList = 
ExecUtils.getInstId2GroupList(allSchemas);

然后我们会对每个DN，获取物理分片上的锁和分支事务信息，结合上一步获取的TrxLookupSet中的分布式事务信息，更新一个全局的事务等待关系图（代码中的DiGraph）。DiGraph是一个全局事务等待关系图，图中每个点是一个事务，每条有向边表示事务的等待关系。

final DiGraph<TrxLookupSet.Transaction> graph = new DiGraph<>();
for (List<TGroupDataSource> groupDataSources : instId2GroupList.values()) {
    if (CollectionUtils.isNotEmpty(groupDataSources)) {
        // Since all data sources are in the same DN, any data source is ok.
        final TGroupDataSource groupDataSource = groupDataSources.get(0);
        // Get all group names in this DN.
        final Set<String> groupNames =
            groupDataSources.stream().map(TGroupDataSource::getDbGroupKey).collect(Collectors.toSet());
        // Fetch lock-wait information for this DN,
        // and update the lookup set and the graph with the information.
        fetchLockWaits(groupDataSource, groupNames, lookupSet, graph);
    }
}

fetchLockWaits方法主要就是查询DN上information_schema.innodb_locks/innodb_trx/innodb_lock_waits这三个视图，来获取具体的行锁信息。获取的内容主要包括锁等待的分支事务的连接id，以及一些最后输出到死锁日志里的信息。然后，根据这个分支事务的连接id，查到对应的分布式事务，将这个分支事务的等待关系转化为分布式事务的等待关系，加到等待图中。

 最后一步，检测图中是否存在环，若存在，则回滚环中的某个事务。

graph.detect().ifPresent((cycle) -> {
    // 若检测到环，先保留一份死锁日志
    DeadlockParser.parseGlobalDeadlock(cycle);
    // 然后选择一个事务回滚掉，目前默认回滚环的第一个事务
    killByFrontendConnId(cycle.get(0));
});

其中，graph.detect()为检测环的算法，具体实现在DiGraph部分的代码中。简单来说，就是在有向图上进行深度优先搜索，检测是否存在环路。此外，保留的死锁日志可以使用SHOW GLOBAL DEADLOCKS查看。

最后，为了解决死锁，会选择回滚掉环中的一个事务。目前默认是回滚环里的第一个事务，相当于随机回滚一个事务。

关于MDL死锁检测的代码与此类似，主体部分在MdlDeadlockDetectionTask。

三、其他细节  

 第一步的sync机制是如何实现的？简单来说，sync的语义就是要让所有CN节点执行同一个action，并得到action的结果。发起sync的节点会将该action对象序列化后发送到所有CN，CN反序列化后执行这个对象的sync方法，并发得到的结果返回。以上行为的逻辑主要实现在ClusterSyncManager#doSync方法里。阅读该部分代码，我们也能知道，给其他CN发送sync请求其实就是执行了一个SYNC schema_name serializedAction的SQL语句，返回的结果也是和普通查询返回的结果类似。

 第三步中，具体是如何回滚事务的？发生死锁时，待回滚的事务当前执行的语句正卡在某个锁等待上。因此，死锁检测任务会发起一个kill query的sync请求，先把这条语句kill掉，并把kill query的错误码设置为ERR_TRANS_DEADLOCK。kill query的入口在KillSyncAction，最后实际运行kill query逻辑的代码为ServerConnection.CancelQueryTask中的doCancel方法。

private void doCancel() throws SQLException {
    // 这个 futureCancelErrorCode 用在后面的错误判断中，
    // 死锁导致的 kill，错误码都是 ERR_TRANS_DEADLOCK
    futureCancelErrorCode = this.errorCode;
    // kill 掉所有物理连接上正在运行的 SQL
    if (conn != null) {
        conn.kill();
    }
    // 这里这个 f 是正在执行逻辑 SQL 的任务
    Future f = executingFuture;
    if (f != null) {
        f.cancel(true);
    }
}

该方法实际上会对每个物理连接（CN和DN的连接）调用kill query的逻辑，把该逻辑语句对应的所有物理语句都kill掉，其中正在等锁的物理语句就会被kill掉，最后会中断正在执行这个逻辑语句的线程。

被中断的线程会在ServerConnection中的handleError方法里处理异常，发现错误码是ERR_TRANS_DEADLOCK时，就会将当前事务回滚掉，并给客户端发送Deadlock found when trying to get lock; try restarting transaction的错误提示。

// Handle deadlock error.
if (isDeadLockException(t)) {
    // Prevent this transaction from committing.
    this.conn.getTrx().setCrucialError(ERR_TRANS_DEADLOCK);
    // Rollback this trx.
    try {
        innerRollback();
    } catch (SQLException exception) {
        logger.warn("rollback failed when deadlock found", exception);
    }
}

四、小结

本文简单介绍了PolarDB-X分布式死锁检测功能的源码，感兴趣的同学可以结合源码解读，在DeadlockDetectionTask#run方法打一个断点，观察每一步执行的结果，可以更容易理解这一功能的实现。