MDEV-23693 Failing assertion: my_atomic_load32_explicit(&lock->lock_word, MY_MEMORY_ORDER_RELAXED) == X_LOCK_DECRbb-10.3-MDEV-23693

InnoDB frees the block lock during buffer pool shrinking when other
thread is yet to release the block lock.  While shrinking the
buffer pool, InnoDB allows the page to be freed unless it is buffer
fixed. In some cases, InnoDB releases the latch after unfixing the
block.

Fix:
====
- InnoDB should unfix the block after releases the latch.

- Add more assertion to check buffer fix while accessing the page.

- Introduced block_hint structure to store buf_block_t pointer
and allow accessing the buf_block_t pointer only by passing a
functor. It returns original buf_block_t* pointer if it is valid
or nullptr if the pointer become stale.

- Replace buf_block_is_uncompressed() with buf_pointer_is_block_field_instance()

This change is motivated by a change in mysql-5.6.50:
46e60de444a8fbd876cc6778a7e64a1d3426a48d
Bug #31036301 ASSERTION FAILURE: SYNC0RW.IC:429:LOCK->LOCK_WORD

18 files changed, 288 insertions, 155 deletions

diff --git a/storage/innobase/CMakeLists.txt b/storage/innobase/CMakeLists.txt
index 4513a63049d..10be3187072 100644
--- a/storage/innobase/CMakeLists.txt
+++ b/storage/innobase/CMakeLists.txt
@@ -28,6 +28,7 @@ SET(INNOBASE_SOURCES
 	btr/btr0scrub.cc
 	btr/btr0sea.cc
 	btr/btr0defragment.cc
+	buf/buf0block_hint.cc
 	buf/buf0buddy.cc
 	buf/buf0buf.cc
 	buf/buf0dblwr.cc
diff --git a/storage/innobase/btr/btr0bulk.cc b/storage/innobase/btr/btr0bulk.cc
index 3de8ee5c1c2..714426e40ee 100644
--- a/storage/innobase/btr/btr0bulk.cc
+++ b/storage/innobase/btr/btr0bulk.cc
@@ -698,6 +698,8 @@ PageBulk::latch()
 		m_index->set_modified(m_mtr);
 	}
 
