MDEV-19435 buf_fix_count > 0 for corrupted page when it exits the LRU list

Problem:
=========
One of the purge thread access the corrupted page and tries to remove from
LRU list. In the mean time, other purge threads are waiting for same page
in buf_wait_for_read(). Assertion(buf_fix_count == 0) fails for the
purge thread which tries to remove the page from LRU list.

Solution:
========
- Set the page id as FIL_NULL to indicate the page is corrupted before
removing the block from LRU list. Acquire hash lock for the particular
page id and wait for the other threads to release buf_fix_count
for the block.

- Added the error check for btr_cur_open() in row_search_on_row_ref().

mysql-test/suite/innodb/t/leaf_page_corrupted_during_recovery.test

@@ -9,6 +9,7 @@ call mtr.add_suppression("\\[ERROR\\] InnoDB: Plugin initialization aborted at s
 call mtr.add_suppression("\\[ERROR\\] Plugin 'InnoDB' (init function|registration)");
 call mtr.add_suppression("\\[ERROR\\] InnoDB: We detected index corruption");
 call mtr.add_suppression("\\[ERROR\\] mysqld.*: Index for table 't1' is corrupt; try to repair it");
+call mtr.add_suppression("InnoDB:  Error code: [0-9][0-9][0-9]* btr_pcur_open_low  level: 0 called from file: ");
 --enable_query_log
 CREATE TABLE t1 (pk INT PRIMARY KEY, c CHAR(255))ENGINE=InnoDB STATS_PERSISTENT=0;
 

storage/innobase/buf/buf0buf.cc

@@ -4801,6 +4801,23 @@ buf_page_get_gen(
 	and block->lock. */
 	buf_wait_for_read(fix_block);
 
+	if (fix_block->page.id != page_id) {
+
+		buf_block_unfix(fix_block);
+
+#ifdef UNIV_DEBUG
+		if (!fsp_is_system_temporary(page_id.space())) {
+			rw_lock_s_unlock(&fix_block->debug_latch);
+		}
+#endif /* UNIV_DEBUG */
+
+		if (err) {
+			*err = DB_PAGE_CORRUPTED;
+		}
+
+		return NULL;
+	}
+
 	mtr_memo_type_t	fix_type;
 
 	switch (rw_latch) {
@@ -5775,13 +5792,18 @@ buf_corrupt_page_release(buf_page_t* bpage, const fil_space_t* space)
 	buf_pool_t*	buf_pool = buf_pool_from_bpage(bpage);
 	const ibool	uncompressed = (buf_page_get_state(bpage)
 					== BUF_BLOCK_FILE_PAGE);
+	page_id_t	old_page_id = bpage->id;
 
 	/* First unfix and release lock on the bpage */
 	buf_pool_mutex_enter(buf_pool);
 	mutex_enter(buf_page_get_mutex(bpage));
 	ut_ad(buf_page_get_io_fix(bpage) == BUF_IO_READ);
-	ut_ad(bpage->buf_fix_count == 0);
+	ut_ad(bpage->id.space() == space->id);
+
+	/* buf_fix_count can be greater than zero. Because other thread
+	can wait in buf_page_wait_read() for the page to be read. */
 
+	bpage->id.set_corrupt_id();
 	/* Set BUF_IO_NONE before we remove the block from LRU list */
 	buf_page_set_io_fix(bpage, BUF_IO_NONE);
 
@@ -5798,7 +5820,7 @@ buf_corrupt_page_release(buf_page_t* bpage, const fil_space_t* space)
 	}
 
 	/* After this point bpage can't be referenced. */
-	buf_LRU_free_one_page(bpage);
+	buf_LRU_free_one_page(bpage, old_page_id);
 
 	ut_ad(buf_pool->n_pend_reads > 0);
 	buf_pool->n_pend_reads--;

storage/innobase/buf/buf0lru.cc

@@ -2166,25 +2166,31 @@ buf_LRU_block_free_hashed_page(
 	buf_page_mutex_exit(block);
 }
 
-/******************************************************************//**
-Remove one page from LRU list and put it to free list */
-void
-buf_LRU_free_one_page(
-/*==================*/
-	buf_page_t*	bpage)	/*!< in/out: block, must contain a file page and
-				be in a state where it can be freed; there
-				may or may not be a hash index to the page */
+/** Remove one page from LRU list and put it to free list.
+@param[in,out]	bpage		block, must contain a file page and be in
+				a freeable state; there may or may not be a
+				hash index to the page
+@param[in]	old_page_id	page number before bpage->id was invalidated */
+void buf_LRU_free_one_page(buf_page_t* bpage, page_id_t old_page_id)
 {
 	buf_pool_t*	buf_pool = buf_pool_from_bpage(bpage);
-
-	rw_lock_t*	hash_lock = buf_page_hash_lock_get(buf_pool, bpage->id);
+	rw_lock_t*	hash_lock = buf_page_hash_lock_get(buf_pool,
+							   old_page_id);
 	BPageMutex*	block_mutex = buf_page_get_mutex(bpage);
 
 	ut_ad(buf_pool_mutex_own(buf_pool));
 
 	rw_lock_x_lock(hash_lock);
+
+	while (buf_block_get_fix(bpage) > 0) {
+		/* Wait for other threads to release the fix count
+		before releasing the bpage from LRU list. */
+	}
+
 	mutex_enter(block_mutex);
 
+	bpage->id = old_page_id;
+
 	if (buf_LRU_block_remove_hashed(bpage, true)) {
 		buf_LRU_block_free_hashed_page((buf_block_t*) bpage);
 	}

storage/innobase/buf/buf0rea.cc

@@ -63,13 +63,14 @@ buf_read_page_handle_error(
 	buf_pool_t*	buf_pool = buf_pool_from_bpage(bpage);
 	const bool	uncompressed = (buf_page_get_state(bpage)
 					== BUF_BLOCK_FILE_PAGE);
+	const page_id_t	old_page_id = bpage->id;
 
 	/* First unfix and release lock on the bpage */
 	buf_pool_mutex_enter(buf_pool);
 	mutex_enter(buf_page_get_mutex(bpage));
 	ut_ad(buf_page_get_io_fix(bpage) == BUF_IO_READ);
-	ut_ad(bpage->buf_fix_count == 0);
 
+	bpage->id.set_corrupt_id();
 	/* Set BUF_IO_NONE before we remove the block from LRU list */
 	buf_page_set_io_fix(bpage, BUF_IO_NONE);
 
@@ -82,7 +83,7 @@ buf_read_page_handle_error(
 	mutex_exit(buf_page_get_mutex(bpage));
 
 	/* remove the block from LRU list */
-	buf_LRU_free_one_page(bpage);
+	buf_LRU_free_one_page(bpage, old_page_id);
 
 	ut_ad(buf_pool->n_pend_reads > 0);
 	buf_pool->n_pend_reads--;

storage/innobase/include/buf0buf.h

 /*****************************************************************************
 
 Copyright (c) 1995, 2016, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2013, 2018, MariaDB Corporation.
+Copyright (c) 2013, 2019, MariaDB Corporation.
 
 This program is free software; you can redistribute it and/or modify it under
 the terms of the GNU General Public License as published by the Free Software
@@ -609,31 +609,27 @@ buf_block_buf_fix_inc_func(
 @return the count */
 UNIV_INLINE
 ulint
-buf_block_fix(
-	buf_page_t*	bpage);
+buf_block_fix(buf_page_t* bpage);
 
 /** Increments the bufferfix count.
 @param[in,out]	block	block to bufferfix
 @return the count */
 UNIV_INLINE
 ulint
-buf_block_fix(
-	buf_block_t*	block);
+buf_block_fix(buf_block_t* block);
 
 /** Decrements the bufferfix count.
 @param[in,out]	bpage	block to bufferunfix
 @return	the remaining buffer-fix count */
 UNIV_INLINE
 ulint
-buf_block_unfix(
-	buf_page_t*	bpage);
+buf_block_unfix(buf_page_t* bpage);
 /** Decrements the bufferfix count.
 @param[in,out]	block	block to bufferunfix
 @return	the remaining buffer-fix count */
 UNIV_INLINE
 ulint
-buf_block_unfix(
-	buf_block_t*	block);
+buf_block_unfix(buf_block_t* block);
 
 # ifdef UNIV_DEBUG
 /** Increments the bufferfix count.
@@ -1435,7 +1431,7 @@ class buf_page_t {
 	page_size_t	size;
 
 	/** Count of how manyfold this block is currently bufferfixed. */
-	ib_uint32_t	buf_fix_count;
+	int32		buf_fix_count;
 
 	/** type of pending I/O operation; also protected by
 	buf_pool->mutex for writes only */

storage/innobase/include/buf0buf.ic

@@ -2,7 +2,7 @@
 
 Copyright (c) 1995, 2016, Oracle and/or its affiliates. All Rights Reserved.
 Copyright (c) 2008, Google Inc.
-Copyright (c) 2014, 2017, MariaDB Corporation.
+Copyright (c) 2014, 2019, MariaDB Corporation.
 
 Portions of this file contain modifications contributed and copyrighted by
 Google, Inc. Those modifications are gratefully acknowledged and are described
@@ -950,10 +950,11 @@ buf_block_get_modify_clock(
 @return the count */
 UNIV_INLINE
 ulint
-buf_block_fix(
-	buf_page_t*	bpage)
+buf_block_fix(buf_page_t* bpage)
 {
-	return(my_atomic_add32((int32*) &bpage->buf_fix_count, 1) + 1);
+	return uint32(my_atomic_add32_explicit(
+			      &bpage->buf_fix_count, 1,
+			      MY_MEMORY_ORDER_RELAXED)) + 1;
 }
 
 /** Increments the bufferfix count.
@@ -961,10 +962,30 @@ buf_block_fix(
 @return the count */
 UNIV_INLINE
 ulint
-buf_block_fix(
-	buf_block_t*	block)
+buf_block_fix(buf_block_t* block)
 {
-	return(buf_block_fix(&block->page));
+	return buf_block_fix(&block->page);
+}
+
+/** Get the bufferfix count.
+@param[in]	bpage	block to bufferfix
+@return the count */
+UNIV_INLINE
+ulint
+buf_block_get_fix(buf_page_t* bpage)
+{
+	return my_atomic_load32_explicit(&bpage->buf_fix_count,
+					 MY_MEMORY_ORDER_RELAXED);
+}
+
+/** Get the bufferfix count.
+@param[in]	bpage	block to bufferfix
+@return the count */
+UNIV_INLINE
+ulint
+buf_block_get_fix(buf_block_t* block)
+{
+	return buf_block_get_fix(&block->page);
 }
 
 /*******************************************************************//**
@@ -998,12 +1019,13 @@ buf_block_buf_fix_inc_func(
 @return	the remaining buffer-fix count */
 UNIV_INLINE
 ulint
-buf_block_unfix(
-	buf_page_t*	bpage)
+buf_block_unfix(buf_page_t* bpage)
 {
-	ulint	count = my_atomic_add32((int32*) &bpage->buf_fix_count, -1) - 1;
-	ut_ad(count + 1 != 0);
-	return(count);
+	uint32	count = uint32(my_atomic_add32_explicit(
+				       &bpage->buf_fix_count,
+				       -1, MY_MEMORY_ORDER_RELAXED));
+	ut_ad(count != 0);
+	return count - 1;
 }
 
 /** Decrements the bufferfix count.
@@ -1011,10 +1033,9 @@ buf_block_unfix(
 @return the remaining buffer-fix count */
 UNIV_INLINE
 ulint
-buf_block_unfix(
-	buf_block_t*	block)
+buf_block_unfix(buf_block_t* block)
 {
-	return(buf_block_unfix(&block->page));
+	return buf_block_unfix(&block->page);
 }
 
 /*******************************************************************//**

storage/innobase/include/buf0lru.h

@@ -201,14 +201,12 @@ void
 buf_LRU_stat_update(void);
 /*=====================*/
 
-/******************************************************************//**
-Remove one page from LRU list and put it to free list */
-void
-buf_LRU_free_one_page(
-/*==================*/
-	buf_page_t*	bpage)	/*!< in/out: block, must contain a file page and
-				be in a state where it can be freed; there
-				may or may not be a hash index to the page */
+/** Remove one page from LRU list and put it to free list.
+@param[in,out]	bpage		block, must contain a file page and be in
+				a freeable state; there may or may not be a
+				hash index to the page
+@param[in]	old_page_id	page number before bpage->id was invalidated */
+void buf_LRU_free_one_page(buf_page_t* bpage, page_id_t old_page_id)
 	MY_ATTRIBUTE((nonnull));
 
 /******************************************************************//**

storage/innobase/include/buf0types.h

@@ -175,6 +175,12 @@ class page_id_t {
 		ut_ad(page_no <= 0xFFFFFFFFU);
 	}
 
+	/** Set the FIL_NULL for the space and page_no */
+	void set_corrupt_id()
+	{
+		m_space = m_page_no = ULINT32_UNDEFINED;
+	}
+
 private:
 
 	/** Tablespace id. */

storage/innobase/row/row0row.cc

@@ -980,7 +980,10 @@ row_search_on_row_ref(
 
 	ut_a(dtuple_get_n_fields(ref) == dict_index_get_n_unique(index));
 
-	btr_pcur_open(index, ref, PAGE_CUR_LE, mode, pcur, mtr);
+	if (btr_pcur_open(index, ref, PAGE_CUR_LE, mode, pcur, mtr)
+	    != DB_SUCCESS) {
+		return FALSE;
+	}
 
 	low_match = btr_pcur_get_low_match(pcur);