Innodb行锁（3）：事务的调度权重、死锁权重和死锁判定方式

首先我们需要明白一点，数据库中存在2种权重：

• 调度（scheduler）权重：主要和锁堵塞的时长和锁堵塞的事务数量相关，主要用于锁释放后唤醒事务顺序的判断依据。
• 死锁权重：这部分主要和修改的行数和加锁的数量有关，主要是死锁发生后回滚哪个事务。

这里我们先描述死锁检测线程死锁检测部分的第一个功能，调度（scheduler）权重计算，为了描述这个问题，我们以如下的等待方式进行描述，以下是一种死锁例子，

image.png

也就是B等待A，C等待A，D等待A，F等待A，A等待D，这实际上就是死锁了，同时为了DEBUG方便，我们将参数设置一下如下：

• innodb_lock_wait_timeout=100000
• innodb_deadlock_detect=off

并且debug的时候，在D事务等待后，我们多模拟几次其他的堵塞，目的在于推进lock_wait_table_reservations全局变量，让计算权重的时候判定为长时间没有获取锁的情况。
注意这里用的8023版本代码，并且debug中我们一般去掉了F事务，为了更加方便一点。

一、事务的调度权重计算流程

首先

会扫描整个LOCK SYSTEM的slot信息，获取其当前等待的一个snapshot，注意这部分加锁lock_sys->wait_mutex，但是和前面LOCK SYSTEM中的lock的查找和加入并不是一个把锁。调用函数为lock_wait_snapshot_waiting_threads，其将当前LOCK SYSTEM的slot信息放入一个容器中，这个容器并没有顺序其中元素包含的信息如下：

• from(trx):当前事务
• to(waits_for):堵塞当前事务的事务
• slot(slot):slot 结构
• slot->reservation_no(reservation_no):事务等待发生时数据库中发生了多少锁等待

其返回值为当前数据库中发生了多少锁等待，也就是全局变量lock_wait_table_reservations的值。我们将这里获取的信息叫做info容器

然后

调用lock_wait_build_wait_for_graph，来建立等待的数据结构，首先对info容器中的信息按照 from(trx)的事务指针地址进行排序，也就是重载的<比较符如下：

/** As we want to quickly find a given trx_t within the snapshot, we use a
sorting criterion which is based on trx only. We use the pointer address, as
any deterministic rule without ties will do. */
bool operator<(const waiting_trx_info_t &a, const waiting_trx_info_t &b) {
  return std::less<trx_t *>{}(a.trx, b.trx);
}

那么经过排序后我们上面的信息可以描述为如下：

info容器：
[A] wait [D] 存储在  info vector[0]
[B] wait [A] 存储在  info vector[1]
[C] wait [A] 存储在  info vector[2]
[D] wait [A] 存储在  info vector[3]
[F] wait [A] 存储在  info vector[4]

接着遍历这个info容器，需要做的就是每次获取到一个元素后，获取其堵塞者的事务，然后进行二分查找也就是std::lower_bound，然后查找容器中是否有响应的事务的，并且计算堵塞者元素和容器开始元素之间的距离,也就是堵塞者在info容器的下标，比如 [A] wait [D]这个元素，我们需要二分查找整个info容器，找到[D] wait [A] 这个元素，其下标为3，而开始元素下标为0，也就是3-0=3，并且将结果放入到outgoing容器中，得到的结果如下：

outgoing容器:
outgoing[0] = 3(info容器下标)    outgoing[0]   [A] wait [D]  D为堵塞者
outgoing[1] = 0(info容器下标)    outgoing[1]   [B] wait [A]  A为堵塞者
outgoing[2] = 0(info容器下标)    outgoing[2]   [C] wait [A]  A为堵塞者
outgoing[3] = 0(info容器下标)    outgoing[3]   [D] wait [A]  A为堵塞者
outgoing[4] = 0(info容器下标)    outgoing[4]   [F] wait [A]  A为堵塞者
       A
| \    |\    |\
|      |     |/   
B      C     D
$60 = std::vector of length 4, capacity 4 = {3, 0, 0, 0}

outgoing容器看起来为一个逆邻接表，这里如果A没有等待D出现死锁环形等待，那么B,C,D的堵塞源头没有计入info容器，因此outgoing容器的值都为初始值-1，如下：

A<-B<-C<-D 的outgoing容器
(gdb) p outgoing
$66 = std::vector of length 3, capacity 3 = {-1, -1, -1}

并且在下面计算incoming容器的时候，因为都为-1，则如下语句生效，

lock_wait_compute_incoming_count：
if (to != -1)

因此incoming容器值为初始值0。那么计算调度权重的时候，就是初始的调度权重，也就是按照单个事务的调度权重，其实这也是对的，B、C、D没有堵塞其他事务，就是它们自己的调度权重了，如下判定，

