||
加入到LOCK SYSTEM hash结构使用的space_id和page no，比如函数lock_rec_insert_to_granted，那么至少同一个page no的lock_t会挂入同一个hash 链表(cell)中. 而在迭代的时候，我们通常是需要迭代是相同page并且某个heap no上锁了，这里以Lock_iter::for_each迭代器为例，因为这个迭代器用得很多，那么需要如下，
UNIV_INLINEulint lock_rec_fold(const page_id_t page_id) { return (ut_fold_ulint_pair(page_id.space(), page_id.page_no()));}也就是根据上面的值，然后根据hash函数进行计算后找到LOCK SYSTEM hash结构的某个链表(cell)。但是，可能某些lock_t并没有对响应的bit位上锁，那么迭代的时候就需要判定bit位，比如在迭代的时候Lock_iter::for_each迭代中有函数，
<code class="language-js_darkmode__2">bool RecID::matches(const lock_t *lock) const { return (lock->rec_lock.page_id == get_page_id() && lock_rec_get_nth_bit(lock, m_heap_no));}也就是先要确认这个lock_t 结构在当前需要上锁的行上已经上了锁了，然后才是判定兼容性，因为我们知道一个lock_t结构可能有很多的行锁，不一定就锁定了本次加锁需要的行。
实际上这里的等待唤醒，唤醒权利可以有两个方面：
实际上两个方面只是时机不同，我们这里主要讨论lock monitor线程的唤醒，因为提交或者回滚事务唤醒堵塞事务调用是一样的。
前面第一篇文章，我们已经解释了等待lock_t资源的事务已经返回一个DB_LOCK_WAIT给上层了，并且将线程属性设置为QUE_THR_LOCK_WAIT，如下开始调入等待函数，
<code class="language-js_darkmode__5">row_mysql_handle_errors case DB_LOCK_WAIT: lock_wait_suspend_thread(thr);ha_innobase::index_read -> row_search_mvcc ->row_mysql_handle_errors ->lock_wait_suspend_thread
那么在函数lock_wait_suspend_thread就开始堵塞了，将唤醒权给与其他线程，步骤大概如下，
醒来后继续干活...
前面我们已经知道堵塞事务开始等待了，我们这里就以lock monitor线程唤醒超时的等待事务为列子，进行讲述。当然lock monitor线程还负责死锁检测在8.0中，这个我们后面在进行描述。先看它的第一个功能，唤醒等待的事务(线程)。本线程大概每1秒都会进行检测，完成的工作主要如下，
<code class="language-js_darkmode__13">for (auto slot = lock_sys->waiting_threads; slot last_slot;++slot)
<code class="language-js_darkmode__15">if (wait_time > (int64_t)slot->wait_timeout
那么完成这一步本次需要释放的lock_t锁结构就在系统中不存在了，但是需要继续做的就是如果本事务超时的lock_t结构释放后会发生什么。比如A<-B<-C<-D，一旦B事务的锁自选超时了，那么可能出现三种情况：
那么一旦B事务超时了，可能情况分别为：
因此需要继续检查，调用lock_rec_grant函数，进行情况判定，并且决定本次事务lock_t锁资源释放后以什么标准唤醒其堵塞的事务，堵塞的时候又处于何种状态，这就是最复杂的一步，我理解的时候也花了不少时间。
首先需要做的是循环本次释放lock_t锁资源在LOCK SYSTEM中查找，确认是否有其他事务在等待本lock_t锁资源，只要找到一个的话就需要详细的判定了，如果没知道就不需要了，就简单的释放本次lock_t锁资源就可以了，如下，
<code class="language-js_darkmode__26">->for (auto lock = lock_rec_get_first_on_page_addr(lock_hash, page_id);lock != nullptr; lock = lock_rec_get_next_on_page(lock)) ->(lock->is_waiting()) ->found_waiter = true;break;
如果找到则需要相信判定如果堵塞的事务如果获取本次释放的lock_t锁资源，并且预测获取后将会处于何种状态，比如前面举例的就是3种不同的状态和结果，如下：
if (found_waiter)接下来就根据本次释放的lock_t锁资源的 page no和其锁定的heap no（也就是具体锁定的行）进行判定，循环每个bit位来确认是本次需要释放的锁资源，如下：
ulint heap_no = 0; heap_no < lock_rec_get_n_bits(in_lock); ++heap_no)if (lock_rec_get_nth_bit(in_lock, heap_no))然后对每行锁定的资源调入函数lock_rec_grant_by_heap_no，本函数首先Lock_iter::for_each迭代与本次释放锁资源lock_t相关的行锁资源，也就是在LOCK SYSTEM 中查找，找到后使用lambda函数。 本lambda函数首先如果不是堵塞的lock_t结构，则放入granted容器中，如下：
<code class="language-js_darkmode__30"> if (!lock->is_waiting()) { //lock_t 是否没有处于waitting状态 LOCK_WAIT /* Granted locks should be before waiting locks. */ ut_ad(!seen_waiting_lock); granted.push_back(lock); //插入到granted中， return (true); //返回后 回调 lambda函数不返回任何lock_t }如果是处于 LOCK_WAIT状态的lock_t资源则获取这个lock_t资源堵塞者，并且确认是不是本次超时释放的lock_t结构的事务，如果是说明这个LOCK_WAIT的lock_t资源可能会在本次超时释放后获取锁资源，如下：
<code class="language-js_darkmode__31"> const auto blocking_trx = trx->lock.blocking_trx.load(std::memory_order_relaxed); //输出本次迭代lock_t堵塞者的事务ID /* No one should be WAITING without good reason! */ ut_ad(blocking_trx); /* We will only consider granting the `lock`, if we are the reason it was waiting. */ if (blocking_trx != in_trx) { //如果迭代事务的堵塞者不是本次释放lock_t的事务,说明不需要操作 return (true); //直接返回，因为不是本事务堵塞的 }接着如果确实是本次释放lock_t资源堵塞的，那么就需要根据权重分别将它们插入不同的容器中如下：
到底是高权重和还是低权重是其权重是否小于1，而权重的计算我们后面再说，这里只需要知道权重和其等待时长和其堵塞的事务多少有关。 接着对于low_priority_heavier容器，虽然都是高权重事务，但是还是要分高下的，因此需要根据权重排序如下，
<code class="language-js_darkmode__35"> std::stable_sort(low_priority_heavier.begin(), low_priority_heavier.end(), [](const LockDescriptorEx &a, const LockDescriptorEx &b) { return (a.first > b.first); }); //进行排序，根据是schedule_weight权重进行比较完成后又放入low_priority_heavier容器，然后将高权重排序好的lock_t锁资源放入到waiting容器中，然后再将低权重排序好的lock_t资源放入waiting容器中，如下，
for (const auto &descriptor : low_priority_heavier) { waiting.push_back(descriptor.second); //高权重加入 } waiting.insert(waiting.end(), low_priority_light.begin(), low_priority_light.end()); //低权重加入那么这个时候waitting容器实际上如下，
堵塞中的高优先级事务|堵塞中的按照权重排序的重权重事务|堵塞中的轻权重事务这个顺序就是锁资源抢占的顺序。
在做这个动作之前需要将granted容器的长度扩大，扩大至greanted容器长度+waitting容器长度，因为每次迭代，如果事务抢占本次释放锁资源成功，也会放进greanted容器，最多当然就是waitting容器中的事务的锁资源都获取成功了，最大长度就是其长度了，并且需要记录granted容器的原始长度，因为前一段是抢占锁资源之前的长度，后半部分是抢占中或者抢占后的状态，如下：
/* New granted locks will be added from this index. */ const auto new_granted_index = granted.size(); granted.reserve(granted.size() + waiting.size()); 接着就是遍历waitting容器了如下，
for (lock_t *wait_lock : waiting)每次迭代都会带入调入lock_rec_has_to_wait_for_granted函数，本函数就是预测和决定释放哪些事务的根本，首先遍历granted的前半部分，也就是new_granted_index 之前的部分，反向遍历，如下：
for (size_t i = new_granted_index; i--;)const auto granted_lock = granted[i];判定的标准是lock_has_to_wait(wait_lock, granted_lock) ，也就是当前已经获取已经获取当前释放行的锁lock_t结构的事务是否会堵塞接下来获取本行锁lock_t结构的事务。简单的说就是， A<-B<-C<-D，B超时释放行锁资源后，A事务的关于B事务释放行锁对于的lock_t结构是否会堵塞C事务。 如果没有堵塞，那么还需要判定是否新的事务唤醒并且拿到关于本次释放行锁资源的行相关的锁资源是否会堵塞剩下的事务，简单的说就是A<-B<-C<-D，B超时释放行锁资源后，C不会被A堵塞，但是C会不会堵塞D，这个就需要正向迭代granted的后半部分，也就是new_granted_index 之后的部分，如下，
for (size_t i = new_granted_index; i < granted.size(); ++i) if (lock_has_to_wait(wait_lock, granted_lock))如果两部分迭代都不存在堵塞，则返回nullptr，这说明本次lock_t释放后，新事务获取lock_t资源后不会堵塞 ，则开始唤醒这个事务如下，首先是将唤醒事务等待的lock_t锁结构清空，然后唤醒事务的lock_t锁结构去掉LOCK_WAIT状态
<code class="language-js_darkmode__43">lock_reset_lock_and_trx_wait ->lock_reset_lock_and_trx_wait(lock) ->lock->trx->lock.wait_lock = nullptr 既然已经开始给本事务进行lock_t授权了，那么本事务等待的锁应该设置为nullptr ->lock->type_mode &= ~LOCK_WAIT 本lock_t的状态去掉LOCK_WAIT状态
然后就是实际的唤醒本次lock_t结构释放后，获取到锁资源的事务了，
<code class="language-js_darkmode__44">->lock_wait_release_thread_if_suspended ->os_event_set(thr->slot->event)
接着调用lock_rec_move_granted_to_front函数，在LOCK SYSTEM结构中更改，就是要把本次获取的锁资源放到对应链表(cell)的头部。 当然还会将唤醒事务没有堵塞的lock_t结构放入到granted容器中，再次迭代，因为我们说过A<-B<-C<-D 这种堵塞的话，B超时了，C唤醒了，但是不能确认D是不是还会堵塞，需要再次遍历waitting容器，直到全部迭代完，才能确认全部的事务中哪些能够被的唤醒。 如果根据前面的两次循环判定，唤醒的lock_t结构也会堵塞，那么就要更新持有本lock_t结构事务的状态了，实际上就是更新堵塞的源头事务了，比如A<-B<-C<-D，B超时了，C还是会被A堵塞的话，就跑这里，如下，
<code class="language-js_darkmode__45"> ->lock_update_wait_for_edge(wait_lock, blocking_lock) ->waiting_lock->trx->lock.blocking_trx.store(blocking_lock->trx)
完成了上面最重要，最复杂的步骤后，就是唤醒超时的事务了，这个没啥说的了，也就是前面我们说的B事务，调用依旧是lock_wait_release_thread_if_suspended。
本次分析中我们主要理解的，我们还是以A<-B<-C<-D 堵塞，且B事务超时来描述。
后面我们将分析死锁判定方式。
<code class="language-js_darkmode__53">锁堵塞和锁超时 在锁等待期间会通过函数que_thr_stop 如果锁状态为TRX_QUE_LOCK_WAIT，将线程状态设置为 QUE_THR_LOCK_WAIT，经过层层返回，通过函数que_run_threads进行挂起，也就是通过函数 lock_wait_suspend_thread(thr)进行挂起 lock_wait_suspend_thread 逻辑比较简单，主要是找到合适的slot，进行等待，其中几个wait相关的变量更改就在里面 比如wait和wait_current都在里面 ->trx = thr_get_trx(thr) 获取当前的事务 ->trx_lock_wait_timeout_get(trx) 获取超时参数 ->lock_wait_mutex_enter() lock_system_t wait加锁 ->trx_mutex_enter(trx) 上 trx mutex，对于这把锁不仅保护事务的字段，同时保护trx_lock_t的相关属性 ->lock_wait_table_reserve_slot 找到对应的slot，用于timeout线程监测,每次加入一个等待的锁 则lock_wait_table_reservations++，并且放入slot中，也就是说 lock_wait_table_reservations是一个全局的计数器 ->slot = lock_sys->waiting_threads 获取slot链表 ->for (uint32_t i = srv_max_n_threads; i--; ++slot) 循环这个slot数组，找到可用的slot，并且设置属性，为线性循环 比如等待开启的时间，session wait timeout的时间 slot->suspend_time = ut_time_monotonic(); slot->wait_timeout = wait_timeout; ->lock_sys->last_slot增加 ->增加global status响应属性 srv_stats.n_lock_wait_count.inc(); srv_stats.n_lock_wait_current_count.inc(); start_time = ut_time_monotonic_us(); ->lock_wait_mutex_exit() lock_system_t wait解锁 ->os_event_wait(slot->event); 进行等待 ->thd_wait_end(trx->mysql_thd); 等待结束 ->增加和减少global status响应属性 srv_stats.n_lock_wait_current_count.dec(); 减少等待行数 srv_stats.n_lock_wait_time.add(diff_time); 增加等待时间 ->thd_set_lock_wait_time(trx->mysql_thd, diff_time); 增加等待时间，慢查询就会使用这个值lock_wait_timeout_thread 线程1、工作1进行超时检测和锁唤醒 ->每1秒进行循环 ->lock_wait_check_slots_for_timeouts -> for (auto slot = lock_sys->waiting_threads; slot last_slot;++slot) 循环每个slot，这个slot就是前面等待的时候设置的 ->lock_wait_check_and_cancel ->wait_time = ut_time_monotonic() - suspend_time; ->if (wait_time > (int64_t)slot->wait_timeout 如果等待事假你大于了参数设置的时间 ->上lock_system独占锁 locksys::Global_exclusive_latch_guard guard{} ->lock_cancel_waiting_and_release 这里只考虑行锁 ->lock_rec_dequeue_from_page(lock) (这里也是事务提交需要调用的函数) ->lock_rec_discard ->in_lock->index->table->n_rec_locks.fetch_sub(1, std::memory_order_relaxed) 减少加锁的行数 ->locksys::remove_from_trx_locks ->UT_LIST_REMOVE(lock->trx->lock.trx_locks, lock); 从trx_lock_t的trx_locks中去除 lock->trx->lock.trx_locks_version++; 锁版本+1 ->HASH_DELETE(lock_t, hash, lock_hash_get(in_lock->type_mode),lock_rec_fold(page_id), in_lock) 从hash结构中去掉，注意输入的lock_t结构体的内存还存在，下面继续使用 ->lock_rec_grant ->for (auto lock = lock_rec_get_first_on_page_addr(lock_hash, page_id);lock != nullptr; lock = lock_rec_get_next_on_page(lock)) 找到对应需要释放lock_t相应space和page no在hash cell的第一个lock_t结构，循环hash结构的链表，查找是否lock_t锁处于等待状态 因为前面本lock_t已经在hash结构中去掉了，那么找到的肯定是其他事务的lock_t结构 ->(lock->is_waiting()) 处于LOCK_WAIT ->found_waiter = true;break; 只要找到一个处于等待状态和要删除的lock_t相同的space id和page no的就进行跳出，标记found_waiter为ture，这个来自其他session. -> if (found_waiter) 如果hash结构中有lock_t处于等待，则需要详细的检查，如果没有堵塞的则没有必要检查了，检查会循环每个bit位 ->(ulint heap_no = 0; heap_no < lock_rec_get_n_bits(in_lock); ++heap_no) 循环本次释放lock_t的 bit结构 也就是循环 每行记录 ->if (lock_rec_get_nth_bit(in_lock, heap_no)) 查看是否是本次释放lock_t的对应的heap no，确认锁定了多少行 对每行锁定的资源调入函数lock_rec_grant_by_heap_no ->lock_rec_grant_by_heap_no(in_lock, heap_no) 如果本次释放的lock_t，是否有连带的事务需要释放比如 A<-B<-C,比如B超时释放了，C到底是堵塞还是从等待中醒来。 而其他情况比如A释放了，比如提交了也是调用这个函数，到底 B和C是释放还是继续堵塞，而且先释放哪个(权重)都是这里决定的 ->Lock_iter::for_each 迭代整个hash结构，通过page no和heap no(rec_id)进行查找对应的链表(cell) 这里使用lambda函数 ->if (!lock->is_waiting()) 是否没有处于waitting状态 !LOCK_WAIT ->granted.push_back(lock) 插入到granted中 ->函数返回 ->trx = lock->trx 获取迭代迭代lock_t的事务 ->blocking_trx = trx->lock.blocking_trx.load(std::memory_order_relaxed) 输出本次迭代lock_t堵塞者的事务ID ->if (blocking_trx != in_trx) 如果迭代事务的堵塞者不是本次释放lock_t的事务，说明不需要操作 直接返回 ->if (trx_is_high_priority(trx)) 如果事务具有高优先级,这样意味着需要在本次释放lock_t，后具有高有限的 事务，优先获取锁资源，而不会管权重，比如SQL线程的事务 -> waiting.push_back(lock); 直接插入到waiting容器的开头，那么就有优先释放的权利了 ->schedule_weight = trx->lock.schedule_weight.load(std::memory_order_relaxed) 获取权重，这里是原子变量，但不使用内存屏障(不进行任何重排，没有barrier属性，这种方式通常用于计数器) ->if (schedule_weight <= 1) 则说明等待是将不长，并且被其堵塞的事务不多 ->low_priority_light.push_back(lock); 插入轻权重容器，实际上轻权重的容器的事务，释放优先级较低 ->low_priority_heavier.push_back(LockDescriptorEx{schedule_weight, lock}); 否则插入到重权重容器 ->函数返回 ->if (waiting.empty() && low_priority_light.empty() && low_priority_heavier.empty()) 如果没有需要释放的事务，则直接返回 ->std::stable_sort(low_priority_heavier.begin(), low_priority_heavier.end(),[](const LockDescriptorEx &a, const LockDescriptorEx &b) {return (a.first > b.first); }); 这里对重权重容器，根据权重排序，显然权重越高的事务，在本次释放的lock_t后应该优先获取锁资源， 因为其堵塞的事务多，堵塞的时间长。 ->for (const auto &descriptor : low_priority_heavier) 排序完成后迭代low_priority_heavier容器 ->waiting.push_back(descriptor.second) 插入到waiting链表中 ->for (const auto &descriptor : low_priority_heavier) ->waiting.push_back(descriptor.second); 轻权重事务插入到waiting链表 到这里waiting容器中，有如下，堵塞中的高优先级事务，堵塞中的按照权重排序的重权重事务，堵塞中的轻权重事务，那么获取释放 的锁资源就是按照这个顺序来的。 ->const auto new_granted_index = granted.size(); 获取当前的已经获取的lock_t资源的容器的长度 ->granted.reserve(granted.size() + waiting.size()); 将granted容器的长度放大，因为后面释放本次锁资源，其他事务获取资源也会放进来，会循环判定。， 因为下面会判定，如果一个本次lock_t释放后，获取锁资源的事务 是否会造成新的堵塞，比如A<-B<-C<-D,一旦B超时了，那么分好几种情况 情况1:A 事务持有行c锁资源，B事务持有a,b锁资源c锁资源被事务A堵塞了，而C和D事务分别堵塞在a,b锁资源被事务B堵塞了 情况2:A 事务持有行c锁资源，B事务持有a,b锁资源c锁资源被事务A堵塞了，而C和D事务分别堵塞在a锁资源被事务B堵塞了 情况3:A 事务持有行c锁资源，而B,C,D事务分别堵塞在c锁资源被事务A堵塞了 那么一旦B事务超时了，可能情况分为 情况1:C和D事务都应该正常的唤醒，并且继续 情况2:C事务唤醒，D事务继续被C事务堵塞 情况3:C,D事务都被A事务继续堵塞 那么接下来就需要判定进行预测各种情况，那么granted前一部分为当前获取的关于本page和heap no的lock_t,而后半部分就是waiting 容器的内容，当前处于堵塞的。 ->for (lock_t *wait_lock : waiting) 迭代waiting链表中的每个lock_t ->lock_rec_has_to_wait_for_granted 预测主要分为两部分，当前状态和新的事务获取lock_t资源后的状态，这里分别反向扫描granted容器和正向扫描waitting容器 ->for (size_t i = new_granted_index; i--;) 反向迭代当前grant容器，new_granted_index为granted的lock_t ->if (lock_has_to_wait(wait_lock, granted_lock)) 比较当前的需要获取锁资源的lock_t，也就是wait容器里面的lock_t,是否和 现有的lock_t有锁冲突，比如这里的情况3，如果出现堵塞这里就能判定出来 而情况1和2是判定不出来的，因为A事务持有的lock_t和C、D事务是兼容的 ->return (granted_lock); 如果堵塞返回这个当前事务的lock_t,也就是事务A的 ->for (size_t i = new_granted_index; i < granted.size(); ++i) 正向迭代waiting容器，也就是granted容器的后半部分，进行释放后 堵塞的预测，也就是使用优先释放锁的顺序进行预测，这里就是预测 情况1和情况2，这里才能预测出来 ->if (lock_has_to_wait(wait_lock, granted_lock)) 如果堵塞则返回当前lock_t释放后优先授权的lock_t ->return (nullptr); 如果不存在堵塞，则返回nullptr，比如情况1就不存在堵塞了 ->if (blocking_lock == nullptr) 如果本次lock_t释放后，新事务获取lock_t资源后不会堵塞 ->lock_grant(wait_lock) ->lock_reset_wait_and_release_thread_if_suspended 加trx mutex，唤醒事务，本次授权的事务 ->lock_reset_lock_and_trx_wait(lock) ->lock->trx->lock.wait_lock = nullptr 既然已经开始给本事务进行lock_t授权了，那么本事务等待的锁应该设置为nullptr ->lock->type_mode &= ~LOCK_WAIT 本lock_t的状态去掉LOCK_WAIT状态 ->lock_wait_release_thread_if_suspended ->os_event_set(thr->slot->event) 实际就是按照slot进行唤醒，因为所有的堵塞事务都会在slot中注册 ->lock_rec_move_granted_to_front 修改LOCK SYSTEM的hash结构，主要是从现有链表(cell)中删除 然后插入到现有链表(cell)的头部 ->granted.push_back 加入到granted容器中继续预测，加入到尾部?貌似无所谓，反正都要预测 -> 否则,也就是blocking_lock存在则说明本次lock_t释放后，获取lock_t资源的事物会继续堵塞 ->lock_update_wait_for_edge(wait_lock, blocking_lock) ->waiting_lock->trx->lock.blocking_trx.store(blocking_lock->trx) 更改堵塞新获取锁资源事务的堵塞源头事务 ->lock_reset_wait_and_release_thread_if_suspended 这个函数和前面一样。这里是唤醒的是本次超时的事务，不做描述合作电话：010-64087828
社区邮箱：greatsql@greatdb.com
