首先我們需要明白一點(diǎn)，數(shù)據(jù)庫中存在2種權(quán)重：

• 調(diào)度（scheduler）權(quán)重：主要和鎖堵塞的時長和鎖堵塞的事務(wù)數(shù)量相關(guān)，主要用于鎖釋放后喚醒事務(wù)順序的判斷依據(jù)。
• 死鎖權(quán)重：這部分主要和修改的行數(shù)和加鎖的數(shù)量有關(guān)，主要是死鎖發(fā)生后回滾哪個事務(wù)。

這里我們先描述死鎖檢測線程死鎖檢測部分的第一個功能，調(diào)度（scheduler）權(quán)重計算，為了描述這個問題，我們以如下的等待方式進(jìn)行描述，以下是一種死鎖例子，

image.png

也就是B等待A，C等待A，D等待A，F(xiàn)等待A，A等待D，這實際上就是死鎖了，同時為了DEBUG方便，我們將參數(shù)設(shè)置一下如下：

• innodb_lock_wait_timeout=100000
• innodb_deadlock_detect=off

并且debug的時候，在D事務(wù)等待后，我們多模擬幾次其他的堵塞，目的在于推進(jìn)lock_wait_table_reservations全局變量，讓計算權(quán)重的時候判定為長時間沒有獲取鎖的情況。
注意這里用的8023版本代碼，并且debug中我們一般去掉了F事務(wù)，為了更加方便一點(diǎn)。

一、事務(wù)的調(diào)度權(quán)重計算流程

首先

會掃描整個LOCK SYSTEM的slot信息，獲取其當(dāng)前等待的一個snapshot，注意這部分加鎖lock_sys->wait_mutex，但是和前面LOCK SYSTEM中的lock的查找和加入并不是一個把鎖。調(diào)用函數(shù)為lock_wait_snapshot_waiting_threads，其將當(dāng)前LOCK SYSTEM的slot信息放入一個容器中，這個容器并沒有順序其中元素包含的信息如下：

• from(trx):當(dāng)前事務(wù)
• to(waits_for):堵塞當(dāng)前事務(wù)的事務(wù)
• slot(slot):slot 結(jié)構(gòu)
• slot->reservation_no(reservation_no):事務(wù)等待發(fā)生時數(shù)據(jù)庫中發(fā)生了多少鎖等待

其返回值為當(dāng)前數(shù)據(jù)庫中發(fā)生了多少鎖等待，也就是全局變量lock_wait_table_reservations的值。我們將這里獲取的信息叫做info容器

然后

調(diào)用lock_wait_build_wait_for_graph，來建立等待的數(shù)據(jù)結(jié)構(gòu)，首先對info容器中的信息按照 from(trx)的事務(wù)指針地址進(jìn)行排序，也就是重載的<比較符如下：

