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;