MDEV-31343 Another server hang with innodb_undo_log_truncate=ON

trx_purge_truncate_history(): While waiting for a write-fixed block
to become available, simply wait for an exclusive latch on it.
Also, simplify the iteration: first check for oldest_modification>2
(to ignore clean pages or pages belonging to the temporary tablespace)
and then compare the tablespace identifier.

Before releasing buf_pool.flush_list_mutex we will buffer-fix the block
of interest. In that way, buf_page_t::can_relocate() will not hold on
the block and it must remain in the buffer pool until we have acquired
an exclusive latch on it. If the block is still dirty, we will register
it with the tablespace truncation mini-transaction; else, we will simply
release the latch and buffer-fix and move to the next block.

This also reverts commit c4d79399
because that fix should no longer be necessary; the wait for an
exclusive block latch should allow buf_pool_t::release_freed_page()
on the same block to proceed.

Tested by: Axel Schwenke, Matthias Leich

storage/innobase/trx/trx0purge.cc

@@ -685,6 +685,7 @@ TRANSACTIONAL_TARGET static void trx_purge_truncate_history()
     mtr_t mtr;
     mtr.start();
     mtr.x_lock_space(&space);
+    const auto space_id= space.id;
 
     /* Lock all modified pages of the tablespace.
 
@@ -694,8 +695,8 @@ TRANSACTIONAL_TARGET static void trx_purge_truncate_history()
     mini-transaction commit and the server was killed, then
     discarding the to-be-trimmed pages without flushing would
     break crash recovery. */
+  rescan:
     mysql_mutex_lock(&buf_pool.flush_list_mutex);
-
     for (buf_page_t *bpage= UT_LIST_GET_LAST(buf_pool.flush_list); bpage; )
     {
       ut_ad(bpage->oldest_modification());
@@ -703,46 +704,47 @@ TRANSACTIONAL_TARGET static void trx_purge_truncate_history()
 
       buf_page_t *prev= UT_LIST_GET_PREV(list, bpage);
 
-      if (bpage->id().space() == space.id &&
-          bpage->oldest_modification() != 1)
+      if (bpage->oldest_modification() > 2 && bpage->id().space() == space_id)
       {
         ut_ad(bpage->frame);
-        auto block= reinterpret_cast<buf_block_t*>(bpage);
-        if (!bpage->lock.x_lock_try())
+        bpage->fix();
         {
-        rescan:
-          /* Let buf_pool_t::release_freed_page() proceed. */
+          /* Try to acquire an exclusive latch while the cache line is
+          fresh after fix(). */
+          const bool got_lock{bpage->lock.x_lock_try()};
+          buf_pool.flush_hp.set(prev);
           mysql_mutex_unlock(&buf_pool.flush_list_mutex);
-          mysql_mutex_lock(&buf_pool.mutex);
-          mysql_mutex_lock(&buf_pool.flush_list_mutex);
-          mysql_mutex_unlock(&buf_pool.mutex);
-          bpage= UT_LIST_GET_LAST(buf_pool.flush_list);
-          continue;
+          if (!got_lock)
+            bpage->lock.x_lock();
         }
-        buf_pool.flush_hp.set(prev);
-        mysql_mutex_unlock(&buf_pool.flush_list_mutex);
 
 #ifdef BTR_CUR_HASH_ADAPT
-        ut_ad(!block->index); /* There is no AHI on undo tablespaces. */
+        /* There is no AHI on undo tablespaces. */
+        ut_ad(!reinterpret_cast<buf_block_t*>(bpage)->index);
 #endif
-        bpage->fix();
         ut_ad(!bpage->is_io_fixed());
-        mysql_mutex_lock(&buf_pool.flush_list_mutex);
+        ut_ad(bpage->id().space() == space_id);
 
-        if (bpage->oldest_modification() > 1)
+        if (bpage->oldest_modification() > 2)
         {
+          mtr.memo_push(reinterpret_cast<buf_block_t*>(bpage),
+                        MTR_MEMO_PAGE_X_FIX);
+          mysql_mutex_lock(&buf_pool.flush_list_mutex);
+          ut_ad(bpage->oldest_modification() > 2);
           bpage->reset_oldest_modification();
-          mtr.memo_push(block, MTR_MEMO_PAGE_X_FIX);
         }
         else
         {
           bpage->unfix();
           bpage->lock.x_unlock();
+          mysql_mutex_lock(&buf_pool.flush_list_mutex);
         }
 
         if (prev != buf_pool.flush_hp.get())
-          /* Rescan, because we may have lost the position. */
+        {
+          mysql_mutex_unlock(&buf_pool.flush_list_mutex);
           goto rescan;
+        }
       }
 
       bpage= prev;