Bug#13789853 SHOW ENGINE INNODB STATUS HANGS DUE TO EXCESSIVE WORK

IN LOCK_VALIDATE()

rb://917
approved by: Marko Makela

In lock_validate() the limit is used to release the kernel_mutex during
the validation, to obey the latching order.
If we do the limit++ then we are rechecking the same lock most times on
each iteration because limit is being incremented by one and
<space, page_no> will nearly always be > limit. If we set the limit
correctly to (space, page+1) then we are actually making progress
during the iteration.

storage/innobase/lock/lock0lock.c

@@ -4964,6 +4964,74 @@ function_exit:
 	return(TRUE);
 }
 
+/*********************************************************************//**
+Validate record locks up to a limit.
+@return lock at limit or NULL if no more locks in the hash bucket */
+static __attribute__((nonnull, warn_unused_result))
+const lock_t*
+lock_rec_validate(
+/*==============*/
+	ulint		start,		/*!< in: lock_sys->rec_hash
+					bucket */
+	ib_uint64_t*	limit)		/*!< in/out: upper limit of
+					(space, page_no) */
+{
+	lock_t*		lock;
+	ut_ad(mutex_own(&kernel_mutex));
+
+	for (lock = HASH_GET_FIRST(lock_sys->rec_hash, start);
+	     lock != NULL;
+	     lock = HASH_GET_NEXT(hash, lock)) {
+
+		ib_uint64_t	current;
+
+		ut_a(trx_in_trx_list(lock->trx));
+		ut_a(lock_get_type(lock) == LOCK_REC);
+
+		current = ut_ull_create(
+			lock->un_member.rec_lock.space,
+			lock->un_member.rec_lock.page_no);
+
+		if (current > *limit) {
+			*limit = current + 1;
+			return(lock);
+		}
+	}
+
+	return(NULL);
+}
+
+/*********************************************************************//**
+Validate a record lock's block */
+static
+void
+lock_rec_block_validate(
+/*====================*/
+	ulint		space,
+	ulint		page_no)
+{
+	/* The lock and the block that it is referring to may be freed at
+	this point. We pass BUF_GET_POSSIBLY_FREED to skip a debug check.
+	If the lock exists in lock_rec_validate_page() we assert
+	!block->page.file_page_was_freed. */
+
+	mtr_t		mtr;
+	buf_block_t*	block;
+
+	mtr_start(&mtr);
+
+	block = buf_page_get_gen(
+		space, fil_space_get_zip_size(space),
+		page_no, RW_X_LATCH, NULL,
+		BUF_GET_POSSIBLY_FREED,
+		__FILE__, __LINE__, &mtr);
+
+	buf_block_dbg_add_level(block, SYNC_NO_ORDER_CHECK);
+
+	ut_ad(lock_rec_validate_page(block));
+	mtr_commit(&mtr);
+}
+
 /*********************************************************************//**
 Validates the lock system.
 @return	TRUE if ok */
@@ -4996,60 +5064,21 @@ lock_validate(void)
 		trx = UT_LIST_GET_NEXT(trx_list, trx);
 	}
 
-	for (i = 0; i < hash_get_n_cells(lock_sys->rec_hash); i++) {
-
-		ulint		space;
-		ulint		page_no;
-		ib_uint64_t	limit	= 0;
-
-		for (;;) {
-			mtr_t		mtr;
-			buf_block_t*	block;
-
-			lock = HASH_GET_FIRST(lock_sys->rec_hash, i);
-
-			while (lock) {
-				ib_uint64_t	space_page;
-				ut_a(trx_in_trx_list(lock->trx));
-
-				space = lock->un_member.rec_lock.space;
-				page_no = lock->un_member.rec_lock.page_no;
+	/* Iterate over all the record locks and validate the locks. We
+	don't want to hog the lock_sys_t::mutex and the trx_sys_t::mutex.
+	Release both mutexes during the validation check. */
 
-				space_page = ut_ull_create(space, page_no);
-
-				if (space_page >= limit) {
-					break;
-				}
-
-				lock = HASH_GET_NEXT(hash, lock);
-			}
+	for (i = 0; i < hash_get_n_cells(lock_sys->rec_hash); i++) {
+		const lock_t*	lock;
+		ib_uint64_t	limit = 0;
 
-			if (!lock) {
+		while ((lock = lock_rec_validate(i, &limit)) != NULL) {
 
-				break;
-			}
+			ulint	space = lock->un_member.rec_lock.space;
+			ulint	page_no = lock->un_member.rec_lock.page_no;
 
 			lock_mutex_exit_kernel();
-
-			/* The lock and the block that it is referring
-			to may be freed at this point. We pass
-			BUF_GET_POSSIBLY_FREED to skip a debug check.
-			If the lock exists in lock_rec_validate_page()
-			we assert !block->page.file_page_was_freed. */
-
-			mtr_start(&mtr);
-			block = buf_page_get_gen(
-				space, fil_space_get_zip_size(space),
-				page_no, RW_X_LATCH, NULL,
-				BUF_GET_POSSIBLY_FREED,
-				__FILE__, __LINE__, &mtr);
-			buf_block_dbg_add_level(block, SYNC_NO_ORDER_CHECK);
-
-			ut_ad(lock_rec_validate_page(block));
-			mtr_commit(&mtr);
-
-			limit++;
-
+			lock_rec_block_validate(space, page_no);
 			lock_mutex_enter_kernel();
 		}
 	}