一 序

   本文接上一篇介紹鎖型別之後。主要分為 兩部分。第一部分介紹表鎖行鎖加鎖流程。第二部分常見的死鎖檢測。

   InnoDB 所有的事務鎖物件都是掛在全域性物件lock_sys上，同時每個事務物件上也維持了其擁有的事務鎖，每個表物件(dict_table_t)上維持了構建在其上的表級鎖物件。下圖來自mysql.taobao.

二 表鎖

表鎖相關結構： 

 table->locks：資料字典table儲存這個表上的所有表鎖資訊   

 trx->lock.table_locks：每個事務trx儲存該事務所加的所有表鎖資訊   

 trx->lock.trx_locks：每個事務trx儲存該事務所有的鎖資訊（包括行鎖）

 lock_table 加鎖流程

• 首先從當前事務的trx_lock_t::table_locks中查詢是否已經加了等同或更高級別的表鎖，如果已經加鎖了，則直接返回成功（lock_table_has）；
• 檢查當前是否有和正在申請的鎖模式衝突的表級鎖物件（lock_table_other_has_incompatible）；
• 如果存在衝突的鎖物件，則需要進入等待佇列（lock_table_enqueue_waiting）
  • 建立等待鎖物件 （lock_table_create）
  • 檢查是否存在死鎖（DeadlockChecker::check_and_resolve），當存在死鎖時：如果當前會話被選作犧牲者，就移除鎖請求(lock_table_remove_low
    )，重置當前事務的wait_lock為空，並返回錯誤碼DB_DEADLOCK；若被選成勝利者，則鎖等待解除，可以認為當前會話已經獲得了鎖，返回成功；
  • 若沒有發生死鎖，設定事務物件的相關變數後，返回錯誤碼DB_LOCK_WAIT，隨後進入鎖等待狀態
• 如果不存在衝突的鎖，則直接建立鎖物件（lock_table_create），加入佇列。

lock_table_create: 建立鎖物件  原始碼在innobase/lock/lock0lock.cc

/*********************************************************************//**
Creates a table lock object and adds it as the last in the lock queue
of the table. Does NOT check for deadlocks or lock compatibility.
@return own: new lock object */
UNIV_INLINE
lock_t*
lock_table_create(
/*==============*/
	dict_table_t*	table,	/*!< in/out: database table
				in dictionary cache */
	ulint		type_mode,/*!< in: lock mode possibly ORed with
				LOCK_WAIT */
	trx_t*		trx)	/*!< in: trx */
{
	lock_t*		lock;

	ut_ad(table && trx);
	ut_ad(lock_mutex_own());
	ut_ad(trx_mutex_own(trx));

	check_trx_state(trx);

	if ((type_mode & LOCK_MODE_MASK) == LOCK_AUTO_INC) {
		++table->n_waiting_or_granted_auto_inc_locks;
	}

	/* For AUTOINC locking we reuse the lock instance only if
	there is no wait involved else we allocate the waiting lock
	from the transaction lock heap. */
	if (type_mode == LOCK_AUTO_INC) {

		lock = table->autoinc_lock;

		table->autoinc_trx = trx;

		ib_vector_push(trx->autoinc_locks, &lock);

	} else if (trx->lock.table_cached < trx->lock.table_pool.size()) {
		lock = trx->lock.table_pool[trx->lock.table_cached++];
	} else {

		lock = static_cast<lock_t*>(
			mem_heap_alloc(trx->lock.lock_heap, sizeof(*lock)));

	}

	lock->type_mode = ib_uint32_t(type_mode | LOCK_TABLE);
	lock->trx = trx;

	lock->un_member.tab_lock.table = table;

	ut_ad(table->n_ref_count > 0 || !table->can_be_evicted);

	UT_LIST_ADD_LAST(trx->lock.trx_locks, lock);

	ut_list_append(table->locks, lock, TableLockGetNode());

	if (type_mode & LOCK_WAIT) {

		lock_set_lock_and_trx_wait(lock, trx);
	}

	lock->trx->lock.table_locks.push_back(lock);

	MONITOR_INC(MONITOR_TABLELOCK_CREATED);
	MONITOR_INC(MONITOR_NUM_TABLELOCK);

	return(lock);
}
 