/** 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);
}

那么經(jīng)過排序后我們上面的信息可以描述為如下：

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容器，需要做的就是每次獲取到一個元素后，獲取其堵塞者的事務(wù)，然后進(jìn)行二分查找也就是std::lower_bound，然后查找容器中是否有響應(yīng)的事務(wù)的，并且計算堵塞者元素和容器開始元素之間的距離,也就是堵塞者在info容器的下標(biāo)，比如 [A] wait [D]這個元素，我們需要二分查找整個info容器，找到[D] wait [A] 這個元素，其下標(biāo)為3，而開始元素下標(biāo)為0，也就是3-0=3，并且將結(jié)果放入到outgoing容器中，得到的結(jié)果如下：

outgoing容器:
outgoing[0] = 3(info容器下標(biāo))    outgoing[0]   [A] wait [D]  D為堵塞者
outgoing[1] = 0(info容器下標(biāo))    outgoing[1]   [B] wait [A]  A為堵塞者
outgoing[2] = 0(info容器下標(biāo))    outgoing[2]   [C] wait [A]  A為堵塞者
outgoing[3] = 0(info容器下標(biāo))    outgoing[3]   [D] wait [A]  A為堵塞者
outgoing[4] = 0(info容器下標(biāo))    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出現(xiàn)死鎖環(huán)形等待，那么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。那么計算調(diào)度權(quán)重的時候，就是初始的調(diào)度權(quán)重，也就是按照單個事務(wù)的調(diào)度權(quán)重，其實這也是對的，B、C、D沒有堵塞其他事務(wù)，就是它們自己的調(diào)度權(quán)重了，如下判定，

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]]) {   //如果只堵塞一個事務(wù)       
        ready.push_back(outgoing[id]);               
      } //將堵塞者放入到ready容器繼續(xù)循環(huán)累加權(quán)重 ,也就是計算各個等待鏈路上的每個事務(wù)的調(diào)度權(quán)重，為累加值    

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

三、開始計算調(diào)度（scheduler）權(quán)重

初始化的情況下所有事務(wù)的調(diào)度權(quán)重都為1，先調(diào)用lock_wait_compute_initial_weights函數(shù)，查找是否有事務(wù)等待的時間比較久，如果比較久就需要設(shè)置其權(quán)重為當(dāng)前等待的事務(wù)數(shù)量，而不在是1，其判定是否等待時間過久的方式如下：

• 如果等待事務(wù)的slot的reservation_no+ MAX_FAIR_WAIT還小于了建立等待快照時候的table_reservations，則代表當(dāng)前事務(wù)可能等待得較久，其中MAX_FAIR_WAIT為(2 * 當(dāng)前等待的事務(wù)) ，這里將值存入new_weights容器中，因為不知道具體多少值我們用X表示。

new_weights容器：
new_weights[0] = X   事務(wù)A的初始化調(diào)度（scheduler）權(quán)重
new_weights[1] = X   事務(wù)B的初始化調(diào)度（scheduler）權(quán)重
new_weights[2] = X   事務(wù)C的初始化調(diào)度（scheduler）權(quán)重
new_weights[3] = X   事務(wù)D的初始化調(diào)度（scheduler）權(quán)重
new_weights[4] = X   事務(wù)F的初始化調(diào)度（scheduler）權(quán)重
       A
| \    |\    |\
|      |     |/   
B      C     D
(gdb) p new_weights
$63 = std::vector of length 4, capacity 4 = {1, 4, 4, 4} (初始化單個事務(wù)的調(diào)度權(quán)重)

也就是說如果A事務(wù)等待的事務(wù)有1000個事務(wù)堵塞過，然后又過了一會，當(dāng)前數(shù)據(jù)庫中10個事務(wù)處于堵塞狀態(tài)，并且當(dāng)前有1100個事務(wù)堵塞過，那么1000+2*10 = 1020小于1100，則判定其調(diào)度權(quán)重為10，也就是當(dāng)前等待的事務(wù)數(shù)量。
然后調(diào)用lock_wait_compute_incoming_count進(jìn)行outing入度計算，也就是計算每個事務(wù)堵塞了多少其他事務(wù)，方式就是循環(huán)outgoing容器，也比較簡單每次掃描到值相同的說明這個事務(wù)堵塞了一個其他事務(wù)，做++操作就可以了，得到的一個incoming容器如下，

incoming容器：
incoming[3] = 1  [3]為容器info的下標(biāo)，D堵塞了事務(wù)A
incoming[0] = 4  [0]為容器info的下標(biāo)，A堵塞了4個事務(wù)，分別為B,C,D,F
incoming[1] = 0  [1]為容器info的下標(biāo)，B沒有堵塞事務(wù)，初始值
incoming[2] = 0  [2]為容器info的下標(biāo)，C沒有堵塞事務(wù)，初始值
incoming[4] = 0  [4]為容器info的下標(biāo)，F(xiàn)沒有堵塞事務(wù) ，初始值    
       A
| \    |\    |\
|      |     |/   
B      C     D
(gdb) p incoming_count
$61 = std::vector of length 4, capacity 4 = {3, 0, 0, 1}

下面就是調(diào)用lock_wait_accumulate_weights，根據(jù)incoming容器、outgoing容器、infos容器計算最終的調(diào)度權(quán)重，這里incoming容器是可以0值的，因為可能其沒有堵塞其他事務(wù)，比如B和C事務(wù)，那么最后計算從B和C事務(wù)開始反推A事務(wù)的權(quán)重，先找到?jīng)]有堵塞其他事務(wù)的事務(wù)，放入ready容器中如下：

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

然后調(diào)用函數(shù)lock_wait_add_subtree_weight，通過outgoing容器不斷反推，因為outgoing容器是一個逆鄰接表，很容易找到其堵塞者，然后將堵塞者設(shè)置為parent_id，而被堵塞者設(shè)置為child_id，然后在new_weights容器中找到各自的調(diào)度權(quán)重信息，堵塞者的調(diào)度權(quán)重加上被堵塞者的調(diào)度權(quán)重即可。

old_parent_weight += child_weight;

如此不斷的迭代即可。那么這里A事務(wù)的調(diào)度權(quán)重還需要加上F事務(wù)的調(diào)度權(quán)重，如下：

• A的調(diào)度權(quán)重為A+B+C+F
• B的調(diào)度權(quán)重為B
• C的調(diào)度權(quán)重為C
• D的調(diào)度權(quán)重為D
• F的調(diào)度權(quán)重為F

       A
| \    |\    |\
|      |     |/   
B      C     D
(gdb) p new_weights
$64 = std::vector of length 4, capacity 4 = {9, 4, 4, 4}（事務(wù)A計算了累積的調(diào)度權(quán)重）

注意這里相互堵塞的事務(wù)調(diào)度權(quán)重并沒有累加，也就是A沒有包含D事務(wù)的調(diào)度權(quán)重，從debug結(jié)果也是如此，從算法看D事務(wù)堵塞了A事務(wù)，因此不是沒有堵塞其他事務(wù)的事務(wù)，因此不計入ready容器。
最后調(diào)用lock_wait_publish_new_weights函數(shù)將調(diào)度權(quán)重寫入到每個事務(wù)中，不做詳細(xì)描述了。

四、調(diào)度權(quán)重在鎖喚醒中的作用

image.png

從上面的解析中，我們可以看到一個反推累加權(quán)重的過程，這里有一張圖，按照前文的描述如果，TRX A事務(wù)提交了，到底先喚醒TRX B還是TRX F就是根據(jù)這個調(diào)度權(quán)重來的，我們做圖中的事務(wù)等待如下：

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)

我們對最終計算調(diào)度權(quán)重的new_weights容器進(jìn)行debug，如下：

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

顯然和我們圖中一致，TRX B的調(diào)度權(quán)重就是4，而有2個調(diào)度權(quán)重為2的就是TRX F和TRX C。那么當(dāng)TRX A提交后， TRX B有優(yōu)先喚醒的權(quán)利，TRX B喚醒后TRX F依舊處于等待狀態(tài)，因為TRX F依舊拿不到LOCK_1，現(xiàn)在持有者變?yōu)榱薚RX B。

五、死鎖判定和死鎖權(quán)重

當(dāng)我們參數(shù)innodb_deadlock_detect設(shè)置為ON的時候，就會進(jìn)行死鎖檢測，死鎖檢測主要使用的上面的兩個容器infos容器和outgoing容器，其中outgoing容器我們從前面的分析來看，通過它很容易找到死鎖的兩個事務(wù)，如下：

outgoing容器:
outgoing[0] = 3(info容器下標(biāo))    outgoing[0]   [A] wait [D]  D為堵塞者
outgoing[1] = 0(info容器下標(biāo))    outgoing[1]   [B] wait [A]  A為堵塞者
outgoing[2] = 0(info容器下標(biāo))    outgoing[2]   [C] wait [A]  A為堵塞者
outgoing[3] = 0(info容器下標(biāo))    outgoing[3]   [D] wait [A]  A為堵塞者
outgoing[4] = 0(info容器下標(biāo))    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事務(wù)A就找到了事務(wù)D，而通過outgoing[3] = 0 事務(wù)D又找到了事務(wù)A，這實際上就是出現(xiàn)了環(huán)形等待死鎖。而MySQL也是這么做的，先循環(huán)整個outgoing容器，每次循環(huán)的時候會再次通過outgoing容器值進(jìn)行循環(huán)，查找是否有環(huán)形等待，通過一個colors容器進(jìn)行標(biāo)記，然后將沖突的2個事務(wù)進(jìn)行死鎖權(quán)重判定，判定的方式如下：

/** 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 */