+	ut_ad(m_block->page.buf_fix_count > 0);
+
 	/* In case the block is S-latched by page_cleaner. */
 	if (!buf_page_optimistic_get(RW_X_LATCH, m_block, m_modify_clock,
 				     __FILE__, __LINE__, &m_mtr)) {
@@ -716,6 +718,8 @@ PageBulk::latch()
 
 	buf_block_buf_fix_dec(m_block);
 
+	ut_ad(m_block->page.buf_fix_count > 0);
+
 	ut_ad(m_cur_rec > m_page && m_cur_rec < m_heap_top);
 
 	return (m_err);
diff --git a/storage/innobase/btr/btr0cur.cc b/storage/innobase/btr/btr0cur.cc
index a682148a482..3fcc3ab9ce6 100644
--- a/storage/innobase/btr/btr0cur.cc
+++ b/storage/innobase/btr/btr0cur.cc
@@ -639,6 +639,8 @@ btr_cur_optimistic_latch_leaves(
 	ulint		mode;
 	ulint		left_page_no;
 	ulint		curr_page_no;
+	ut_ad(block->page.buf_fix_count > 0);
+	ut_ad(buf_block_get_state(block) == BUF_BLOCK_FILE_PAGE);
 
 	switch (*latch_mode) {
 	case BTR_SEARCH_LEAF:
@@ -650,20 +652,10 @@ btr_cur_optimistic_latch_leaves(
 		mode = *latch_mode == BTR_SEARCH_PREV
 			? RW_S_LATCH : RW_X_LATCH;
 
-		buf_page_mutex_enter(block);
-		if (buf_block_get_state(block) != BUF_BLOCK_FILE_PAGE) {
-			buf_page_mutex_exit(block);
-			return(false);
-		}
-		/* pin the block not to be relocated */
-		buf_block_buf_fix_inc(block, file, line);
-		buf_page_mutex_exit(block);
-
 		rw_lock_s_lock(&block->lock);
 		if (block->modify_clock != modify_clock) {
 			rw_lock_s_unlock(&block->lock);
-
-			goto unpin_failed;
+			return false;
 		}
 
 		curr_page_no = block->page.id.page_no();
@@ -690,7 +682,7 @@ btr_cur_optimistic_latch_leaves(
 				/* release the left block */
 				btr_leaf_page_release(
 					cursor->left_block, mode, mtr);
-				goto unpin_failed;
+				return false;
 			}
 		} else {
 			cursor->left_block = NULL;
@@ -700,23 +692,28 @@ btr_cur_optimistic_latch_leaves(
 					    file, line, mtr)) {
 			if (btr_page_get_prev(buf_block_get_frame(block))
 			    == left_page_no) {
-				buf_block_buf_fix_dec(block);
+				/* block was already buffer-fixed while
+				entering the function and
+				buf_page_optimistic_get() buffer-fixes
+				it again. */
+				ut_ad(2 <= block->page.buf_fix_count);
 				*latch_mode = mode;
 				return(true);
 			} else {
-				/* release the block */
+				/* release the block and decrement of
+				buf_fix_count which was incremented
+				in buf_page_optimistic_get() */
 				btr_leaf_page_release(block, mode, mtr);
 			}
 		}
 
+		ut_ad(0 < block->page.buf_fix_count);
 		/* release the left block */
 		if (cursor->left_block != NULL) {
 			btr_leaf_page_release(cursor->left_block,
 					      mode, mtr);
 		}
-unpin_failed:
-		/* unpin the block */
-		buf_block_buf_fix_dec(block);
+
 		return(false);
 
 	default:
@@ -1281,12 +1278,7 @@ btr_cur_search_to_nth_level_func(
 	guess = NULL;
 #else
 	info = btr_search_get_info(index);
-
-	if (!buf_pool_is_obsolete(info->withdraw_clock)) {
-		guess = info->root_guess;
-	} else {
-		guess = NULL;
-	}
+	guess = info->root_guess;
 
 #ifdef BTR_CUR_HASH_ADAPT
 
@@ -1722,10 +1714,7 @@ retry_page_get:
 		}
 
 #ifdef BTR_CUR_ADAPT
-		if (block != guess) {
-			info->root_guess = block;
-			info->withdraw_clock = buf_withdraw_clock;
-		}
+		info->root_guess = block;
 #endif
 	}
 
diff --git a/storage/innobase/btr/btr0pcur.cc b/storage/innobase/btr/btr0pcur.cc
index 4f06251d0bf..220898e939e 100644
--- a/storage/innobase/btr/btr0pcur.cc
+++ b/storage/innobase/btr/btr0pcur.cc
@@ -174,11 +174,10 @@ before_first:
 		index, rec, &cursor->old_n_fields,
 		&cursor->old_rec_buf, &cursor->buf_size);
 
-	cursor->block_when_stored = block;
+	cursor->block_when_stored.store(block);
 
 	/* Function try to check if block is S/X latch. */
 	cursor->modify_clock = buf_block_get_modify_clock(block);
-	cursor->withdraw_clock = buf_withdraw_clock;
 }
 
 /**************************************************************//**
@@ -208,6 +207,24 @@ btr_pcur_copy_stored_position(
 	pcur_receive->old_n_fields = pcur_donate->old_n_fields;
 }
 
+/** Structure acts as functor to do the latching of leaf pages.
+It returns true if latching of leaf pages succeeded and false
+otherwise. */
+struct Btr_cur_optimistic_latch_leaves_functor_t{
+        btr_pcur_t * &cursor;
+        ulint &latch_mode;
+        const char *file;
+        ulint line;
+        mtr_t *&mtr;
+
+	bool operator() (buf_block_t *hint) const
+	{
+		return hint != NULL && btr_cur_optimistic_latch_leaves(
+			hint, cursor->modify_clock, &latch_mode,
+			btr_pcur_get_btr_cur(cursor), file, line, mtr);
+	}
+};
+
 /**************************************************************//**
 Restores the stored position of a persistent cursor bufferfixing the page and
 obtaining the specified latches. If the cursor position was saved when the
@@ -270,7 +287,7 @@ btr_pcur_restore_position_func(
 		cursor->latch_mode =
 			BTR_LATCH_MODE_WITHOUT_INTENTION(latch_mode);
 		cursor->pos_state = BTR_PCUR_IS_POSITIONED;
-		cursor->block_when_stored = btr_pcur_get_block(cursor);
+		cursor->block_when_stored.clear();
 
 		return(FALSE);
 	}
@@ -285,11 +302,9 @@ btr_pcur_restore_position_func(
 	case BTR_MODIFY_PREV:
 		/* Try optimistic restoration. */
 
