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);
 	}