• 當前請求的是AUTO-INC鎖時；
  • 遞增dict_table_t::n_waiting_or_granted_auto_inc_locks。前面我們已經提到過，當這個值非0時，對於自增列的插入操作就會退化到OLD-STYLE;
  • 鎖物件直接引用已經預先建立好的dict_table_t::autoinc_lock，並加入到trx_t::autoinc_locks集合中;
• 對於非AUTO-INC鎖，則從一個pool中分配鎖物件
  • 在事務物件trx_t::lock中，維持了兩個pool，一個是trx_lock_t::rec_pool，預分配了一組鎖物件用於記錄鎖分配，另外一個是trx_lock_t::table_pool，用於表級鎖的鎖物件分配。通過預分配記憶體的方式，可以避免在持有全域性大鎖時(lock_sys->mutex)進行昂貴的記憶體分配操作。rec_pool和table_pool預分配的大小都為8個鎖物件。（lock_trx_alloc_locks）;
  • 如果table_pool已經用滿，則走記憶體分配，建立一個鎖物件；
• 構建好的鎖物件分別加入到事務的trx_t::lock.trx_locks連結串列上以及表物件的dict_table_t::locks連結串列上；
• 構建好的鎖物件加入到當前事務的trx_t::lock.table_locks集合中。

     可以看到鎖物件會加入到不同的集合或者連結串列中，通過掛載到事務物件上，可以快速檢查當前事務是否已經持有表鎖；通過掛到表物件的鎖鏈表上，可以用於檢查該表上的全域性衝突情況。

三 行鎖

   記錄鎖的儲存單位為頁，一個事務在一個頁上的所有同類型鎖儲存為一個lock,lock中通過bitmap來表示那些記錄已加鎖

 行鎖型別：行鎖型別由基本型別和精確型別組成

 行鎖基本型別：

    LOCK_S,LOCK_X

 行鎖精確型別：

    LOCK_ORDINARY(NK)：next-key lock 既鎖記錄也鎖記錄前的gap

    LOCK_GAP(GAP)：不鎖記錄，只鎖記錄前的gap

    LOCK_REC_NOT_GAP(NG):只鎖記錄，不鎖記錄前的gap

    LOCK_INSERT_INTENTION(I)：insert時，若insert位置已有gap鎖，則需加LOCK_INSERT_INTENTION鎖

相關結構：

  trx->lock.trx_locks：每個事務trx儲存該事務所有的鎖資訊（表鎖和行鎖）

  lock_sys->rec_hash：儲存所有事務的行鎖資訊

給記錄加S鎖之前必會給表加IS鎖

給記錄加X鎖之前必會給表加IX鎖

行加鎖流程 

函式入口： lock_rec_lock  原始碼在innobase/lock/lock0lock.cc

/*********************************************************************//**
Tries to lock the specified record in the mode requested. If not immediately
possible, enqueues a waiting lock request. This is a low-level function
which does NOT look at implicit locks! Checks lock compatibility within
explicit locks. This function sets a normal next-key lock, or in the case
of a page supremum record, a gap type lock.
@return DB_SUCCESS, DB_SUCCESS_LOCKED_REC, DB_LOCK_WAIT, DB_DEADLOCK,
or DB_QUE_THR_SUSPENDED */
static
dberr_t
lock_rec_lock(
/*==========*/
	bool			impl,	/*!< in: if true, no lock is set
					if no wait is necessary: we
					assume that the caller will
					set an implicit lock */
	ulint			mode,	/*!< in: lock mode: LOCK_X or
					LOCK_S possibly ORed to either
					LOCK_GAP or LOCK_REC_NOT_GAP */
	const buf_block_t*	block,	/*!< in: buffer block containing
					the record */
	ulint			heap_no,/*!< in: heap number of record */
	dict_index_t*		index,	/*!< in: index of record */
	que_thr_t*		thr)	/*!< in: query thread */
{
	ut_ad(lock_mutex_own());
	ut_ad(!srv_read_only_mode);
	ut_ad((LOCK_MODE_MASK & mode) != LOCK_S
	      || lock_table_has(thr_get_trx(thr), index->table, LOCK_IS));
	ut_ad((LOCK_MODE_MASK & mode) != LOCK_X
	      || lock_table_has(thr_get_trx(thr), index->table, LOCK_IX));
	ut_ad((LOCK_MODE_MASK & mode) == LOCK_S
	      || (LOCK_MODE_MASK & mode) == LOCK_X);
	ut_ad(mode - (LOCK_MODE_MASK & mode) == LOCK_GAP
	      || mode - (LOCK_MODE_MASK & mode) == LOCK_REC_NOT_GAP
	      || mode - (LOCK_MODE_MASK & mode) == 0);
	ut_ad(dict_index_is_clust(index) || !dict_index_is_online_ddl(index));

	/* We try a simplified and faster subroutine for the most
	common cases */
	switch (lock_rec_lock_fast(impl, mode, block, heap_no, index, thr)) {
	case LOCK_REC_SUCCESS:
		return(DB_SUCCESS);
	case LOCK_REC_SUCCESS_CREATED:
		return(DB_SUCCESS_LOCKED_REC);
	case LOCK_REC_FAIL:
		return(lock_rec_lock_slow(impl, mode, block,
					  heap_no, index, thr));
	}

	ut_error;
	return(DB_ERROR);
}

