MDEV-29336: Potential deadlock in btr_page_alloc_low() with the AHI

The index root page contains the fields BTR_SEG_TOP and BTR_SEG_LEAF
which keep track of allocated pages in the index tree. These fields
are normally protected by an Update latch, so that concurrent read
access to other parts of the page will be possible.

When the index root page is already exclusively latched in the
mini-transaction, we must not try to acquire a lower-grade Update latch.
In fact, when the root page is already X or U latched in the
mini-transaction, there is no point to acquire another latch.
Moreover, after a U latch was acquired on top of an X-latch,
mtr_t::defer_drop_ahi() would trigger an assertion failure or
lock corruption in block->page.lock.u_x_upgrade() because X locks
already exist on the block.

This problem may have been introduced in
commit 03ca6495 (MDEV-24142).

btr_page_alloc_low(), btr_page_free(): Initially buffer-fix the root page.
If it is already U or X latched, release the buffer-fix. Else, upgrade
the buffer-fix to a U latch.

mtr_t::u_lock_register(): Upgrade a buffer-fix to U latch.

mtr_t::have_u_or_x_latch(): Check if U or X latches are already
registered in the mini-transaction.

storage/innobase/btr/btr0btr.cc

@@ -513,9 +513,28 @@ btr_page_alloc_low(
 					page should be initialized. */
 	dberr_t*	err)		/*!< out: error code */
 {
-  buf_block_t *root= btr_root_block_get(index, RW_SX_LATCH, mtr, err);
+  const auto savepoint= mtr->get_savepoint();
+  buf_block_t *root= btr_root_block_get(index, RW_NO_LATCH, mtr, err);
   if (UNIV_UNLIKELY(!root))
     return root;
+
+  if (mtr->have_u_or_x_latch(*root))
+  {
+#ifdef BTR_CUR_HASH_ADAPT
+    ut_ad(!root->index || !root->index->freed());
+#endif
+    mtr->release_block_at_savepoint(savepoint, root);
+  }
+  else
+  {
+    mtr->u_lock_register(savepoint);
+    root->page.lock.u_lock();
+#ifdef BTR_CUR_HASH_ADAPT
+    if (root->index)
+      mtr_t::defer_drop_ahi(root, MTR_MEMO_PAGE_SX_FIX);
+#endif
+  }
+
   fseg_header_t *seg_header= root->page.frame +
     (level ? PAGE_HEADER + PAGE_BTR_SEG_TOP : PAGE_HEADER + PAGE_BTR_SEG_LEAF);
   return fseg_alloc_free_page_general(seg_header, hint_page_no, file_direction,
@@ -610,11 +629,30 @@ dberr_t btr_page_free(dict_index_t* index, buf_block_t* block, mtr_t* mtr,
 
   fil_space_t *space= index->table->space;
   dberr_t err;
-  if (page_t* root = btr_root_get(index, mtr, &err))
+
+  const auto savepoint= mtr->get_savepoint();
+  if (buf_block_t *root= btr_root_block_get(index, RW_NO_LATCH, mtr, &err))
   {
-    err= fseg_free_page(&root[blob || page_is_leaf(block->page.frame)
-                              ? PAGE_HEADER + PAGE_BTR_SEG_LEAF
-                              : PAGE_HEADER + PAGE_BTR_SEG_TOP],
+    if (mtr->have_u_or_x_latch(*root))
+    {
+#ifdef BTR_CUR_HASH_ADAPT
+      ut_ad(!root->index || !root->index->freed());
+#endif
+      mtr->release_block_at_savepoint(savepoint, root);
+    }
+    else
+    {
+      mtr->u_lock_register(savepoint);
+      root->page.lock.u_lock();
+#ifdef BTR_CUR_HASH_ADAPT
+      if (root->index)
+        mtr_t::defer_drop_ahi(root, MTR_MEMO_PAGE_SX_FIX);
+#endif
+    }
+    err= fseg_free_page(&root->page.frame[blob ||
+                                          page_is_leaf(block->page.frame)
+                                          ? PAGE_HEADER + PAGE_BTR_SEG_LEAF
+                                          : PAGE_HEADER + PAGE_BTR_SEG_TOP],
                         space, page, mtr, space_latched);
   }
   if (err == DB_SUCCESS)

storage/innobase/include/mtr0mtr.h

@@ -177,6 +177,10 @@ struct mtr_t {
   @param block  buffer pool block to search for */
   bool have_x_latch(const buf_block_t &block) const;
 
+  /** Check if we are holding a block latch in S or U mode
+  @param block  buffer pool block to search for */
+  bool have_u_or_x_latch(const buf_block_t &block) const;
+
 	/** Copy the tablespaces associated with the mini-transaction
 	(needed for generating FILE_MODIFY records)
 	@param[in]	mtr	mini-transaction that may modify
@@ -336,6 +340,15 @@ struct mtr_t {
   @param rw_latch RW_S_LATCH, RW_SX_LATCH, RW_X_LATCH, RW_NO_LATCH */
   void page_lock(buf_block_t *block, ulint rw_latch);
 
+  /** Register a page latch on a buffer-fixed block was buffer-fixed.
+  @param latch   latch type */
+  void u_lock_register(ulint savepoint)
+  {
+    mtr_memo_slot_t *slot= m_memo.at<mtr_memo_slot_t*>(savepoint);
+    ut_ad(slot->type == MTR_MEMO_BUF_FIX);
+    slot->type= MTR_MEMO_PAGE_SX_FIX;
+  }
+
   /** Upgrade U locks on a block to X */
   void page_lock_upgrade(const buf_block_t &block);
   /** Upgrade X lock to X */

storage/innobase/mtr/mtr0mtr.cc

@@ -1220,6 +1220,21 @@ struct FindBlockX
   }
 };
 
+/** Find out whether a block was not X or U latched by the mini-transaction */
+struct FindBlockUX
+{
+  const buf_block_t █
+
+  FindBlockUX(const buf_block_t &block): block(block) {}
+
+  /** @return whether the block was not found x-latched */
+  bool operator()(const mtr_memo_slot_t *slot) const
+  {
+    return slot->object != &block ||
+      !(slot->type & (MTR_MEMO_PAGE_X_FIX | MTR_MEMO_PAGE_SX_FIX));
+  }
+};
+
 #ifdef UNIV_DEBUG
 /** Assert that the block is not present in the mini-transaction */
 struct FindNoBlock
@@ -1250,6 +1265,14 @@ bool mtr_t::have_x_latch(const buf_block_t &block) const
   return true;
 }
 
+bool mtr_t::have_u_or_x_latch(const buf_block_t &block) const
+{
+  if (m_memo.for_each_block(CIterate<FindBlockUX>(FindBlockUX(block))))
+    return false;
+  ut_ad(block.page.lock.have_u_or_x());
+  return true;
+}
+
 /** Check if we are holding exclusive tablespace latch
 @param space  tablespace to search for
 @param shared whether to look for shared latch, instead of exclusive