MDEV-27716 mtr_t::commit() acquires log_sys.mutex when writing no log

mtr_t::is_block_dirtied(), mtr_t::memo_push(): Never set m_made_dirty
for pages of the temporary tablespace. Ever since
commit 5eb539555b36a60944eefeb84d5d6d436ba61e63
we never add those pages to buf_pool.flush_list.

mtr_t::commit(): Implement part of mtr_t::prepare_write() here,
and avoid acquiring log_sys.mutex if no log is written.
During IMPORT TABLESPACE fixup, we do not write log, but we must
add pages to buf_pool.flush_list and for that, be prepared
to acquire log_sys.flush_order_mutex.

mtr_t::do_write(): Replaces mtr_t::prepare_write().

3 files changed, 31 insertions, 30 deletions

diff --git a/storage/innobase/include/mtr0mtr.h b/storage/innobase/include/mtr0mtr.h
index 3e72bf9acd7..3f9777ad225 100644
--- a/storage/innobase/include/mtr0mtr.h
+++ b/storage/innobase/include/mtr0mtr.h
@@ -623,9 +623,9 @@ private:
   @param type   extended record subtype; @see mrec_ext_t */
   inline void log_write_extended(const buf_block_t &block, byte type);
 
-  /** Prepare to write the mini-transaction log to the redo log buffer.
-  @return number of bytes to write in finish_write() */
-  inline ulint prepare_write();
+  /** Append the redo log records to the redo log buffer.
+  @return {start_lsn,flush_ahead} */
+  std::pair<lsn_t,page_flush_ahead> do_write();
 
   /** Append the redo log records to the redo log buffer.
   @param len   number of bytes to write
diff --git a/storage/innobase/include/mtr0mtr.inl b/storage/innobase/include/mtr0mtr.inl
index 48bdb9bb6d1..bc2986503f9 100644
--- a/storage/innobase/include/mtr0mtr.inl
+++ b/storage/innobase/include/mtr0mtr.inl
@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 1995, 2014, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2017, 2021, MariaDB Corporation.
+Copyright (c) 2017, 2022, MariaDB Corporation.
 
 This program is free software; you can redistribute it and/or modify it under
 the terms of the GNU General Public License as published by the Free Software
@@ -32,7 +32,8 @@ inline bool mtr_t::is_block_dirtied(const buf_block_t *block)
 {
   ut_ad(block->page.state() == BUF_BLOCK_FILE_PAGE);
   ut_ad(block->page.buf_fix_count());
-  return block->page.oldest_modification() <= 1;
+  return block->page.oldest_modification() <= 1 &&
+    block->page.id().space() < SRV_TMP_SPACE_ID;
 }
 
 /**
@@ -51,8 +52,8 @@ mtr_t::memo_push(void* object, mtr_memo_type_t type)
 	grab log_sys.flush_order_mutex at mtr_t::commit() so that we
 	can insert the dirtied page into the flush list. */
 
-	if ((type == MTR_MEMO_PAGE_X_FIX || type == MTR_MEMO_PAGE_SX_FIX)
-	    && !m_made_dirty) {
+	if (!m_made_dirty
+            && (type == MTR_MEMO_PAGE_X_FIX || type == MTR_MEMO_PAGE_SX_FIX)) {
 
 		m_made_dirty = is_block_dirtied(
 			reinterpret_cast<const buf_block_t*>(object));
diff --git a/storage/innobase/mtr/mtr0mtr.cc b/storage/innobase/mtr/mtr0mtr.cc
index a8dffa48c35..ca97a9e77e4 100644
--- a/storage/innobase/mtr/mtr0mtr.cc
+++ b/storage/innobase/mtr/mtr0mtr.cc
@@ -404,18 +404,27 @@ void mtr_t::commit()
 
     std::pair<lsn_t,page_flush_ahead> lsns;
 
-    if (const ulint len= prepare_write())
-      lsns= finish_write(len);
-    else
-      lsns= { m_commit_lsn, PAGE_FLUSH_NO };
+    if (UNIV_LIKELY(m_log_mode == MTR_LOG_ALL))
+    {
+      lsns= do_write();
 
-    if (m_made_dirty)
-      mysql_mutex_lock(&log_sys.flush_order_mutex);
+      if (m_made_dirty)
+        mysql_mutex_lock(&log_sys.flush_order_mutex);
 
-    /* It is now safe to release the log mutex because the
-    flush_order mutex will ensure that we are the first one
-    to insert into the flush list. */
-    mysql_mutex_unlock(&log_sys.mutex);
+      /* It is now safe to release log_sys.mutex because the
+      buf_pool.flush_order_mutex will ensure that we are the first one
+      to insert into buf_pool.flush_list. */
+      mysql_mutex_unlock(&log_sys.mutex);
+    }
+    else
+    {
+      ut_ad(m_log_mode == MTR_LOG_NO_REDO);
+      ut_ad(m_log.size() == 0);
+      m_commit_lsn= log_sys.get_lsn();
+      lsns= { m_commit_lsn, PAGE_FLUSH_NO };
+      if (UNIV_UNLIKELY(m_made_dirty)) /* This should be IMPORT TABLESPACE */
+        mysql_mutex_lock(&log_sys.flush_order_mutex);
+    }
 
     if (m_freed_pages)
     {
@@ -517,7 +526,7 @@ void mtr_t::commit_shrink(fil_space_t &space)
 
   log_write_and_flush_prepare();
 
-  const lsn_t start_lsn= finish_write(prepare_write()).first;
+  const lsn_t start_lsn= do_write().first;
 
   mysql_mutex_lock(&log_sys.flush_order_mutex);
   /* Durably write the reduced FSP_SIZE before truncating the data file. */
@@ -924,19 +933,10 @@ struct mtr_write_log
   }
 };
 
-/** Prepare to write the mini-transaction log to the redo log buffer.
-@return number of bytes to write in finish_write() */
-inline ulint mtr_t::prepare_write()
+std::pair<lsn_t,mtr_t::page_flush_ahead> mtr_t::do_write()
 {
 	ut_ad(!recv_no_log_write);
-
-	if (UNIV_UNLIKELY(m_log_mode != MTR_LOG_ALL)) {
-		ut_ad(m_log_mode == MTR_LOG_NO_REDO);
-		ut_ad(m_log.size() == 0);
-		mysql_mutex_lock(&log_sys.mutex);
-		m_commit_lsn = log_sys.get_lsn();
-		return 0;
-	}
+	ut_ad(m_log_mode == MTR_LOG_ALL);
 
 	ulint	len	= m_log.size();
 	ut_ad(len > 0);
@@ -968,7 +968,7 @@ inline ulint mtr_t::prepare_write()
 	/* check and attempt a checkpoint if exceeding capacity */
 	log_margin_checkpoint_age(len);
 
-	return(len);
+	return finish_write(len);
 }
 
 /** Append the redo log records to the redo log buffer.