/*********************************************************************//**
This is a fast routine for locking a record in the most common cases:
there are no explicit locks on the page, or there is just one lock, owned
by this transaction, and of the right type_mode. This is a low-level function
which does NOT look at implicit locks! Checks lock compatibility within
explicit locks. This function sets a normal next-key lock, or in the case of
a page supremum record, a gap type lock.
@return whether the locking succeeded */
UNIV_INLINE
lock_rec_req_status
lock_rec_lock_fast(
/*===============*/
	bool			impl,	/*!< in: if TRUE, no lock is set
					if no wait is necessary: we
					assume that the caller will
					set an implicit lock */
	ulint			mode,	/*!< in: lock mode: LOCK_X or
					LOCK_S possibly ORed to either
					LOCK_GAP or LOCK_REC_NOT_GAP */
	const buf_block_t*	block,	/*!< in: buffer block containing
					the record */
	ulint			heap_no,/*!< in: heap number of record */
	dict_index_t*		index,	/*!< in: index of record */
	que_thr_t*		thr)	/*!< in: query thread */
{
	ut_ad(lock_mutex_own());
	ut_ad(!srv_read_only_mode);
	ut_ad((LOCK_MODE_MASK & mode) != LOCK_S
	      || lock_table_has(thr_get_trx(thr), index->table, LOCK_IS));
	ut_ad((LOCK_MODE_MASK & mode) != LOCK_X
	      || lock_table_has(thr_get_trx(thr), index->table, LOCK_IX)
	      || srv_read_only_mode);
	ut_ad((LOCK_MODE_MASK & mode) == LOCK_S
	      || (LOCK_MODE_MASK & mode) == LOCK_X);
	ut_ad(mode - (LOCK_MODE_MASK & mode) == LOCK_GAP
	      || mode - (LOCK_MODE_MASK & mode) == 0
	      || mode - (LOCK_MODE_MASK & mode) == LOCK_REC_NOT_GAP);
	ut_ad(dict_index_is_clust(index) || !dict_index_is_online_ddl(index));

	DBUG_EXECUTE_IF("innodb_report_deadlock", return(LOCK_REC_FAIL););

	lock_t*	lock = lock_rec_get_first_on_page(lock_sys->rec_hash, block);

	trx_t*	trx = thr_get_trx(thr);

	lock_rec_req_status	status = LOCK_REC_SUCCESS;

	if (lock == NULL) {

		if (!impl) {
			RecLock	rec_lock(index, block, heap_no, mode);

			/* Note that we don't own the trx mutex. */
			rec_lock.create(trx, false, true);
		}

		status = LOCK_REC_SUCCESS_CREATED;
	} else {
		trx_mutex_enter(trx);

		if (lock_rec_get_next_on_page(lock)
		     || lock->trx != trx
		     || lock->type_mode != (mode | LOCK_REC)
		     || lock_rec_get_n_bits(lock) <= heap_no) {

			status = LOCK_REC_FAIL;
		} else if (!impl) {
			/* If the nth bit of the record lock is already set
			then we do not set a new lock bit, otherwise we do
			set */
			if (!lock_rec_get_nth_bit(lock, heap_no)) {
				lock_rec_set_nth_bit(lock, heap_no);
				status = LOCK_REC_SUCCESS_CREATED;
			}
		}

		trx_mutex_exit(trx);
	}

	return(status);
}

