MDEV-30216 Read-ahead unnecessarily allocates and frees pages when a page is in the buffer pool

buf_pool_t::page_hash_contains(): Check if a page is cached.

buf_read_ahead_random(), buf_read_page_background(),
buf_read_ahead_linear(): Before invoking buf_read_page_low(),
preallocate a buffer page for the read request.

buf_read_page(), buf_page_init_for_read(), buf_read_page_low():
Add a parameter for the buf_pool.page_hash chain, to avoid duplicated
computations.

buf_page_t::read_complete(): Only attempt recovery if an uncompressed
page frame has been allocated.

buf_page_init_for_read(): Before trying to acquire buf_pool.mutex, acquire
an exclusive buf_pool.page_hash latch and check if the page is already
located in the buffer pool. If the buf_pool.mutex is not immediately
available, release both latches and acquire them in the correct order,
and then recheck if the page is already in the buffer pool. This should
hopefully reduce some contention on buf_pool.mutex.

buf_page_init_for_read(), buf_read_page_low(): Input the "recovery needed"
flag in the least significant bit of zip_size.

buf_read_acquire(), buf_read_release(): Interface for allocating and
freeing buffer pages for reading.

buf_read_recv_pages(): Set the flag that recovery is needed.
Other ROW_FORMAT=COMPRESSED reads during recovery
will not need any recovery.

storage/innobase/buf/buf0buf.cc

@@ -2095,7 +2095,7 @@ buf_page_t* buf_page_get_zip(const page_id_t page_id, ulint zip_size)
   return bpage;
 
 must_read_page:
-  if (dberr_t err= buf_read_page(page_id, zip_size))
+  if (dberr_t err= buf_read_page(page_id, zip_size, chain))
   {
     ib::error() << "Reading compressed page " << page_id
                 << " failed with error: " << err;
@@ -2319,7 +2319,7 @@ buf_page_get_low(
 	corrupted, or if an encrypted page with a valid
 	checksum cannot be decypted. */
 
-	if (dberr_t local_err = buf_read_page(page_id, zip_size)) {
+	if (dberr_t local_err = buf_read_page(page_id, zip_size, chain)) {
 		if (local_err != DB_CORRUPTION
 		    && mode != BUF_GET_POSSIBLY_FREED
 		    && retries++ < BUF_PAGE_READ_MAX_RETRIES) {
@@ -3235,18 +3235,16 @@ dberr_t buf_page_t::read_complete(const fil_node_t &node)
                  << FORCE_RECOVERY_MSG;
     }
 
-    if (!srv_force_recovery)
-      goto release_page;
-  }
-
-  if (err == DB_PAGE_CORRUPTED || err == DB_DECRYPTION_FAILED)
-  {
+    if (err == DB_PAGE_CORRUPTED || err == DB_DECRYPTION_FAILED ||
+        !srv_force_recovery)
+    {
 release_page:
-    buf_pool.corrupted_evict(this, buf_page_t::READ_FIX);
-    return err;
+      buf_pool.corrupted_evict(this, buf_page_t::READ_FIX);
+      return err;
+    }
   }
 
-  const bool recovery= recv_recovery_is_on();
+  const bool recovery= frame && recv_recovery_is_on();
 
   if (recovery && !recv_recover_page(node.space, this))
     return DB_PAGE_CORRUPTED;

storage/innobase/include/buf0buf.h

@@ -1385,6 +1385,12 @@ class buf_pool_t
   }
 
 public:
+  /** @return whether the buffer pool contains a page
+  @param page_id       page identifier
+  @param chain         hash table chain for page_id.fold() */
+  TRANSACTIONAL_TARGET
+  bool page_hash_contains(const page_id_t page_id, hash_chain &chain);
+
   /** @return whether less than 1/4 of the buffer pool is available */
   TPOOL_SUPPRESS_TSAN
   bool running_out() const

storage/innobase/include/buf0rea.h

@@ -32,14 +32,16 @@ Created 11/5/1995 Heikki Tuuri
 buffer buf_pool if it is not already there. Sets the io_fix flag and sets
 an exclusive lock on the buffer frame. The flag is cleared and the x-lock
 released by the i/o-handler thread.
-@param[in]	page_id		page id
-@param[in]	zip_size	ROW_FORMAT=COMPRESSED page size, or 0
+@param page_id    page id
+@param zip_size   ROW_FORMAT=COMPRESSED page size, or 0
+@param chain      buf_pool.page_hash cell for page_id
 @retval DB_SUCCESS if the page was read and is not corrupted,
 @retval DB_PAGE_CORRUPTED if page based on checksum check is corrupted,
 @retval DB_DECRYPTION_FAILED if page post encryption checksum matches but
 after decryption normal page checksum does not match.
 @retval DB_TABLESPACE_DELETED if tablespace .ibd file is missing */
-dberr_t buf_read_page(const page_id_t page_id, ulint zip_size);
+dberr_t buf_read_page(const page_id_t page_id, ulint zip_size,
+                      buf_pool_t::hash_chain &chain);
 
 /** High-level function which reads a page asynchronously from a file to the
 buffer buf_pool if it is not already there. Sets the io_fix flag and sets