lock_wait_accumulate_weights:
 if (outgoing[id] != -1) {               //为-1 不做计算                             
      lock_wait_add_subtree_weight(new_weights, outgoing[id], id); //to blocker 和 from trx
      if (!--incoming_count[outgoing[id]]) {   //如果只堵塞一个事务       
        ready.push_back(outgoing[id]);               
      } //将堵塞者放入到ready容器继续循环累加权重 ,也就是计算各个等待链路上的每个事务的调度权重，为累加值    

       A
| \    |\    |\
|      |     |
B      C     D
(gdb) p new_weights
$76 = std::vector of length 3, capacity 3 = {3, 3, 3}

三、开始计算调度（scheduler）权重

初始化的情况下所有事务的调度权重都为1，先调用lock_wait_compute_initial_weights函数，查找是否有事务等待的时间比较久，如果比较久就需要设置其权重为当前等待的事务数量，而不在是1，其判定是否等待时间过久的方式如下：

• 如果等待事务的slot的reservation_no+ MAX_FAIR_WAIT还小于了建立等待快照时候的table_reservations，则代表当前事务可能等待得较久，其中MAX_FAIR_WAIT为(2 * 当前等待的事务) ，这里将值存入new_weights容器中，因为不知道具体多少值我们用X表示。

new_weights容器：
new_weights[0] = X   事务A的初始化调度（scheduler）权重
new_weights[1] = X   事务B的初始化调度（scheduler）权重
new_weights[2] = X   事务C的初始化调度（scheduler）权重
new_weights[3] = X   事务D的初始化调度（scheduler）权重
new_weights[4] = X   事务F的初始化调度（scheduler）权重
       A
| \    |\    |\
|      |     |/   
B      C     D
(gdb) p new_weights
$63 = std::vector of length 4, capacity 4 = {1, 4, 4, 4} (初始化单个事务的调度权重)

也就是说如果A事务等待的事务有1000个事务堵塞过，然后又过了一会，当前数据库中10个事务处于堵塞状态，并且当前有1100个事务堵塞过，那么1000+2*10 = 1020小于1100，则判定其调度权重为10，也就是当前等待的事务数量。
然后调用lock_wait_compute_incoming_count进行outing入度计算，也就是计算每个事务堵塞了多少其他事务，方式就是循环outgoing容器，也比较简单每次扫描到值相同的说明这个事务堵塞了一个其他事务，做++操作就可以了，得到的一个incoming容器如下，

incoming容器：
incoming[3] = 1  [3]为容器info的下标，D堵塞了事务A
incoming[0] = 4  [0]为容器info的下标，A堵塞了4个事务，分别为B,C,D,F
incoming[1] = 0  [1]为容器info的下标，B没有堵塞事务，初始值
incoming[2] = 0  [2]为容器info的下标，C没有堵塞事务，初始值
incoming[4] = 0  [4]为容器info的下标，F没有堵塞事务 ，初始值    
       A
| \    |\    |\
|      |     |/   
B      C     D
(gdb) p incoming_count
$61 = std::vector of length 4, capacity 4 = {3, 0, 0, 1}

下面就是调用lock_wait_accumulate_weights，根据incoming容器、outgoing容器、infos容器计算最终的调度权重，这里incoming容器是可以0值的，因为可能其没有堵塞其他事务，比如B和C事务，那么最后计算从B和C事务开始反推A事务的权重，先找到没有堵塞其他事务的事务，放入ready容器中如下：

  for (size_t id = 0; id < n; ++id) {
    if (!incoming_count[id]) {
      ready.push_back(id); //先找到没有堵塞其他事务的事务，因为在incoming没有堵塞其他事务的其值为0
    }
  } 
       A
| \    |\    |\
|      |     |/   
B      C     D
(gdb) p ready
$62 = std::vector of length 2, capacity 2 = {1, 2}

然后调用函数lock_wait_add_subtree_weight，通过outgoing容器不断反推，因为outgoing容器是一个逆邻接表，很容易找到其堵塞者，然后将堵塞者设置为parent_id，而被堵塞者设置为child_id，然后在new_weights容器中找到各自的调度权重信息，堵塞者的调度权重加上被堵塞者的调度权重即可。

old_parent_weight += child_weight;

如此不断的迭代即可。那么这里A事务的调度权重还需要加上F事务的调度权重，如下：

• A的调度权重为A+B+C+F
• B的调度权重为B
• C的调度权重为C
• D的调度权重为D
• F的调度权重为F

       A
| \    |\    |\
|      |     |/   
B      C     D
(gdb) p new_weights
$64 = std::vector of length 4, capacity 4 = {9, 4, 4, 4}（事务A计算了累积的调度权重）

注意这里相互堵塞的事务调度权重并没有累加，也就是A没有包含D事务的调度权重，从debug结果也是如此，从算法看D事务堵塞了A事务，因此不是没有堵塞其他事务的事务，因此不计入ready容器。
最后调用lock_wait_publish_new_weights函数将调度权重写入到每个事务中，不做详细描述了。

四、调度权重在锁唤醒中的作用

image.png

从上面的解析中，我们可以看到一个反推累加权重的过程，这里有一张图，按照前文的描述如果，TRX A事务提交了，到底先唤醒TRX B还是TRX F就是根据这个调度权重来的，我们做图中的事务等待如下：

create table lock1(id int);
create table lock2(id int);
create table lock3(id int);
create table lock4(id int);

insert into lock1 values(1);
insert into lock2 values(1);
insert into lock3 values(1);
insert into lock4 values(1);

TRX A:
begin;
delete from lock1;

TRX B:
begin;
delete from lock2;
delete from lock1;(wait)

TRX C:
begin;
delete from lock3;
delete from lock2;(wait)

TRX D:
begin;
delete from lock3;(wait)

TRX E:
begin;
delete from lock2;(wait)

TRX F:
begin;
delete from lock4;
delete from lock1;(wait)

TRX G:
begin;
delete from lock4;(wait)

我们对最终计算调度权重的new_weights容器进行debug，如下：

(gdb) p new_weights
$2 = std::vector of length 6, capacity 6 = {4, 2, 1, 2, 1, 1}

显然和我们图中一致，TRX B的调度权重就是4，而有2个调度权重为2的就是TRX F和TRX C。那么当TRX A提交后， TRX B有优先唤醒的权利，TRX B唤醒后TRX F依旧处于等待状态，因为TRX F依旧拿不到LOCK_1，现在持有者变为了TRX B。

五、死锁判定和死锁权重

当我们参数innodb_deadlock_detect设置为ON的时候，就会进行死锁检测，死锁检测主要使用的上面的两个容器infos容器和outgoing容器，其中outgoing容器我们从前面的分析来看，通过它很容易找到死锁的两个事务，如下：

outgoing容器:
outgoing[0] = 3(info容器下标)    outgoing[0]   [A] wait [D]  D为堵塞者
outgoing[1] = 0(info容器下标)    outgoing[1]   [B] wait [A]  A为堵塞者
outgoing[2] = 0(info容器下标)    outgoing[2]   [C] wait [A]  A为堵塞者
outgoing[3] = 0(info容器下标)    outgoing[3]   [D] wait [A]  A为堵塞者
outgoing[4] = 0(info容器下标)    outgoing[4]   [F] wait [A]  A为堵塞者
       A
| \    |\    |\
|      |     |/   
B      C     D
$60 = std::vector of length 4, capacity 4 = {3, 0, 0, 0}

因为通过outgoing[0] = 3事务A就找到了事务D，而通过outgoing[3] = 0 事务D又找到了事务A，这实际上就是出现了环形等待死锁。而MySQL也是这么做的，先循环整个outgoing容器，每次循环的时候会再次通过outgoing容器值进行循环，查找是否有环形等待，通过一个colors容器进行标记，然后将冲突的2个事务进行死锁权重判定，判定的方式如下：

/** Compares the "weight" (or size) of two transactions. Transactions that
 have edited non-transactional tables are considered heavier than ones
 that have not.
 @return true if weight(a) >= weight(b) */
return (TRX_WEIGHT(a) >= TRX_WEIGHT(b));

TRX_WEIGHT宏：
/** Calculates the "weight" of a transaction. The weight of one transaction
 is estimated as the number of altered rows + the number of locked rows.
 @param t transaction
 @return transaction weight */

也就是死锁权重包含2个部分：

• number of altered rows
• number of locked rows

这比调度权重可简单多了。而最终会调用Deadlock_notifier::notify打印我们常见的死锁日志，然后调用lock_wait_rollback_deadlock_victim唤醒死锁的牺牲事务，当然每次死锁释放后也会更新调度权重。

六、相关代码

这部分代码可能有点乱，死锁没有写详细的流程，仅供参考。

  ->每1秒进行循环       
    ->lock_wait_update_schedule_and_check_for_deadlocks();  
      ->lock_wait_snapshot_waiting_threads
        完成第一步构建infos素组，其中存储的就是相关slot信息和当前session和等待session的信息
        ->lock_wait_mutex_enter();
          加lock_system_t wait mutex
        ->table_reservations = lock_wait_table_reservations;
          获取当前的lock_wait_table_reservations
        ->for (auto slot = lock_sys->waiting_threads; slot < lock_sys->last_slot;++slot) 
          循环整个slot数组
          if (slot->in_use)
          -> 如果slot正在使用    
            ->thr_get_trx(slot->thr)
              获取当前的事务(waitting)
            ->from->lock.blocking_trx.load()
              获取堵塞当前事务的事务(blocker)
              这里的lock为trx_lock_t,因为这里在等待时候已经记录
              了blocker的信息，因此可以直接获取
            -> infos.push_back({from, to, slot, slot->reservation_no});
               将信息入到到infos中，这是一个容器
        ->lock_wait_mutex_exit()    
          进行解锁
        ->return table_reservations       
          返回当前的这个信息
    ->lock_wait_build_wait_for_graph
      应该是构建等待图
      ->outgoing.resize(n, -1)
        初始化为-1
      ->sort(infos.begin(), infos.end())
        进行排序，按照事务指针地址的大小
      ->for (uint from = 0; from < n; ++from)
        循环整个infos信息，这个信息是前面通过扫描整个waiting_threads出来的
        ->needle.trx = infos[from].waits_for;
        获取blocker 事务的的信息
        ->auto it = std::lower_bound(infos.begin(), infos.end(), needle)
          获取第一个比事务地址大或者相等的事务的迭代器
          ->if (it == infos.end() || it->trx != needle.trx) continue; 
            如果没有找到堵塞的事务，则继续
          ->auto to = it - infos.begin(); 
            返回一个堵塞者的容器下标
          ->outgoing[from] = static_cast<int>(to)
            记录这个下标到容器outgoing       
    ->lock_wait_compute_and_publish_weights_except_cycles
      ->lock_wait_compute_initial_weights
        ->const size_t n = infos.size();
          设置n为当前处于等待事务的数量
        ->WEIGHT_BOOST = n == 0 ? 1 : std::min<trx_schedule_weight_t>(n, 1e9 / n)
          如果等待的事务为0，则设置WEIGHT_BOOST为1
          否则取小值n, 1e9 / n，一般就是n了，大表等待事务的数量
          比如debug p n $2 = 2 这个时候有两个事务堵塞
        ->new_weights.clear();
        ->new_weights.resize(n, 1)
          初始化所有权重为1
        ->for (size_t from = 0; from < n; ++from)
          循环整个等待快照info容器的vector数组
          ->if (infos[from].reservation_no + MAX_FAIR_WAIT < table_reservations)
          如果等待事务的slot的reservation_no+ MAX_FAIR_WAIT(2 * n) 还小于了建立
          等待快照时候的table_reservations，则代表当前事务可能等待得较久
          ->new_weights[from] = WEIGHT_BOOST
          则设置其权重为WEIGHT_BOOST
      ->lock_wait_compute_incoming_count
        ->const size_t n = outgoing.size();
          获取outgoing容器的长度
        ->incoming_count.resize(n, 0)
          incoming容器大小为outgoing容器的大小，初始值为0
        ->for (size_t id = 0; id < n; ++id)
          循环outgoing队列，计算入度，to为就是其堵塞源头的下标（如果不为-1的话）
          ->if (to != -1)
            ->incoming_count[to]++;   
              对于找到一次就 ++           
      ->lock_wait_accumulate_weights
        ->const size_t n = incoming_count.size();
          初始化n为incoming_count容器大小
        ->for (size_t id = 0; id < n; ++id) {
         ->if (!incoming_count[id]) {//如果incoming_count容器的值为0就代表没有堵塞
           ->ready.push_back(id); /先找到没有堵塞的id，因为在incoming为0
        ->while (!ready.empty())
          查询ready容器
          ->size_t id = ready.back();ready.pop_back();
          从尾部获取并且弹出元素
          ->lock_wait_add_subtree_weight(new_weights, outgoing[id], id)
          获取to blocker 和 from trx
            ->old_parent_weight += child_weight
              累加权重
          ->if (!--incoming_count[outgoing[id]]) 
            如果只堵塞一个事务
            ->ready.push_back(outgoing[id]);
              将堵塞者放入到ready容器继续循环累加权重，也就是计算各个等待链路上的每个事务的权重，为累加值  
      ->lock_wait_publish_new_weights
        发布权重个事务，用于唤醒权利的判定。

如果开启死锁检测则，
lock_wait_find_and_handle_deadlocks
 ->lock_wait_extract_cycle_ids
 ->lock_wait_check_candidate_cycle
   ->lock_wait_choose_victim
     ->lock_wait_order_for_choosing_victim
     ->trx_weight_ge 
       死锁权重比较
 ->lock_wait_handle_deadlock
   ->lock_wait_update_weights_on_cycle
   ->lock_notify_about_deadlock
     ->lock_wait_trxs_rotated_for_notification
     ->Deadlock_notifier::notify
   ->lock_wait_rollback_deadlock_victim
     ->lock_cancel_waiting_and_release