MDEV-24188 fixup: Simplify the wait loop

Starting with commit 7cffb5f (MDEV-23399)
the function buf_flush_page() will first acquire block->lock and only
after that invoke set_io_fix(). Before that, it was possible to reach
a livelock between buf_page_create() and buf_flush_page().

buf_page_create(): Directly try acquiring the exclusive page latch
without checking whether the page is io-fixed or buffer-fixed.
(As a matter of fact, the have_x_latch() check is not strictly necessary,
because we still support recursive X-latches.)
In case of a latch conflict, wait while allowing buf_page_write_complete()
to acquire buf_pool.mutex and release the block->lock.

An attempt to wait for exclusive block->lock while holding buf_pool.mutex
would lead to a hang in the tests parts.part_supported_sql_func_innodb
and stress.ddl_innodb, due to a deadlock between buf_page_write_complete()
and buf_page_create().

Similarly, in case of an I/O fixed compressed-only
ROW_FORMAT=COMPRESSED page, we will sleep before retrying.

In both cases, we will sleep for 1ms or until a flush batch is completed.

Author: dr-m

storage/innobase/buf/buf0buf.cc

@@ -3755,9 +3755,9 @@ buf_page_create(fil_space_t *space, uint32_t offset,
   free_block->initialise(page_id, zip_size, 1);
 
   const ulint fold= page_id.fold();
-loop:
   mysql_mutex_lock(&buf_pool.mutex);
 
+loop:
   buf_block_t *block= reinterpret_cast<buf_block_t*>
     (buf_pool.page_hash_get_low(page_id, fold));
 
@@ -3775,17 +3775,15 @@ buf_page_create(fil_space_t *space, uint32_t offset,
       if (!mtr->have_x_latch(*block))
       {
         buf_block_buf_fix_inc(block, __FILE__, __LINE__);
+        while (!rw_lock_x_lock_nowait(&block->lock))
         {
-          while (block->page.io_fix() != BUF_IO_NONE ||
-                 block->page.buf_fix_count() != 1)
-          {
-            timespec abstime;
-            set_timespec_nsec(abstime, 1000000);
-            mysql_cond_timedwait(&buf_pool.done_flush_list, &buf_pool.mutex,
-                                 &abstime);
-          }
+          /* Wait for buf_page_write_complete() to release block->lock.
+          We must not hold buf_pool.mutex while waiting. */
+          timespec abstime;
+          set_timespec_nsec(abstime, 1000000);
+          mysql_cond_timedwait(&buf_pool.done_flush_list, &buf_pool.mutex,
+                               &abstime);
         }
-        rw_lock_x_lock(&block->lock);
         mtr_memo_push(mtr, block, MTR_MEMO_PAGE_X_FIX);
       }
       else
@@ -3805,7 +3803,11 @@ buf_page_create(fil_space_t *space, uint32_t offset,
       if (block->page.io_fix() != BUF_IO_NONE)
       {
         hash_lock->write_unlock();
-        mysql_mutex_unlock(&buf_pool.mutex);
+        /* Wait for buf_page_write_complete() to release the I/O fix. */
+        timespec abstime;
+        set_timespec_nsec(abstime, 1000000);
+        mysql_cond_timedwait(&buf_pool.done_flush_list, &buf_pool.mutex,
+                             &abstime);
         goto loop;
       }