MDEV-24188: Merge 10.2 into 10.3

storage/innobase/buf/buf0buf.cc

@@ -5573,6 +5573,7 @@ loop:
 	    && !buf_pool_watch_is_sentinel(buf_pool, &block->page)) {
 		ut_d(block->page.file_page_was_freed = FALSE);
 		buf_page_state page_state = buf_block_get_state(block);
+		bool have_x_latch = false;
 #ifdef BTR_CUR_HASH_ADAPT
 		const dict_index_t *drop_hash_entry= NULL;
 #endif
@@ -5625,26 +5626,26 @@ loop:
 			free_block = NULL;
 			break;
 		case BUF_BLOCK_FILE_PAGE:
+			have_x_latch = mtr->have_x_latch(*block);
+			if (!have_x_latch) {
 				buf_block_fix(block);
-			const int32_t num_fix_count =
-				mtr->get_fix_count(block) + 1;
 				buf_page_mutex_enter(block);
-			while (buf_block_get_io_fix(block) != BUF_IO_NONE
-			       || (num_fix_count
-				   != block->page.buf_fix_count)) {
+				while (buf_block_get_io_fix(block)
+				       != BUF_IO_NONE
+				       || block->page.buf_fix_count != 1) {
 					buf_page_mutex_exit(block);
 					buf_pool_mutex_exit(buf_pool);
 					rw_lock_x_unlock(hash_lock);
 
-				os_thread_yield();
+					os_thread_sleep(1000);
 
 					buf_pool_mutex_enter(buf_pool);
 					rw_lock_x_lock(hash_lock);
 					buf_page_mutex_enter(block);
 				}
-
 				rw_lock_x_lock(&block->lock);
 				buf_page_mutex_exit(block);
+			}
 #ifdef BTR_CUR_HASH_ADAPT
 			drop_hash_entry = block->index;
 #endif
@@ -5663,6 +5664,7 @@ loop:
 		}
 #endif /* BTR_CUR_HASH_ADAPT */
 
+		if (!have_x_latch) {
 #ifdef UNIV_DEBUG
 			if (!fsp_is_system_temporary(page_id.space())) {
 				rw_lock_s_lock_nowait(
@@ -5672,7 +5674,7 @@ loop:
 #endif /* UNIV_DEBUG */
 
 			mtr_memo_push(mtr, block, MTR_MEMO_PAGE_X_FIX);
-
+		}
 		return block;
 	}
 

storage/innobase/include/dyn0buf.h

@@ -330,6 +330,24 @@ public:
 		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

@@ -437,10 +437,9 @@ struct mtr_t {
 	static inline bool is_block_dirtied(const buf_block_t* block)
 		MY_ATTRIBUTE((warn_unused_result));
 
-	/** Get the buffer fix count for the block added by this mtr.
-	@param[in]	block	block to be checked
-	@return number of buffer count added by this mtr */
-	int32_t get_fix_count(const buf_block_t *block);
+	/** 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;
 private:
 	/** Prepare to write the mini-transaction log to the redo log buffer.
 	@return number of bytes to write in finish_write() */

storage/innobase/include/sync0rw.h

@@ -508,7 +508,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

@@ -308,24 +308,6 @@ struct DebugCheck {
 };
 #endif
 
-/** Find buffer fix count of the given block acquired by the
-mini-transaction */
-struct FindBlock
-{
-  int32_t num_fix;
-  const buf_block_t *const block;
-
-  FindBlock(const buf_block_t *block_buf): num_fix(0), block(block_buf) {}
-
-  bool operator()(const mtr_memo_slot_t* slot)
-  {
-    if (slot->object == block)
-      ut_d(if (slot->type != MTR_MEMO_MODIFY))
-      num_fix++;
-    return true;
-  }
-};
-
 /** Release a resource acquired by the mini-transaction. */
 struct ReleaseBlocks {
 	/** Release specific object */
@@ -754,12 +736,48 @@ inline lsn_t mtr_t::finish_write(ulint len)
 	return start_lsn;
 }
 
-int32_t mtr_t::get_fix_count(const buf_block_t *block)
+/** Find out whether a block was X-latched by the mini-transaction */
+struct FindBlockX
 {
-  Iterate<FindBlock> iteration((FindBlock(block)));
-  if (m_memo.for_each_block(iteration))
-    return iteration.functor.num_fix;
-  return 0;
+  const buf_block_t &block;
+
+  FindBlockX(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;
+  }
+};
+
+#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 */
+
+bool mtr_t::have_x_latch(const buf_block_t &block) const
+{
+  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
@@ -776,15 +794,17 @@ mtr_t::memo_contains(
 		return(false);
 	}
 
+	const rw_lock_t *lock = static_cast<const rw_lock_t*>(object);
+
 	switch (type) {
 	case MTR_MEMO_X_LOCK:
-		ut_ad(rw_lock_own((rw_lock_t*) object, RW_LOCK_X));
+		ut_ad(rw_lock_own(lock, RW_LOCK_X));
 		break;
 	case MTR_MEMO_SX_LOCK:
-		ut_ad(rw_lock_own((rw_lock_t*) object, RW_LOCK_SX));
+		ut_ad(rw_lock_own(lock, RW_LOCK_SX));
 		break;
 	case MTR_MEMO_S_LOCK:
-		ut_ad(rw_lock_own((rw_lock_t*) object, RW_LOCK_S));
+		ut_ad(rw_lock_own(lock, RW_LOCK_S));
 		break;
 	}
 

storage/innobase/sync/sync0rw.cc

@@ -1000,7 +1000,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 */
 {