-		if (!buf_pool_is_obsolete(cursor->withdraw_clock)
-		    && btr_cur_optimistic_latch_leaves(
-			cursor->block_when_stored, cursor->modify_clock,
-			&latch_mode, btr_pcur_get_btr_cur(cursor),
-			file, line, mtr)) {
+		Btr_cur_optimistic_latch_leaves_functor_t functor=
+			{cursor,latch_mode,file,line,mtr};
+		if (cursor->block_when_stored.run_with_hint(functor)) {
 
 			cursor->pos_state = BTR_PCUR_IS_POSITIONED;
 			cursor->latch_mode = latch_mode;
@@ -388,11 +403,10 @@ btr_pcur_restore_position_func(
 		since the cursor can now be on a different page!
 		But we can retain the value of old_rec */
 
-		cursor->block_when_stored = btr_pcur_get_block(cursor);
+		cursor->block_when_stored.store(btr_pcur_get_block(cursor));
 		cursor->modify_clock = buf_block_get_modify_clock(
-						cursor->block_when_stored);
+					cursor->block_when_stored.block());
 		cursor->old_stored = true;
-		cursor->withdraw_clock = buf_withdraw_clock;
 
 		mem_heap_free(heap);
 
diff --git a/storage/innobase/buf/buf0block_hint.cc b/storage/innobase/buf/buf0block_hint.cc
new file mode 100644
index 00000000000..9ccdfc9f8f3
--- /dev/null
+++ b/storage/innobase/buf/buf0block_hint.cc
@@ -0,0 +1,91 @@
+/*****************************************************************************
+
+Copyright (c) 2020, Oracle and/or its affiliates. All Rights Reserved.
+Copyright (c) 2020, MariaDB Corporation.
+
+This program is free software; you can redistribute it and/or modify it under
+the terms of the GNU General Public License, version 2.0, as published by the
+Free Software Foundation.
+
+This program is also distributed with certain software (including but not
+limited to OpenSSL) that is licensed under separate terms, as designated in a
+particular file or component or in included license documentation. The authors
+of MySQL hereby grant you an additional permission to link the program and
+your derivative works with the separately licensed software that they have
+included with MySQL.
+
+This program is distributed in the hope that it will be useful, but WITHOUT
+ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
+FOR A PARTICULAR PURPOSE. See the GNU General Public License, version 2.0,
+for more details.
+
+You should have received a copy of the GNU General Public License along with
+this program; if not, write to the Free Software Foundation, Inc.,
+51 Franklin St, Fifth Floor, Boston, MA 02110-1301  USA
+
+*****************************************************************************/
+
+#include "buf0block_hint.h"
+#include "buf0buf.h"
+namespace buf {
+
+void Block_hint::store(buf_block_t *block) {
+  ut_ad(block->page.buf_fix_count > 0);
+  m_block = block;
+  m_page_id.copy_from(block->page.id);
+}
+
+void Block_hint::clear() { m_block = NULL; }
+
+void Block_hint::buffer_fix_block_if_still_valid() {
+  /* We need to check if m_block points to one of chunks. For this to be
+  meaningful we need to prevent freeing memory while we check, and until we
+  buffer-fix the block. For this purpose it is enough to latch any of the many
+  latches taken by buf_resize().
+  However, for buffer-fixing to be meaningful, the block has to contain a page
+  (as opposed to being already empty, which might mean that buf_pool_resize()
+  can proceed and free it once we free the s-latch), so we confirm that the
+  block contains a page. However, it is not sufficient to check that this is
+  just any page, because just after we check it could get freed, unless we
+  have a latch which prevents this. This is tricky because page_hash latches
+  are sharded by page_id and we don't know the page_id until we look into the
+  block. To solve this chicken-and-egg problem somewhat, we latch the shard
+  for the m_page_id and compare block->page.id to it - so if is equal then we
+  can be reasonably sure that we have the correct latch.
+  There is still a theoretical problem here, where other threads might try
+  to modify the m_block->page.id while we are comparing it, but the chance of
+  accidentally causing the old space_id == m_page_id.m_space and the new
+  page_no == m_page_id.m_page_no is minimal as compilers emit a single 8-byte
+  comparison instruction to compare both at the same time atomically, and f()
+    will probably double-check the block->page.id again, anyway.
+  Finally, assuming that we have correct hash bucket latched, we should check if
+  the state of the block is BUF_BLOCK_FILE_PAGE before buffer-fixing the block,
+  as otherwise we risk buffer-fixing and operating on a block, which is already
+  meant to be freed. In particular, buf_LRU_free_page() first calls
+  buf_LRU_block_remove_hashed() under hash bucket latch protection to change the
+  state to BUF_BLOCK_REMOVE_HASH and then releases the latch. Later it calls
+  buf_LRU_block_free_hashed_page() without any latch to change the state to
+  BUF_BLOCK_MEMORY and reset the page's id, which means buf_resize() can free it
+  regardless of our buffer-fixing. */
+  if (m_block != NULL) {
+    const buf_pool_t *const pool = buf_pool_get(m_page_id);
+    rw_lock_t *latch = buf_page_hash_lock_get(pool, m_page_id);
+    rw_lock_s_lock(latch);
+    /* If not own buf_pool_mutex, page_hash can be changed. */
+    latch = buf_page_hash_lock_s_confirm(latch, pool, m_page_id);
+    if (buf_pointer_is_block_field_instance(pool, m_block) &&
+        m_page_id == m_block->page.id &&
+        buf_block_get_state(m_block) == BUF_BLOCK_FILE_PAGE) {
+      buf_block_buf_fix_inc(m_block, __FILE__, __LINE__);
+    } else {
+      clear();
+    }
+    rw_lock_s_unlock(latch);
+  }
+}
+void Block_hint::buffer_unfix_block_if_needed(buf_block_t *block) {
+  if (block != NULL) {
+    buf_block_buf_fix_dec(block);
+  }
+}
+}  // namespace buf
diff --git a/storage/innobase/buf/buf0buf.cc b/storage/innobase/buf/buf0buf.cc
index 5e7c49becc2..ebf61ac9131 100644
--- a/storage/innobase/buf/buf0buf.cc
+++ b/storage/innobase/buf/buf0buf.cc
@@ -339,14 +339,6 @@ buf_pool_t*	buf_pool_ptr;
 /** true when resizing buffer pool is in the critical path. */
 volatile bool	buf_pool_resizing;
 
-/** true when withdrawing buffer pool pages might cause page relocation */
-volatile bool	buf_pool_withdrawing;
-
-/** the clock is incremented every time a pointer to a page may become obsolete;
-if the withdrwa clock has not changed, the pointer is still valid in buffer
-pool. if changed, the pointer might not be in buffer pool any more. */
-volatile ulint	buf_withdraw_clock;
-
 /** Map of buffer pool chunks by its first frame address
 This is newly made by initialization of buffer pool and buf_resize_thread.
 Currently, no need mutex protection for update. */
@@ -2110,8 +2102,6 @@ buf_pool_init(
 	NUMA_MEMPOLICY_INTERLEAVE_IN_SCOPE;
 
 	buf_pool_resizing = false;
-	buf_pool_withdrawing = false;
-	buf_withdraw_clock = 0;
 
 	buf_pool_ptr = (buf_pool_t*) ut_zalloc_nokey(
 		n_instances * sizeof *buf_pool_ptr);
@@ -2171,7 +2161,6 @@ buf_page_realloc(
 {
 	buf_block_t*	new_block;
 
-	ut_ad(buf_pool_withdrawing);
 	ut_ad(buf_pool_mutex_own(buf_pool));
 	ut_ad(buf_block_get_state(block) == BUF_BLOCK_FILE_PAGE);
 
@@ -2593,9 +2582,6 @@ buf_pool_withdraw_blocks(
 	ib::info() << "buffer pool " << i << " : withdrawn target "
 		<< UT_LIST_GET_LEN(buf_pool->withdraw) << " blocks.";
 
-	/* retry is not needed */
-	++buf_withdraw_clock;
-
 	return(false);
 }
 
@@ -2692,7 +2678,6 @@ buf_pool_resize()
 	NUMA_MEMPOLICY_INTERLEAVE_IN_SCOPE;
 
 	ut_ad(!buf_pool_resizing);
-	ut_ad(!buf_pool_withdrawing);
 	ut_ad(srv_buf_pool_chunk_unit > 0);
 
 	new_instance_size = srv_buf_pool_size / srv_buf_pool_instances;
@@ -2760,7 +2745,6 @@ buf_pool_resize()
 
 			ut_ad(buf_pool->withdraw_target == 0);
 			buf_pool->withdraw_target = withdraw_target;
-			buf_pool_withdrawing = true;
 		}
 	}
 
@@ -2785,7 +2769,6 @@ withdraw_retry:
 
 	if (srv_shutdown_state != SRV_SHUTDOWN_NONE) {
 		/* abort to resize for shutdown. */
-		buf_pool_withdrawing = false;
 		return;
 	}
 
@@ -2847,7 +2830,6 @@ withdraw_retry:
 		goto withdraw_retry;
 	}
 
-	buf_pool_withdrawing = false;
 
 	buf_resize_status("Latching whole of buffer pool.");
 
@@ -4044,35 +4026,26 @@ buf_block_from_ahi(const byte* ptr)
 }
 #endif /* BTR_CUR_HASH_ADAPT */
 
-/********************************************************************//**
-Find out if a pointer belongs to a buf_block_t. It can be a pointer to
-the buf_block_t itself or a member of it. This functions checks one of
-the buffer pool instances.
-@return TRUE if ptr belongs to a buf_block_t struct */
-static
 ibool
-buf_pointer_is_block_field_instance(
-/*================================*/
-	buf_pool_t*	buf_pool,	/*!< in: buffer pool instance */
-	const void*	ptr)		/*!< in: pointer not dereferenced */
+buf_pointer_is_block_field_instance(const buf_pool_t* buf_pool,
+                                    const void*	ptr)
 {
-	const buf_chunk_t*		chunk	= buf_pool->chunks;
-	const buf_chunk_t* const	echunk	= chunk + ut_min(
-		buf_pool->n_chunks, buf_pool->n_chunks_new);
-
-	/* TODO: protect buf_pool->chunks with a mutex (the older pointer will
-	currently remain while during buf_pool_resize()) */
-	while (chunk < echunk) {
-		if (ptr >= (void*) chunk->blocks
-		    && ptr < (void*) (chunk->blocks + chunk->size)) {
+  const buf_chunk_t* chunk= buf_pool->chunks;
+  const buf_chunk_t* const echunk= chunk + ut_min(
+    buf_pool->n_chunks, buf_pool->n_chunks_new);
 
-			return(TRUE);
-		}
+  /* TODO: protect buf_pool->chunks with a mutex (the older
+  pointer will currently remain while during buf_pool_resize()) */
+  while (chunk < echunk)
+  {
+    if (ptr >= (void*) chunk->blocks &&
+        ptr < (void*) (chunk->blocks + chunk->size))
+      return TRUE;
 
-		chunk++;
-	}
+    chunk++;
+  }
 
-	return(FALSE);
+  return FALSE;
 }
 
 /********************************************************************//**
@@ -4099,25 +4072,6 @@ buf_pointer_is_block_field(
 	return(FALSE);
 }
 
-/********************************************************************//**
-Find out if a buffer block was created by buf_chunk_init().
-@return TRUE if "block" has been added to buf_pool->free by buf_chunk_init() */
-static
-ibool
-buf_block_is_uncompressed(
-/*======================*/
-	buf_pool_t*		buf_pool,	/*!< in: buffer pool instance */
-	const buf_block_t*	block)		/*!< in: pointer to block,
-						not dereferenced */
-{
-	if ((((ulint) block) % sizeof *block) != 0) {
-		/* The pointer should be aligned. */
-		return(FALSE);
-	}
-
-	return(buf_pointer_is_block_field_instance(buf_pool, (void*) block));
-}
-
 #if defined UNIV_DEBUG || defined UNIV_IBUF_DEBUG
 /********************************************************************//**
 Return true if probe is enabled.
@@ -4356,7 +4310,7 @@ loop:
 		has been allocated by buf_page_alloc_descriptor(),
 		it may have been freed by buf_relocate(). */
 
-		if (!buf_block_is_uncompressed(buf_pool, block)
+		if (!buf_pointer_is_block_field_instance(buf_pool, block)
 		    || page_id != block->page.id
 		    || buf_block_get_state(block) != BUF_BLOCK_FILE_PAGE) {
 
diff --git a/storage/innobase/gis/gis0sea.cc b/storage/innobase/gis/gis0sea.cc
index 65079260aae..0d8384b74fa 100644
--- a/storage/innobase/gis/gis0sea.cc
+++ b/storage/innobase/gis/gis0sea.cc
@@ -1256,6 +1256,22 @@ rtr_check_discard_page(
 	lock_mutex_exit();
 }
 
+/** Structure acts as functor to get the optimistic access of the page.
+It returns true if it successfully gets the page. */ 
+struct Buf_page_optimistic_get_functor_t
+{
+  btr_pcur_t * &r_cursor;
+  const char *file;
+  ulint line;
+  mtr_t * &mtr;
+
+  bool operator()(buf_block_t *hint) const
+  {
+    return hint != NULL && buf_page_optimistic_get(
+       RW_X_LATCH, hint, r_cursor->modify_clock, file, line, mtr);
+  }
+};
+
 /** Restore the stored position of a persistent cursor bufferfixing the page */
 static
 bool
@@ -1289,11 +1305,9 @@ rtr_cur_restore_position(
 
 	ut_ad(latch_mode == BTR_CONT_MODIFY_TREE);
 
-	if (!buf_pool_is_obsolete(r_cursor->withdraw_clock)
-	    && buf_page_optimistic_get(RW_X_LATCH,
-				       r_cursor->block_when_stored,
-				       r_cursor->modify_clock,
-				       __FILE__, __LINE__, mtr)) {
+	Buf_page_optimistic_get_functor_t functor=
+		{r_cursor,__FILE__, __LINE__, mtr};
+	if (r_cursor->block_when_stored.run_with_hint(functor)) {
 		ut_ad(r_cursor->pos_state == BTR_PCUR_IS_POSITIONED);
 
 		ut_ad(r_cursor->rel_pos == BTR_PCUR_ON);
diff --git a/storage/innobase/include/btr0pcur.h b/storage/innobase/include/btr0pcur.h
index c20b971de98..38960b1d15c 100644
--- a/storage/innobase/include/btr0pcur.h
+++ b/storage/innobase/include/btr0pcur.h
@@ -29,6 +29,7 @@ Created 2/23/1996 Heikki Tuuri
 
 #include "dict0dict.h"
 #include "btr0cur.h"
+#include "buf0block_hint.h"
 #include "btr0btr.h"
 #include "gis0rtree.h"
 
@@ -502,13 +503,10 @@ struct btr_pcur_t{
 	whether cursor was on, before, or after the old_rec record */
 	enum btr_pcur_pos_t	rel_pos;
 	/** buffer block when the position was stored */
-	buf_block_t*	block_when_stored;
+	buf::Block_hint		block_when_stored;
 	/** the modify clock value of the buffer block when the cursor position
 	was stored */
 	ib_uint64_t	modify_clock;
-	/** the withdraw clock value of the buffer pool when the cursor
-	position was stored */
-	ulint		withdraw_clock;
 	/** btr_pcur_store_position() and btr_pcur_restore_position() state. */
 	enum pcur_pos_t	pos_state;
 	/** PAGE_CUR_G, ... */
@@ -528,9 +526,8 @@ struct btr_pcur_t{
 	btr_pcur_t() :
 		btr_cur(), latch_mode(0), old_stored(false), old_rec(NULL),
 		old_n_fields(0), rel_pos(btr_pcur_pos_t(0)),
-		block_when_stored(NULL),
-		modify_clock(0), withdraw_clock(0),
-		pos_state(BTR_PCUR_NOT_POSITIONED),
+		block_when_stored(),
+		modify_clock(0), pos_state(BTR_PCUR_NOT_POSITIONED),
 		search_mode(PAGE_CUR_UNSUPP), trx_if_known(NULL),
 		old_rec_buf(NULL), buf_size(0)
 	{
diff --git a/storage/innobase/include/btr0sea.h b/storage/innobase/include/btr0sea.h
index 2fa9aaa38fe..adb14a7c16f 100644
--- a/storage/innobase/include/btr0sea.h
+++ b/storage/innobase/include/btr0sea.h
@@ -206,8 +206,6 @@ struct btr_search_t{
 	the machine word, i.e., they cannot be turned into bit-fields. */
 	buf_block_t* root_guess;/*!< the root page frame when it was last time
 				fetched, or NULL */
-	ulint	withdraw_clock;	/*!< the withdraw clock value of the buffer
-				pool when root_guess was stored */
 #ifdef BTR_CUR_HASH_ADAPT
 	ulint	hash_analysis;	/*!< when this exceeds
 				BTR_SEARCH_HASH_ANALYSIS, the hash
diff --git a/storage/innobase/include/buf0block_hint.h b/storage/innobase/include/buf0block_hint.h
new file mode 100644
index 00000000000..b1a3209ef5b
--- /dev/null
+++ b/storage/innobase/include/buf0block_hint.h
@@ -0,0 +1,88 @@
+/*****************************************************************************
+
+Copyright (c) 2020, Oracle and/or its affiliates. All Rights Reserved.
+Copyright (c) 2020, MariaDB Corporation.
+This program is free software; you can redistribute it and/or modify it under
+the terms of the GNU General Public License, version 2.0, as published by the
+Free Software Foundation.
+
+This program is also distributed with certain software (including but not
+limited to OpenSSL) that is licensed under separate terms, as designated in a
+particular file or component or in included license documentation. The authors
+of MySQL hereby grant you an additional permission to link the program and
+your derivative works with the separately licensed software that they have
+included with MySQL.
+
+This program is distributed in the hope that it will be useful, but WITHOUT
+ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
+FOR A PARTICULAR PURPOSE. See the GNU General Public License, version 2.0,
+for more details.
+
+You should have received a copy of the GNU General Public License along with
+this program; if not, write to the Free Software Foundation, Inc.,
+51 Franklin St, Fifth Floor, Boston, MA 02110-1301  USA
+
+*****************************************************************************/
+#ifndef buf0block_hint_h
+#define buf0block_hint_h
+#include "buf0buf.h"
+
+namespace buf {
+class Block_hint {
+ public:
+  Block_hint():m_block(NULL),m_page_id(0,0){}
+  /** Stores the pointer to the block, which is currently buffer-fixed.
+  @param[in]  block   a pointer to a buffer-fixed block to be stored */
+  void store(buf_block_t *block);
+
+  /** Clears currently stored pointer. */
+  void clear();
+
+  /** Executes given function with the block pointer which was previously stored
+  or with nullptr if the pointer is no longer valid, was cleared or not stored.
+  @param[in]  f   The function to be executed. It will be passed the pointer.
+                  If you wish to use the block pointer subsequently, you need to
+                  ensure you buffer-fix it before returning from f.
+  @return the return value of f
+  */
+  template <typename F>
+  bool run_with_hint(const F &f) {
+    buffer_fix_block_if_still_valid();
+/* m_block could be changed during f() call, so we use local variable to
+    remember which block we need to unfix */
+    buf_block_t *buffer_fixed_block = m_block;
+    bool res = f(buffer_fixed_block);
+    buffer_unfix_block_if_needed(buffer_fixed_block);
+    return res;
+  }
+
+  Block_hint &operator=(const Block_hint&other){
+    m_block=other.m_block;
+    m_page_id.copy_from(other.m_page_id);
+    return *this;
+  }
+
+  /** Return block */
+  buf_block_t* block()
+  {
+    return m_block;
+  }
+
+ private:
+  /** The block pointer stored by store(). */
+  buf_block_t *m_block;
+  /** If m_block is non-null, the m_block->page.id at time it was stored. */
+  page_id_t m_page_id;
+
+  /** A helper function which checks if m_block is not a dangling pointer and
+  still points to block with page with m_page_id and if so, buffer-fixes it,
+  otherwise clear()s it */
+  void buffer_fix_block_if_still_valid();
+
+  /** A helper function which decrements block->buf_fix_count if it's non-null
+  @param[in]  block   A pointer to a block or nullptr */
+  static void buffer_unfix_block_if_needed(buf_block_t *block);
+};
+
+}  // namespace buf
+#endif /* buf0hint_h*/
diff --git a/storage/innobase/include/buf0buf.h b/storage/innobase/include/buf0buf.h
index bc14be6a3c5..a614bdb0100 100644
--- a/storage/innobase/include/buf0buf.h
+++ b/storage/innobase/include/buf0buf.h
@@ -101,10 +101,6 @@ extern	buf_pool_t*	buf_pool_ptr;	/*!< The buffer pools
 extern	volatile bool	buf_pool_withdrawing; /*!< true when withdrawing buffer
 					pool pages might cause page relocation */
 
-extern	volatile ulint	buf_withdraw_clock; /*!< the clock is incremented
-					every time a pointer to a page may
-					become obsolete */
-
 # ifdef UNIV_DEBUG
 extern my_bool	buf_disable_resize_buffer_pool_debug; /*!< if TRUE, resizing
 					buffer pool is not allowed. */
@@ -1159,6 +1155,16 @@ buf_pointer_is_block_field(
 #define buf_pool_is_block_lock(l)			\
 	buf_pointer_is_block_field((const void*)(l))
 
+/** Find out if a pointer belongs to a buf_block_t. It can be a
+pointer to the buf_block_t itself or a member of it. This functions
+checks one of the buffer pool instances.
+@param[in] buf_pool  buffer pool instance
+@param[in] ptr       pointer not dereferenced
+@return TRUE if ptr belongs to a buf_block_t struct */
+ibool
+buf_pointer_is_block_field_instance(const buf_pool_t *buf_pool,
+                                    const void *ptr);
+
 /** Initialize a page for read to the buffer buf_pool. If the page is
 (1) already in buf_pool, or
 (2) if we specify to read only ibuf pages and the page is not an ibuf page, or
@@ -1373,14 +1379,6 @@ buf_get_nth_chunk_block(
 	ulint		n,		/*!< in: nth chunk in the buffer pool */
 	ulint*		chunk_size);	/*!< in: chunk size */
 
-/** Verify the possibility that a stored page is not in buffer pool.
-@param[in]	withdraw_clock	withdraw clock when stored the page
-@retval true	if the page might be relocated */
-UNIV_INLINE
-bool
-buf_pool_is_obsolete(
-	ulint	withdraw_clock);
-
 /** Calculate aligned buffer pool size based on srv_buf_pool_chunk_unit,
 if needed.
 @param[in]	size	size in bytes
diff --git a/storage/innobase/include/buf0buf.ic b/storage/innobase/include/buf0buf.ic
index f331091a1d7..de780fc506c 100644
--- a/storage/innobase/include/buf0buf.ic
+++ b/storage/innobase/include/buf0buf.ic
@@ -1056,8 +1056,6 @@ buf_block_buf_fix_dec(
 /*==================*/
 	buf_block_t*	block)	/*!< in/out: block to bufferunfix */
 {
-	buf_block_unfix(block);
-
 #ifdef UNIV_DEBUG
 	/* No debug latch is acquired if block belongs to system temporary.
 	Debug latch is not of much help if access to block is single
@@ -1066,6 +1064,8 @@ buf_block_buf_fix_dec(
 		rw_lock_s_unlock(&block->debug_latch);
 	}
 #endif /* UNIV_DEBUG */
+
+	buf_block_unfix(block);
 }
 
 /** Returns the buffer pool instance given a page id.
@@ -1439,18 +1439,6 @@ buf_page_get_frame(
 	}
 }
 
-/** Verify the possibility that a stored page is not in buffer pool.
-@param[in]	withdraw_clock	withdraw clock when stored the page
-@retval true	if the page might be relocated */
-UNIV_INLINE
-bool
-buf_pool_is_obsolete(
-	ulint	withdraw_clock)
-{
-	return(UNIV_UNLIKELY(buf_pool_withdrawing
-			     || buf_withdraw_clock != withdraw_clock));
-}
-
 /** Calculate aligned buffer pool size based on srv_buf_pool_chunk_unit,
 if needed.
 @param[in]	size	size in bytes
diff --git a/storage/innobase/include/buf0types.h b/storage/innobase/include/buf0types.h
index bd5e26df47b..c27d4a4dc23 100644
--- a/storage/innobase/include/buf0types.h
+++ b/storage/innobase/include/buf0types.h
@@ -181,6 +181,11 @@ public:
 		m_space = m_page_no = ULINT32_UNDEFINED;
 	}
 
+	inline void copy_from(const page_id_t& src)
+	{
+		m_space = src.space();
+		m_page_no = src.page_no();
+	}
 private:
 
 	/** Tablespace id. */
diff --git a/storage/innobase/include/mtr0mtr.ic b/storage/innobase/include/mtr0mtr.ic
index 845b2ff625f..17b7f04a29d 100644
--- a/storage/innobase/include/mtr0mtr.ic
+++ b/storage/innobase/include/mtr0mtr.ic
@@ -170,10 +170,10 @@ mtr_t::release_block_at_savepoint(
 
 	ut_a(slot->object == block);
 
-	buf_block_unfix(reinterpret_cast<buf_block_t*>(block));
-
 	buf_page_release_latch(block, slot->type);
 
+	buf_block_unfix(reinterpret_cast<buf_block_t*>(block));
+
 	slot->object = NULL;
 }
 
diff --git a/storage/innobase/include/trx0undo.h b/storage/innobase/include/trx0undo.h
index 7be4314ecbc..4ebf269917b 100644
--- a/storage/innobase/include/trx0undo.h
+++ b/storage/innobase/include/trx0undo.h
@@ -366,9 +366,6 @@ struct trx_undo_t {
 					(IB_ID_MAX if the undo log is empty) */
 	buf_block_t*	guess_block;	/*!< guess for the buffer block where
 					the top page might reside */
-	ulint		withdraw_clock;	/*!< the withdraw clock value of the
-					buffer pool when guess_block was stored */
-
 	/** @return whether the undo log is empty */
 	bool empty() const { return top_undo_no == IB_ID_MAX; }
 
diff --git a/storage/innobase/mtr/mtr0mtr.cc b/storage/innobase/mtr/mtr0mtr.cc
index 1749c582f2c..ef6dc58e749 100644
--- a/storage/innobase/mtr/mtr0mtr.cc
+++ b/storage/innobase/mtr/mtr0mtr.cc
@@ -233,8 +233,8 @@ static void memo_slot_release(mtr_memo_slot_t *slot)
   case MTR_MEMO_PAGE_SX_FIX:
   case MTR_MEMO_PAGE_X_FIX:
     buf_block_t *block= reinterpret_cast<buf_block_t*>(slot->object);
-    buf_block_unfix(block);
     buf_page_release_latch(block, slot->type);
+    buf_block_unfix(block);
     break;
   }
   slot->object= NULL;
@@ -276,8 +276,8 @@ struct ReleaseLatches {
     case MTR_MEMO_PAGE_SX_FIX:
     case MTR_MEMO_PAGE_X_FIX:
       buf_block_t *block= reinterpret_cast<buf_block_t*>(slot->object);
-      buf_block_unfix(block);
       buf_page_release_latch(block, slot->type);
+      buf_block_unfix(block);
       break;
     }
     slot->object= NULL;
diff --git a/storage/innobase/trx/trx0rec.cc b/storage/innobase/trx/trx0rec.cc
index c7be4f44ee0..259f1a927a0 100644
--- a/storage/innobase/trx/trx0rec.cc
+++ b/storage/innobase/trx/trx0rec.cc
@@ -1915,7 +1915,6 @@ dberr_t trx_undo_report_rename(trx_t* trx, const dict_table_t* table)
 
 			if (ulint offset = trx_undo_page_report_rename(
 				    trx, table, block, &mtr)) {
-				undo->withdraw_clock = buf_withdraw_clock;
 				undo->top_page_no = undo->last_page_no;
 				undo->top_offset  = offset;
 				undo->top_undo_no = trx->undo_no++;
@@ -2055,13 +2054,12 @@ trx_undo_report_row_operation(
 			mtr_commit(&mtr);
 		} else {
 			/* Success */
-			undo->withdraw_clock = buf_withdraw_clock;
+			undo->guess_block = undo_block;
 			mtr_commit(&mtr);
 
 			undo->top_page_no = undo_block->page.id.page_no();
 			undo->top_offset  = offset;
 			undo->top_undo_no = trx->undo_no++;
-			undo->guess_block = undo_block;
 			ut_ad(!undo->empty());
 
 			if (!is_temp) {
diff --git a/storage/innobase/trx/trx0undo.cc b/storage/innobase/trx/trx0undo.cc
index e43902006d0..b4955ca43a3 100644
--- a/storage/innobase/trx/trx0undo.cc
+++ b/storage/innobase/trx/trx0undo.cc
@@ -1213,7 +1213,6 @@ trx_undo_mem_create(
 	undo->top_undo_no = IB_ID_MAX;
 	undo->top_page_no = page_no;
 	undo->guess_block = NULL;
-	undo->withdraw_clock = 0;
 	ut_ad(undo->empty());
 
 	return(undo);
@@ -1403,8 +1402,7 @@ trx_undo_assign(trx_t* trx, dberr_t* err, mtr_t* mtr)
 		return buf_page_get_gen(
 			page_id_t(undo->rseg->space->id, undo->last_page_no),
 			univ_page_size, RW_X_LATCH,
-			buf_pool_is_obsolete(undo->withdraw_clock)
-			? NULL : undo->guess_block,
+			undo->guess_block,
 			BUF_GET, __FILE__, __LINE__, mtr, err);
 	}
 
@@ -1459,8 +1457,7 @@ trx_undo_assign_low(trx_t* trx, trx_rseg_t* rseg, trx_undo_t** undo,
 		return buf_page_get_gen(
 			page_id_t(rseg->space->id, (*undo)->last_page_no),
 			univ_page_size, RW_X_LATCH,
-			buf_pool_is_obsolete((*undo)->withdraw_clock)
-			? NULL : (*undo)->guess_block,
+			(*undo)->guess_block,
 			BUF_GET, __FILE__, __LINE__, mtr, err);
 	}