/*********************************************************************//**
This is the general, and slower, routine for locking a record. This is a
low-level function which does NOT look at implicit locks! Checks lock
compatibility within explicit locks. This function sets a normal next-key
lock, or in the case of a page supremum record, a gap type lock.
@return DB_SUCCESS, DB_SUCCESS_LOCKED_REC, DB_LOCK_WAIT, DB_DEADLOCK,
or DB_QUE_THR_SUSPENDED */
static
dberr_t
lock_rec_lock_slow(
/*===============*/
	ibool			impl,	/*!< in: if TRUE, no lock is set
					if no wait is necessary: we
					assume that the caller will
					set an implicit lock */
	ulint			mode,	/*!< in: lock mode: LOCK_X or
					LOCK_S possibly ORed to either
					LOCK_GAP or LOCK_REC_NOT_GAP */
	const buf_block_t*	block,	/*!< in: buffer block containing
					the record */
	ulint			heap_no,/*!< in: heap number of record */
	dict_index_t*		index,	/*!< in: index of record */
	que_thr_t*		thr)	/*!< in: query thread */
{
	ut_ad(lock_mutex_own());
	ut_ad(!srv_read_only_mode);
	ut_ad((LOCK_MODE_MASK & mode) != LOCK_S
	      || lock_table_has(thr_get_trx(thr), index->table, LOCK_IS));
	ut_ad((LOCK_MODE_MASK & mode) != LOCK_X
	      || lock_table_has(thr_get_trx(thr), index->table, LOCK_IX));
	ut_ad((LOCK_MODE_MASK & mode) == LOCK_S
	      || (LOCK_MODE_MASK & mode) == LOCK_X);
	ut_ad(mode - (LOCK_MODE_MASK & mode) == LOCK_GAP
	      || mode - (LOCK_MODE_MASK & mode) == 0
	      || mode - (LOCK_MODE_MASK & mode) == LOCK_REC_NOT_GAP);
	ut_ad(dict_index_is_clust(index) || !dict_index_is_online_ddl(index));

	DBUG_EXECUTE_IF("innodb_report_deadlock", return(DB_DEADLOCK););

	dberr_t	err;
	trx_t*	trx = thr_get_trx(thr);

	trx_mutex_enter(trx);

	if (lock_rec_has_expl(mode, block, heap_no, trx)) {

		/* The trx already has a strong enough lock on rec: do
		nothing */

		err = DB_SUCCESS;

	} else {

		const lock_t* wait_for = lock_rec_other_has_conflicting(
			mode, block, heap_no, trx);

		if (wait_for != NULL) {

			/* If another transaction has a non-gap conflicting
			request in the queue, as this transaction does not
			have a lock strong enough already granted on the
			record, we may have to wait. */

			RecLock	rec_lock(thr, index, block, heap_no, mode);

			err = rec_lock.add_to_waitq(wait_for);

		} else if (!impl) {

			/* Set the requested lock on the record, note that
			we already own the transaction mutex. */

			lock_rec_add_to_queue(
				LOCK_REC | mode, block, heap_no, index, trx,
				true);

			err = DB_SUCCESS_LOCKED_REC;
		} else {
			err = DB_SUCCESS;
		}
	}

	trx_mutex_exit(trx);

	return(err);
}

行級鎖加鎖的入口函式為lock_rec_lock，其中第一個引數impl如果為TRUE，則噹噹前記錄上已有的鎖和LOCK_X | LOCK_REC_NOT_GAP不衝突時，就無需建立鎖物件。（見上文關於記錄鎖LOCK_X相關描述部分），為了描述清晰，下文的流程描述，預設impl為FALSE。

lock_rec_lock：

• 首先嚐試fast lock的方式，對於衝突少的場景，這是比較普通的加鎖方式(lock_rec_lock_fast), 符合如下情況時，可以走fast lock:
  • 記錄所在的page上沒有任何記錄鎖時，直接建立鎖物件，加入rec_hash，並返回成功;
  • 記錄所在的page上只存在一個記錄鎖，並且屬於當前事務，且這個記錄鎖預分配的bitmap能夠描述當前的heap no（預分配的bit數為建立鎖物件時的page上記錄數 + 64，參閱函式RecLock::lock_size），則直接設定對應的bit位並返回;
