MDEV-28708 Increased congestion on buf_pool.flush_list_mutex

In commit f80deb9 (MDEV-27868)
a fix for a correctness regression caused a performance regression
by increasing the amount of work that is executed while holding
buf_pool.flush_list_mutex.

buf_page_t::set_temp_modified(): Relax an assertion, to allow an
already dirty block to be marked as dirty.

buf_page_t::flush_list_requests: Note that the variable is not
always protected by buf_pool.flush_list_mutex. Already dirty blocks
that are being written to will increment the counter without
holding buf_pool.flush_list_mutex.

mtr_t::process_freed_pages(): Handle pages that were freed during
the execution of the mini-transaction.

ReleaseUnlogged, mtr_t::release_unlogged(): Release modified pages when
no log was written. This is for pages of the temporary tablespace, or for
IMPORT TABLESPACE.

ReleaseModified: Renamed from ReleaseBlocks.
Assume that buf_pool.flush_list_mutex was acquired by the caller.

ReleaseSimple: A combination of ReleaseLatches and ReleaseModified,
for the case that for any modified pages, some earlier modifications
are already waiting to be written.

mtr_t::commit(): Invoke one of release_unlogged(), ReleaseModified,
ReleaseSimple, ReleaseAll. Acquire and release buf_pool.flush_list_mutex
at most once.

memo_slot_release(): Simplify the code.

mtr_t::sx_latch_at_savepoint(), mtr_t::x_latch_at_savepoint():
Reduce the size of the critical section.

fil_space_t::update_last_freed_lsn(), fil_space_t::clear_freed_ranges(),
fil_space_t::add_free_range(): Assume that freed_range_mutex is held
by the caller.

buf_pool_t::prepare_insert_into_flush_list(): Determine the insert
position for buf_pool_t::insert_into_flush_list(). Remove any clean
blocks from buf_pool.flush_list that were encountered while searching.

buf_pool_t::insert_into_flush_list(): Insert the block at the
predetermined position.

storage/innobase/buf/buf0flu.cc

@@ -164,68 +164,6 @@ inline void buf_pool_t::delete_from_flush_list_low(buf_page_t *bpage) noexcept
   UT_LIST_REMOVE(flush_list, bpage);
 }
 
-/** Insert a modified block into the flush list.
-@param block    modified block
-@param lsn      start LSN of the mini-transaction that modified the block */
-void buf_pool_t::insert_into_flush_list(buf_block_t *block, lsn_t lsn) noexcept
-{
-#ifndef SUX_LOCK_GENERIC
-  ut_ad(recv_recovery_is_on() || log_sys.latch.is_locked());
-#endif
-  ut_ad(lsn > 2);
-  static_assert(log_t::FIRST_LSN >= 2, "compatibility");
-  ut_ad(!fsp_is_system_temporary(block->page.id().space()));
-
-  mysql_mutex_lock(&flush_list_mutex);
-  if (ut_d(const lsn_t old=) block->page.oldest_modification())
-  {
-    ut_ad(old == 1);
-    delete_from_flush_list_low(&block->page);
-  }
-  else
-    stat.flush_list_bytes+= block->physical_size();
-  ut_ad(stat.flush_list_bytes <= curr_pool_size);
-  ut_ad(lsn >= log_sys.last_checkpoint_lsn);
-
-  block->page.set_oldest_modification(lsn);
-  MEM_CHECK_DEFINED(block->page.zip.data
-                    ? block->page.zip.data : block->page.frame,
-                    block->physical_size());
-rescan:
-  if (buf_page_t *prev= UT_LIST_GET_FIRST(flush_list))
-  {
-    lsn_t om= prev->oldest_modification();
-    if (om == 1)
-    {
-      delete_from_flush_list(prev);
-      goto rescan;
-    }
-    ut_ad(om > 2);
-    if (om <= lsn)
-      goto insert_first;
-    while (buf_page_t *next= UT_LIST_GET_NEXT(list, prev))
-    {
-      om= next->oldest_modification();
-      if (om == 1)
-      {
-        delete_from_flush_list(next);
-        continue;
-      }
-      ut_ad(om > 2);
-      if (om <= lsn)
-        break;
-      else
-        prev= next;
-    }
-    flush_hp.adjust(prev);
-    UT_LIST_INSERT_AFTER(flush_list, prev, &block->page);
-  }
-  else
-  insert_first:
-    UT_LIST_ADD_FIRST(flush_list, &block->page);
-  mysql_mutex_unlock(&flush_list_mutex);
-}
-
 /** Remove a block from flush_list.
 @param bpage   buffer pool page
 @param clear   whether to invoke buf_page_t::clear_oldest_modification() */

storage/innobase/include/buf0buf.h

@@ -808,7 +808,7 @@ class buf_page_t
   {
     ut_ad(fsp_is_system_temporary(id().space()));
     ut_ad(in_file());
-    ut_ad(!oldest_modification());
+    ut_ad(!oldest_modification() || oldest_modification() == 2);
     oldest_modification_= 2;
   }
 
@@ -1744,8 +1744,11 @@ class buf_pool_t
   /** modified blocks (a subset of LRU) */
   UT_LIST_BASE_NODE_T(buf_page_t) flush_list;
   /** number of blocks ever added to flush_list;
-  protected by flush_list_mutex */
+  sometimes protected by flush_list_mutex */
   size_t flush_list_requests;
+
+  TPOOL_SUPPRESS_TSAN void add_flush_list_requests(size_t size)
+  { ut_ad(size); flush_list_requests+= size; }
 private:
   /** whether the page cleaner needs wakeup from indefinite sleep */
   bool page_cleaner_is_idle;
