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().

8 files changed, 101 insertions, 48 deletions

diff --git a/storage/innobase/buf/buf0buf.cc b/storage/innobase/buf/buf0buf.cc
index 18decb492f3..644d033e1b7 100644
--- a/storage/innobase/buf/buf0buf.cc
+++ b/storage/innobase/buf/buf0buf.cc
@@ -4801,6 +4801,23 @@ evict_from_pool:
 	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--;
diff --git a/storage/innobase/buf/buf0lru.cc b/storage/innobase/buf/buf0lru.cc
index ff20bd570d7..8673c8d9d72 100644
--- a/storage/innobase/buf/buf0lru.cc
+++ b/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);
 	}
diff --git a/storage/innobase/buf/buf0rea.cc b/storage/innobase/buf/buf0rea.cc
index 1b5abe8f252..188d0aa24b6 100644
--- a/storage/innobase/buf/buf0rea.cc
+++ b/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--;
diff --git a/storage/innobase/include/buf0buf.h b/storage/innobase/include/buf0buf.h
index 086bfb12710..e75a1d9ec44 100644
--- a/storage/innobase/include/buf0buf.h
+++ b/storage/innobase/include/buf0buf.h
@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 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 @@ public:
 	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 */
diff --git a/storage/innobase/include/buf0buf.ic b/storage/innobase/include/buf0buf.ic
index 4f1c6a7110a..98026159b8a 100644
--- a/storage/innobase/include/buf0buf.ic
+++ b/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);
 }
 
 /*******************************************************************//**
diff --git a/storage/innobase/include/buf0lru.h b/storage/innobase/include/buf0lru.h
index d81c95fd224..1efbb1f03ef 100644
--- a/storage/innobase/include/buf0lru.h
+++ b/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));
 
 /******************************************************************//**
diff --git a/storage/innobase/include/buf0types.h b/storage/innobase/include/buf0types.h
index 63299681823..27ffee03d4c 100644
--- a/storage/innobase/include/buf0types.h
+++ b/storage/innobase/include/buf0types.h
@@ -175,6 +175,12 @@ public:
 		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. */
diff --git a/storage/innobase/row/row0row.cc b/storage/innobase/row/row0row.cc
index 3bb351eed8f..3e65dc1d28b 100644
--- a/storage/innobase/row/row0row.cc
+++ b/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);