MDEV-24188: Merge 10.4 into 10.5

storage/innobase/buf/buf0buf.cc

@@ -3772,18 +3772,29 @@ buf_page_create(fil_space_t *space, uint32_t offset,
       ut_ad(0);
       break;
     case BUF_BLOCK_FILE_PAGE:
-      buf_block_buf_fix_inc(block, __FILE__, __LINE__);
+      if (!mtr->have_x_latch(*block))
       {
-        const auto num_fix_count= mtr->get_fix_count(block) + 1;
-        while (block->page.io_fix() != BUF_IO_NONE ||
-               num_fix_count != block->page.buf_fix_count())
+        buf_block_buf_fix_inc(block, __FILE__, __LINE__);
         {
-          mysql_mutex_unlock(&buf_pool.mutex);
-          os_thread_yield();
-          mysql_mutex_lock(&buf_pool.mutex);
+          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);
+          }
         }
+        rw_lock_x_lock(&block->lock);
+        mtr_memo_push(mtr, block, MTR_MEMO_PAGE_X_FIX);
+      }
+      else
+      {
+        ut_ad(!block->page.ibuf_exist);
+#ifdef BTR_CUR_HASH_ADAPT
+        ut_ad(!block->index);
+#endif
       }
-      rw_lock_x_lock(&block->lock);
 #ifdef BTR_CUR_HASH_ADAPT
       drop_hash_entry= block->index;
 #endif
@@ -3808,6 +3819,7 @@ buf_page_create(fil_space_t *space, uint32_t offset,
       buf_page_free_descriptor(&block->page);
       block= free_block;
       buf_block_buf_fix_inc(block, __FILE__, __LINE__);
+      mtr_memo_push(mtr, block, MTR_MEMO_PAGE_X_FIX);
       break;
     }
 
@@ -3818,8 +3830,6 @@ buf_page_create(fil_space_t *space, uint32_t offset,
       btr_search_drop_page_hash_index(block);
 #endif /* BTR_CUR_HASH_ADAPT */
 
-    mtr_memo_push(mtr, block, MTR_MEMO_PAGE_X_FIX);
-
     if (block->page.ibuf_exist)
     {
       if (!recv_recovery_is_on())

storage/innobase/include/dyn0buf.h

@@ -330,6 +330,24 @@ class mtr_buf_t {
 		return(true);
 	}
 
+	/**
+	Iterate over each block and call the functor.
+	@return	false if iteration was terminated. */
+	template <typename Functor>
+	bool for_each_block(const Functor& functor) const
+	{
+		for (typename list_t::iterator it = m_list.begin(),
+					       end = m_list.end();
+		     it != end; ++it) {
+
+			if (!functor(&*it)) {
+				return false;
+			}
+		}
+
+		return(true);
+	}
+
 	/**
 	Iterate over all the blocks in reverse and call the iterator
 	@return	false if iteration was terminated. */

storage/innobase/include/mtr0mtr.h

@@ -151,6 +151,10 @@ struct mtr_t {
     return old_mode;
   }
 
+  /** Check if we are holding a block latch in exclusive mode
+  @param block  buffer pool block to search for */
+  bool have_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

storage/innobase/include/sync0rw.h

@@ -506,7 +506,7 @@ the pass value == 0. */
 bool
 rw_lock_own(
 /*========*/
-	rw_lock_t*	lock,		/*!< in: rw-lock */
+	const rw_lock_t*lock,		/*!< in: rw-lock */
 	ulint		lock_type)	/*!< in: lock type: RW_LOCK_S,
 					RW_LOCK_X */
 	MY_ATTRIBUTE((warn_unused_result));

storage/innobase/mtr/mtr0mtr.cc

@@ -893,29 +893,48 @@ inline std::pair<lsn_t,bool> mtr_t::finish_write(ulint len)
 	return std::make_pair(start_lsn, flush);
 }
 
-/** Find buffer fix count of the given block acquired by the
-mini-transaction */
-struct FindBlock
+/** Find out whether a block was X-latched by the mini-transaction */
+struct FindBlockX
 {
-  int32_t num_fix;
-  const buf_block_t *const block;
+  const buf_block_t &block;
 
-  FindBlock(const buf_block_t *block_buf): num_fix(0), block(block_buf) {}
+  FindBlockX(const buf_block_t &block): block(block) {}
 
-  bool operator()(const mtr_memo_slot_t* slot)
+  /** @return whether the block was not found x-latched */
+  bool operator()(const mtr_memo_slot_t *slot) const
   {
-    if (slot->object == block)
-      num_fix++;
-    return true;
+    return slot->object != &block || slot->type == MTR_MEMO_PAGE_X_FIX;
+  }
+};
+
+#ifdef UNIV_DEBUG
+/** Assert that the block is not present in the mini-transaction */
+struct FindNoBlock
+{
+  const buf_block_t &block;
+
+  FindNoBlock(const buf_block_t &block): block(block) {}
+
+  /** @return whether the block was not found */
+  bool operator()(const mtr_memo_slot_t *slot) const
+  {
+    return slot->object != &block;
   }
 };
+#endif /* UNIV_DEBUG */
 
-uint32_t mtr_t::get_fix_count(const buf_block_t *block) const
+bool mtr_t::have_x_latch(const buf_block_t &block) const
 {
-  Iterate<FindBlock> iteration((FindBlock(block)));
-  if (m_memo.for_each_block(iteration))
-    return iteration.functor.num_fix;
-  return 0;
+  if (m_memo.for_each_block(CIterate<FindBlockX>(FindBlockX(block))))
+  {
+    ut_ad(m_memo.for_each_block(CIterate<FindNoBlock>(FindNoBlock(block))));
+    ut_ad(!memo_contains_flagged(&block,
+                                 MTR_MEMO_PAGE_S_FIX | MTR_MEMO_PAGE_SX_FIX |
+                                 MTR_MEMO_BUF_FIX | MTR_MEMO_MODIFY));
+    return false;
+  }
+  ut_ad(rw_lock_own(&block.lock, RW_LOCK_X));
+  return true;
 }
 
 #ifdef UNIV_DEBUG
@@ -931,13 +950,13 @@ bool mtr_t::memo_contains(const rw_lock_t &lock, mtr_memo_type_t type)
 
   switch (type) {
   case MTR_MEMO_X_LOCK:
-    ut_ad(rw_lock_own(const_cast<rw_lock_t*>(&lock), RW_LOCK_X));
+    ut_ad(rw_lock_own(&lock, RW_LOCK_X));
     break;
   case MTR_MEMO_SX_LOCK:
-    ut_ad(rw_lock_own(const_cast<rw_lock_t*>(&lock), RW_LOCK_SX));
+    ut_ad(rw_lock_own(&lock, RW_LOCK_SX));
     break;
   case MTR_MEMO_S_LOCK:
-    ut_ad(rw_lock_own(const_cast<rw_lock_t*>(&lock), RW_LOCK_S));
+    ut_ad(rw_lock_own(&lock, RW_LOCK_S));
     break;
   default:
     break;

storage/innobase/sync/sync0rw.cc

@@ -990,7 +990,7 @@ the pass value == 0.
 bool
 rw_lock_own(
 /*========*/
-	rw_lock_t*	lock,		/*!< in: rw-lock */
+	const rw_lock_t*lock,		/*!< in: rw-lock */
 	ulint		lock_type)	/*!< in: lock type: RW_LOCK_S,
 					RW_LOCK_X */
 {