@@ -1892,10 +1895,17 @@ class buf_pool_t
   void delete_from_flush_list(buf_page_t *bpage) noexcept
   { delete_from_flush_list(bpage, true); }
 
+  /** Prepare to insert a modified blcok into flush_list.
+  @param lsn start LSN of the mini-transaction
+  @return insert position for insert_into_flush_list() */
+  inline buf_page_t *prepare_insert_into_flush_list(lsn_t lsn) noexcept;
+
   /** Insert a modified block into the flush list.
+  @param prev     insert position (from prepare_insert_into_flush_list())
   @param block    modified block
   @param lsn      start LSN of the mini-transaction that modified the block */
-  void insert_into_flush_list(buf_block_t *block, lsn_t lsn) noexcept;
+  inline void insert_into_flush_list(buf_page_t *prev, buf_block_t *block,
+                                     lsn_t lsn) noexcept;
 
   /** Free a page whose underlying file page has been freed. */
   inline void release_freed_page(buf_page_t *bpage) noexcept;

storage/innobase/include/fil0fil.h

@@ -425,9 +425,10 @@ struct fil_space_t final
   /** Whether any corrupton of this tablespace has been reported */
   mutable std::atomic_flag is_corrupted;
 
+public:
   /** mutex to protect freed_ranges and last_freed_lsn */
   std::mutex freed_range_mutex;
-
+private:
   /** Ranges of freed page numbers; protected by freed_range_mutex */
   range_set freed_ranges;
 
@@ -647,11 +648,7 @@ struct fil_space_t final
   /** @return last_freed_lsn */
   lsn_t get_last_freed_lsn() { return last_freed_lsn; }
   /** Update last_freed_lsn */
-  void update_last_freed_lsn(lsn_t lsn)
-  {
-    std::lock_guard<std::mutex> freed_lock(freed_range_mutex);
-    last_freed_lsn= lsn;
-  }
+  void update_last_freed_lsn(lsn_t lsn) { last_freed_lsn= lsn; }
 
   /** Note that the file will need fsync().
   @return whether this needs to be added to fil_system.unflushed_spaces */
@@ -672,11 +669,7 @@ struct fil_space_t final
 
   /** Clear all freed ranges for undo tablespace when InnoDB
   encounters TRIM redo log record */
-  void clear_freed_ranges()
-  {
-    std::lock_guard<std::mutex> freed_lock(freed_range_mutex);
-    freed_ranges.clear();
-  }
+  void clear_freed_ranges() { freed_ranges.clear(); }
 #endif /* !UNIV_INNOCHECKSUM */
   /** FSP_SPACE_FLAGS and FSP_FLAGS_MEM_ flags;
   check fsp0types.h to more info about flags. */
@@ -949,7 +942,6 @@ struct fil_space_t final
   /** Add the set of freed page ranges */
   void add_free_range(const range_t range)
   {
-    std::lock_guard<std::mutex> freed_lock(freed_range_mutex);
     freed_ranges.add_range(range);
   }
 

storage/innobase/include/mtr0mtr.h

@@ -320,12 +320,9 @@ struct mtr_t {
   /** @return true if we are inside the change buffer code */
   bool is_inside_ibuf() const { return m_inside_ibuf; }
 
-  /** Note that pages has been trimed */
+  /** Note that some pages have been freed */
   void set_trim_pages() { m_trim_pages= true; }
 
-  /** @return true if pages has been trimed */
-  bool is_trim_pages() { return m_trim_pages; }
-
   /** Latch a buffer pool block.
   @param block    block to be latched
   @param rw_latch RW_S_LATCH, RW_SX_LATCH, RW_X_LATCH, RW_NO_LATCH */
@@ -614,6 +611,11 @@ struct mtr_t {
 #endif
 
 private:
+  /** Handle any pages that were freed during the mini-transaction. */
+  void process_freed_pages();
+  /** Release modified pages when no log was written. */
+  void release_unlogged();
+
   /** Log a write of a byte string to a page.
   @param block   buffer page
   @param offset  byte offset within page

storage/innobase/include/mtr0mtr.inl

@@ -106,18 +106,15 @@ mtr_t::sx_latch_at_savepoint(
 	mtr_memo_slot_t* slot = m_memo.at<mtr_memo_slot_t*>(savepoint);
 
 	ut_ad(slot->object == block);
-
-	/* == RW_NO_LATCH */
-	ut_a(slot->type == MTR_MEMO_BUF_FIX);
+	ut_ad(slot->type == MTR_MEMO_BUF_FIX); /* == RW_NO_LATCH */
+	slot->type = MTR_MEMO_PAGE_SX_FIX;
 
 	block->page.lock.u_lock();
 	ut_ad(!block->page.is_io_fixed());
 
 	if (!m_made_dirty) {
 		m_made_dirty = is_block_dirtied(block);
 	}
-
-	slot->type = MTR_MEMO_PAGE_SX_FIX;
 }
 
 /**
@@ -140,18 +137,15 @@ mtr_t::x_latch_at_savepoint(
 	mtr_memo_slot_t* slot = m_memo.at<mtr_memo_slot_t*>(savepoint);
 
 	ut_ad(slot->object == block);
-
-	/* == RW_NO_LATCH */
-	ut_a(slot->type == MTR_MEMO_BUF_FIX);
+	ut_ad(slot->type == MTR_MEMO_BUF_FIX); /* == RW_NO_LATCH */
+	slot->type = MTR_MEMO_PAGE_X_FIX;
 
 	block->page.lock.x_lock();
 	ut_ad(!block->page.is_io_fixed());
 
 	if (!m_made_dirty) {
 		m_made_dirty = is_block_dirtied(block);
 	}
-
-	slot->type = MTR_MEMO_PAGE_X_FIX;
 }
 
 /**