• 無法走fast lock時，再呼叫slow lock的邏輯(lock_rec_lock_slow)
  • 判斷當前事務是否已經持有了一個優先順序更高的鎖，如果是的話，直接返回成功（lock_rec_has_expl）;
  • 檢查是否存在和當前申請鎖模式衝突的鎖（lock_rec_other_has_conflicting），如果存在的話，就建立一個鎖物件（RecLock::RecLock），並加入到等待佇列中（RecLock::add_to_waitq），這裡會進行死鎖檢測;
  • 如果沒有衝突的鎖，則入佇列（lock_rec_add_to_queue）：已經有在同一個Page上的鎖物件且沒有別的會話等待相同的heap no時，可以直接設定對應的bitmap（lock_rec_find_similar_on_page）；否則需要建立一個新的鎖物件;
• 返回錯誤碼，對於DB_LOCK_WAIT, DB_DEADLOCK等錯誤碼，會在上層進行處理。

四 死鎖檢測

/**
Check and resolve any deadlocks
@param[in, out] lock		The lock being acquired
@return DB_LOCK_WAIT, DB_DEADLOCK, or DB_QUE_THR_SUSPENDED, or
	DB_SUCCESS_LOCKED_REC; DB_SUCCESS_LOCKED_REC means that
	there was a deadlock, but another transaction was chosen
	as a victim, and we got the lock immediately: no need to
	wait then */
dberr_t
RecLock::deadlock_check(lock_t* lock)
{
	ut_ad(lock_mutex_own());
	ut_ad(lock->trx == m_trx);
	ut_ad(trx_mutex_own(m_trx));

	const trx_t*	victim_trx =
			DeadlockChecker::check_and_resolve(lock, m_trx);

	/* Check the outcome of the deadlock test. It is possible that
	the transaction that blocked our lock was rolled back and we
	were granted our lock. */

	dberr_t	err = check_deadlock_result(victim_trx, lock);

	if (err == DB_LOCK_WAIT) {

		set_wait_state(lock);

		MONITOR_INC(MONITOR_LOCKREC_WAIT);
	}

	return(err);
}

/** Checks if a joining lock request results in a deadlock. If a deadlock is
found this function will resolve the deadlock by choosing a victim transaction
and rolling it back. It will attempt to resolve all deadlocks. The returned
transaction id will be the joining transaction instance or NULL if some other
transaction was chosen as a victim and rolled back or no deadlock found.
@param[in]	lock lock the transaction is requesting
@param[in,out]	trx transaction requesting the lock
@return transaction instanace chosen as victim or 0 */
const trx_t*
DeadlockChecker::check_and_resolve(const lock_t* lock, trx_t* trx)
{
	ut_ad(lock_mutex_own());
	ut_ad(trx_mutex_own(trx));
	check_trx_state(trx);
	ut_ad(!srv_read_only_mode);

	/* If transaction is marked for ASYNC rollback then we should
	not allow it to wait for another lock causing possible deadlock.
	We return current transaction as deadlock victim here. */
	if (trx->in_innodb & TRX_FORCE_ROLLBACK_ASYNC) {
		return(trx);
	} else if (!innobase_deadlock_detect) {
		return(NULL);
	}

	/*  Release the mutex to obey the latching order.
	This is safe, because DeadlockChecker::check_and_resolve()
	is invoked when a lock wait is enqueued for the currently
	running transaction. Because m_trx is a running transaction
	(it is not currently suspended because of a lock wait),
	its state can only be changed by this thread, which is
	currently associated with the transaction. */

	trx_mutex_exit(trx);

	const trx_t*	victim_trx;

	/* Try and resolve as many deadlocks as possible. */
	do {
		DeadlockChecker	checker(trx, lock, s_lock_mark_counter);

		victim_trx = checker.search();

		/* Search too deep, we rollback the joining transaction only
		if it is possible to rollback. Otherwise we rollback the
		transaction that is holding the lock that the joining
		transaction wants. */
		if (checker.is_too_deep()) {

			ut_ad(trx == checker.m_start);
			ut_ad(trx == victim_trx);

			rollback_print(victim_trx, lock);

			MONITOR_INC(MONITOR_DEADLOCK);

			break;

		} else if (victim_trx != NULL && victim_trx != trx) {

			ut_ad(victim_trx == checker.m_wait_lock->trx);

			checker.trx_rollback();

			lock_deadlock_found = true;

			MONITOR_INC(MONITOR_DEADLOCK);
		}

	} while (victim_trx != NULL && victim_trx != trx);

	/* If the joining transaction was selected as the victim. */
	if (victim_trx != NULL) {

		print("*** WE ROLL BACK TRANSACTION (2)\n");

		lock_deadlock_found = true;
	}

	trx_mutex_enter(trx);

	return(victim_trx);
}