也就是死鎖權(quán)重包含2個部分：

• number of altered rows
• number of locked rows

這比調(diào)度權(quán)重可簡單多了。而最終會調(diào)用Deadlock_notifier::notify打印我們常見的死鎖日志，然后調(diào)用lock_wait_rollback_deadlock_victim喚醒死鎖的犧牲事務(wù)，當(dāng)然每次死鎖釋放后也會更新調(diào)度權(quán)重。

六、相關(guān)代碼

這部分代碼可能有點(diǎn)亂，死鎖沒有寫詳細(xì)的流程，僅供參考。

  ->每1秒進(jìn)行循環(huán)       
    ->lock_wait_update_schedule_and_check_for_deadlocks();  
      ->lock_wait_snapshot_waiting_threads
        完成第一步構(gòu)建infos素組，其中存儲的就是相關(guān)slot信息和當(dāng)前session和等待session的信息
        ->lock_wait_mutex_enter();
          加lock_system_t wait mutex
        ->table_reservations = lock_wait_table_reservations;
          獲取當(dāng)前的lock_wait_table_reservations
        ->for (auto slot = lock_sys->waiting_threads; slot < lock_sys->last_slot;++slot) 
          循環(huán)整個slot數(shù)組
          if (slot->in_use)
          -> 如果slot正在使用    
            ->thr_get_trx(slot->thr)
              獲取當(dāng)前的事務(wù)(waitting)
            ->from->lock.blocking_trx.load()
              獲取堵塞當(dāng)前事務(wù)的事務(wù)(blocker)
              這里的lock為trx_lock_t,因為這里在等待時候已經(jīng)記錄
              了blocker的信息，因此可以直接獲取
            -> infos.push_back({from, to, slot, slot->reservation_no});
               將信息入到到infos中，這是一個容器
        ->lock_wait_mutex_exit()    
          進(jìn)行解鎖
        ->return table_reservations       
          返回當(dāng)前的這個信息
    ->lock_wait_build_wait_for_graph
      應(yīng)該是構(gòu)建等待圖
      ->outgoing.resize(n, -1)
        初始化為-1
      ->sort(infos.begin(), infos.end())
        進(jìn)行排序，按照事務(wù)指針地址的大小
      ->for (uint from = 0; from < n; ++from)
        循環(huán)整個infos信息，這個信息是前面通過掃描整個waiting_threads出來的
        ->needle.trx = infos[from].waits_for;
        獲取blocker 事務(wù)的的信息
        ->auto it = std::lower_bound(infos.begin(), infos.end(), needle)
          獲取第一個比事務(wù)地址大或者相等的事務(wù)的迭代器
          ->if (it == infos.end() || it->trx != needle.trx) continue; 
            如果沒有找到堵塞的事務(wù)，則繼續(xù)
          ->auto to = it - infos.begin(); 
            返回一個堵塞者的容器下標(biāo)
          ->outgoing[from] = static_cast<int>(to)
            記錄這個下標(biāo)到容器outgoing       
    ->lock_wait_compute_and_publish_weights_except_cycles
      ->lock_wait_compute_initial_weights
        ->const size_t n = infos.size();
          設(shè)置n為當(dāng)前處于等待事務(wù)的數(shù)量
        ->WEIGHT_BOOST = n == 0 ? 1 : std::min<trx_schedule_weight_t>(n, 1e9 / n)
          如果等待的事務(wù)為0，則設(shè)置WEIGHT_BOOST為1
          否則取小值n, 1e9 / n，一般就是n了，大表等待事務(wù)的數(shù)量
          比如debug p n $2 = 2 這個時候有兩個事務(wù)堵塞
        ->new_weights.clear();
        ->new_weights.resize(n, 1)
          初始化所有權(quán)重為1
        ->for (size_t from = 0; from < n; ++from)
          循環(huán)整個等待快照info容器的vector數(shù)組
          ->if (infos[from].reservation_no + MAX_FAIR_WAIT < table_reservations)
          如果等待事務(wù)的slot的reservation_no+ MAX_FAIR_WAIT(2 * n) 還小于了建立
          等待快照時候的table_reservations，則代表當(dāng)前事務(wù)可能等待得較久
          ->new_weights[from] = WEIGHT_BOOST
          則設(shè)置其權(quán)重為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)
          循環(huán)outgoing隊列，計算入度，to為就是其堵塞源頭的下標(biāo)（如果不為-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); /先找到?jīng)]有堵塞的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
              累加權(quán)重
          ->if (!--incoming_count[outgoing[id]]) 
            如果只堵塞一個事務(wù)
            ->ready.push_back(outgoing[id]);
              將堵塞者放入到ready容器繼續(xù)循環(huán)累加權(quán)重，也就是計算各個等待鏈路上的每個事務(wù)的權(quán)重，為累加值  
      ->lock_wait_publish_new_weights
        發(fā)布權(quán)重個事務(wù)，用于喚醒權(quán)利的判定。

如果開啟死鎖檢測則，
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 
       死鎖權(quán)重比較
 ->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