MDEV-32820 Race condition between trx_purge_free_segment() and trx_un…

…do_create()

trx_purge_free_segment(): If fseg_free_step_not_header() needs to be
called multiple times, acquire an exclusive latch on the
rollback segment header page after restarting the mini-transaction
so that the rest of this function cannot execute concurrently
with trx_undo_create() on the same rollback segment.

This fixes a regression that was introduced in
commit c14a394 (MDEV-30753).

Note: The buffer-fixes that we are holding across the mini-transaction
restart will prevent the pages from being evicted from the buffer pool.
They may be accessed by other threads or written back to data files
while we are not holding exclusive latches.

Reviewed by: Vladislav Lesin

storage/innobase/trx/trx0purge.cc

@@ -365,8 +365,10 @@ void trx_purge_free_segment(mtr_t &mtr, trx_rseg_t* rseg, fil_addr_t hdr_addr)
   while (!fseg_free_step_not_header(TRX_UNDO_SEG_HDR + TRX_UNDO_FSEG_HEADER +
                                     block->frame, &mtr))
   {
-    block->fix();
-    const page_id_t id{block->page.id()};
+    buf_block_buf_fix_inc(rseg_hdr, __FILE__, __LINE__);
+    buf_block_buf_fix_inc(block, __FILE__, __LINE__);
+    ut_d(const page_id_t rseg_hdr_id{rseg_hdr->page.id()});
+    ut_d(const page_id_t id{block->page.id()});
     mtr.commit();
     /* NOTE: If the server is killed after the log that was produced
     up to this point was written, and before the log from the mtr.commit()
@@ -376,19 +378,12 @@ void trx_purge_free_segment(mtr_t &mtr, trx_rseg_t* rseg, fil_addr_t hdr_addr)
     This does not matter when using multiple innodb_undo_tablespaces;
     innodb_undo_log_truncate=ON will be able to reclaim the space. */
     mtr.start();
-    ut_ad(rw_lock_s_lock_nowait(block->debug_latch, __FILE__, __LINE__));
+    rw_lock_x_lock(&rseg_hdr->lock);
     rw_lock_x_lock(&block->lock);
-    if (UNIV_UNLIKELY(block->page.id() != id))
-    {
-      block->unfix();
-      rw_lock_x_unlock(&block->lock);
-      ut_d(rw_lock_s_unlock(block->debug_latch));
-      block= buf_page_get(id, 0, RW_X_LATCH, &mtr);
-      if (!block)
-        return;
-    }
-    else
-      mtr_memo_push(&mtr, block, MTR_MEMO_PAGE_X_FIX);
+    ut_ad(rseg_hdr->page.id() == rseg_hdr_id);
+    ut_ad(block->page.id() == id);
+    mtr_memo_push(&mtr, rseg_hdr, MTR_MEMO_PAGE_X_FIX);
+    mtr_memo_push(&mtr, block, MTR_MEMO_PAGE_X_FIX);
   }
 
   while (!fseg_free_step(TRX_UNDO_SEG_HDR + TRX_UNDO_FSEG_HEADER +