MDEV-28845 InnoDB: Failing assertion: bpage->can_relocate() in buf0lru.cc

Since commit 0b47c126 (MDEV-13542)
we treat all-zero pages as corrupted ones.

During a stress test, a read-ahead of an all-zero page was triggered
and the page read was completed concurrently with buf_page_create_low().
This caused the assertion to fail, because buf_page_create_low() was
waiting for the page latch.

buf_page_get_low(): Only invoke buf_pool_t::corrupted_evict()
if the block was not already marked as corrupted.

buf_page_create_low(): On page identifier mismatch, retry the
buf_pool.page_hash lookup.

buf_pool_t::corrupted_evict(): Set the state of the block to FREED
so that a concurrent buf_page_get_low() will refuse to load the page.
Wait for the page latch to be vacant before proceeding to remove
the block from buf_pool.page_hash and buf_pool.LRU.

page_id_t::set_corrupted(), page_id_t::is_corrupted(): Accessors
for indicating a corrupted page identifier.

Tested by Matthias Leich

storage/innobase/buf/buf0buf.cc

@@ -2872,7 +2872,9 @@ buf_page_get_low(
 				*err = e;
 			}
 
+			if (block->page.id().is_corrupted()) {
 				buf_pool.corrupted_evict(&block->page, state);
+			}
 			return nullptr;
 		}
 
@@ -3210,6 +3212,7 @@ static buf_block_t *buf_page_create_low(page_id_t page_id, ulint zip_size,
   free_block->initialise(page_id, zip_size, buf_page_t::MEMORY);
 
   buf_pool_t::hash_chain &chain= buf_pool.page_hash.cell_get(page_id.fold());
+retry:
   mysql_mutex_lock(&buf_pool.mutex);
 
   buf_page_t *bpage= buf_pool.page_hash.get(page_id, chain);
@@ -3228,6 +3231,12 @@ static buf_block_t *buf_page_create_low(page_id_t page_id, ulint zip_size,
       {
         mysql_mutex_unlock(&buf_pool.mutex);
         bpage->lock.x_lock();
+        const page_id_t id{bpage->id()};
+        if (UNIV_UNLIKELY(id != page_id))
+        {
+          ut_ad(id.is_corrupted());
+          goto retry;
+        }
         mysql_mutex_lock(&buf_pool.mutex);
       }
 

storage/innobase/buf/buf0lru.cc

@@ -1218,14 +1218,14 @@ void buf_pool_t::corrupted_evict(buf_page_t *bpage, uint32_t state)
 
   ut_ad(!bpage->oldest_modification());
   bpage->set_corrupt_id();
-  auto unfix= state - buf_page_t::UNFIXED;
+  auto unfix= state - buf_page_t::FREED;
   auto s= bpage->zip.fix.fetch_sub(unfix) - unfix;
   bpage->lock.x_unlock(true);
 
-  while (s != buf_page_t::UNFIXED)
+  while (s != buf_page_t::FREED || bpage->lock.is_locked_or_waiting())
   {
-    ut_ad(s > buf_page_t::UNFIXED);
-    ut_ad(s < buf_page_t::READ_FIX);
+    ut_ad(s >= buf_page_t::FREED);
+    ut_ad(s < buf_page_t::UNFIXED);
     /* Wait for other threads to release the fix count
     before releasing the bpage from LRU list. */
     (void) LF_BACKOFF();

storage/innobase/include/buf0buf.h

@@ -2011,7 +2011,7 @@ inline void buf_page_t::set_corrupt_id()
           is_write_locked());
   }
 #endif
-  id_= page_id_t(~0ULL);
+  id_.set_corrupted();
 }
 
 /** Set oldest_modification when adding to buf_pool.flush_list */

storage/innobase/include/buf0types.h

@@ -148,6 +148,12 @@ class page_id_t
 
   constexpr ulonglong raw() const { return m_id; }
 
+  /** Flag the page identifier as corrupted. */
+  void set_corrupted() { m_id= ~0ULL; }
+
+  /** @return whether the page identifier belongs to a corrupted page */
+  constexpr bool is_corrupted() const { return m_id == ~0ULL; }
+
 private:
   /** The page identifier */
   uint64_t m_id;