當發現有衝突的鎖時，呼叫函式RecLock::add_to_waitq進行判斷。通常我們沒有機會設定事務的優先順序，在建立了一個處於WAIT狀態的鎖物件後，我們需要進行死鎖檢測，死鎖檢測採用深度優先遍歷的方式，通過事務物件上的trx_t::lock.wait_lock構造事務的wait-for graph進行判斷，當最終發現一個鎖請求等待閉環時，可以判定發生了死鎖。另外一種情況是，如果檢測深度過長（即鎖等待的會話形成的檢測鏈路非常長），也會認為發生死鎖，最大深度預設為LOCK_MAX_DEPTH_IN_DEADLOCK_CHECK，值為200。

當發生死鎖時，需要選擇一個犧牲者（DeadlockChecker::select_victim()）來解決死鎖，通常事務權重低的回滾（trx_weight_ge）。

• 修改了非事務表的會話具有更高的權重；
• 如果兩個表都修改了、或者都沒有修改事務表，那麼就根據的事務的undo數量加上持有的事務鎖個數來決定權值（TRX_WEIGHT）；
• 低權重的事務被回滾，高權重的獲得鎖物件。

當無法立刻獲得鎖時，會將錯誤碼傳到上層進行處理（row_mysql_handle_errors）

釋放鎖及喚醒

大多數情況下事務鎖都是在事務提交時釋放，但有兩種意外：

• AUTO-INC鎖在SQL結束時直接釋放（innobase_commit --> lock_unlock_table_autoinc）；
• 在RC隔離級別下執行DML語句時，從引擎層返回到Server層的記錄，如果不滿足where條件，則需要立刻unlock掉（ha_innobase::unlock_row）。

      除這兩種情況外，其他的事務鎖都是在事務提交時釋放的(lock_trx_release_locks --> lock_release)。事務持有的所有鎖都維護在連結串列trx_t::lock.trx_locks上，依次遍歷釋放即可。

      對於行鎖，從全域性hash中刪除後，還需要判斷別的正在等待的會話是否可以被喚醒（lock_rec_dequeue_from_page）。例如如果當前釋放的是某個記錄的X鎖，那麼所有的S鎖請求的會話都可以被喚醒。

       對於表鎖，如果表級鎖的型別不為LOCK_IS，且當前事務修改了資料，就將表物件的dict_table_t::query_cache_inv_id設定為當前最大的事務id。在檢查是否可以使用該表的Query Cache時會使用該值進行判斷（row_search_check_if_query_cache_permitted），如果某個使用者會話的事務物件的low_limit_id（即最大可見事務id）比這個值還小，說明它不應該使用當前table cache的內容，也不應該儲存到query cache中。

表級鎖物件的釋放呼叫函式lock_table_dequeue。

/*************************************************************//**
Removes a table lock request, waiting or granted, from the queue and grants
locks to other transactions in the queue, if they now are entitled to a
lock. */
static
void
lock_table_dequeue(
/*===============*/
	lock_t*	in_lock)/*!< in/out: table lock object; transactions waiting
			behind will get their lock requests granted, if
			they are now qualified to it */
{
	ut_ad(lock_mutex_own());
	ut_a(lock_get_type_low(in_lock) == LOCK_TABLE);

	lock_t*	lock = UT_LIST_GET_NEXT(un_member.tab_lock.locks, in_lock);

	lock_table_remove_low(in_lock);

	/* Check if waiting locks in the queue can now be granted: grant
	locks if there are no conflicting locks ahead. */

	for (/* No op */;
	     lock != NULL;
	     lock = UT_LIST_GET_NEXT(un_member.tab_lock.locks, lock)) {

		if (lock_get_wait(lock)
		    && !lock_table_has_to_wait_in_queue(lock)) {

			/* Grant the lock */
			ut_ad(in_lock->trx != lock->trx);
			lock_grant(lock);
		}
	}
}

注意在釋放鎖時，如果該事務持有的鎖物件太多，每釋放1000（LOCK_RELEASE_INTERVAL）個鎖物件，會暫時釋放下lock_sys->mutex再重新持有，防止InnoDB hang住。

if (count == LOCK_RELEASE_INTERVAL) {
			/* Release the mutex for a while, so that we
			do not monopolize it */

			lock_mutex_exit();

			lock_mutex_enter();

			count = 0;
		}
...

下一篇結合例子來看常見的死鎖。

參考：