MDEV-15053 follow-up to reduce buf_pool.mutex contention

buf_LRU_make_block_young(): Merge with buf_page_make_young().

buf_pool_check_no_pending_io(): Remove. Replaced with
buf_pool.any_io_pending() and buf_pool.io_pending(),
which do not unnecessarily acquire buf_pool.mutex.

buf_pool_t::init_flush[]: Use atomic access, so that
buf_flush_wait_LRU_batch_end() can avoid acquiring buf_pool.mutex.

buf_pool_t::try_LRU_scan: Declare as bool.

storage/innobase/buf/buf0buf.cc

@@ -2556,22 +2556,6 @@ inline buf_page_t *buf_pool_t::watch_set(const page_id_t id,
   return nullptr;
 }
 
-/********************************************************************//**
-Moves a page to the start of the buffer pool LRU list. This high-level
-function can be used to prevent an important page from slipping out of
-the buffer pool.
-@param[in,out]	bpage	buffer block of a file page */
-void buf_page_make_young(buf_page_t* bpage)
-{
-	mutex_enter(&buf_pool.mutex);
-
-	ut_a(bpage->in_file());
-
-	buf_LRU_make_block_young(bpage);
-
-	mutex_exit(&buf_pool.mutex);
-}
-
 /** Mark the page status as FREED for the given tablespace id and
 page number. If the page is not in the buffer pool then ignore it.
 X-lock should be taken on the page before marking the page status
@@ -4965,22 +4949,6 @@ bool buf_page_verify_crypt_checksum(const byte* page, ulint fsp_flags)
 	return !buf_page_is_corrupted(true, page, fsp_flags);
 }
 
-/** Checks that there currently are no I/O operations pending.
-@return number of pending i/o */
-ulint buf_pool_check_no_pending_io()
-{
-	/* FIXME: use atomics, no mutex */
-	ulint pending_io = buf_pool.n_pend_reads;
-	mutex_enter(&buf_pool.mutex);
-	pending_io +=
-		+ buf_pool.n_flush[IORequest::LRU]
-		+ buf_pool.n_flush[IORequest::FLUSH_LIST]
-		+ buf_pool.n_flush[IORequest::SINGLE_PAGE];
-	mutex_exit(&buf_pool.mutex);
-
-	return(pending_io);
-}
-
 /** Print the given page_id_t object.
 @param[in,out]	out	the output stream
 @param[in]	page_id	the page_id_t object to be printed

storage/innobase/buf/buf0flu.cc

@@ -1936,11 +1936,7 @@ static ulint buf_flush_LRU_list()
 /** Wait for any possible LRU flushes to complete. */
 void buf_flush_wait_LRU_batch_end()
 {
-  mutex_enter(&buf_pool.mutex);
-  bool wait= buf_pool.n_flush[IORequest::LRU] ||
-    buf_pool.init_flush[IORequest::LRU];
-  mutex_exit(&buf_pool.mutex);
-  if (wait)
+  if (buf_pool.n_flush[IORequest::LRU] || buf_pool.init_flush[IORequest::LRU])
     buf_flush_wait_batch_end(true);
 }
 

storage/innobase/buf/buf0lru.cc

@@ -771,7 +771,7 @@ buf_block_t* buf_LRU_get_free_block(bool have_mutex)
 
 	MONITOR_INC( MONITOR_LRU_GET_FREE_LOOPS );
 	freed = false;
-	if (buf_pool.try_LRU_scan || n_iterations > 0) {
+	if (n_iterations || buf_pool.try_LRU_scan) {
 		/* If no block was in the free list, search from the
 		end of the LRU list and try to free a block there.
 		If we are doing for the first time we'll scan only
@@ -784,7 +784,7 @@ buf_block_t* buf_LRU_get_free_block(bool have_mutex)
 			in scanning the LRU list. This flag is set to
 			TRUE again when we flush a batch from this
 			buffer pool. */
-			buf_pool.try_LRU_scan = FALSE;
+			buf_pool.try_LRU_scan = false;
 
 			/* Also tell the page_cleaner thread that
 			there is work for it to do. */
@@ -1121,21 +1121,20 @@ buf_LRU_add_block(
 	}
 }
 
-/******************************************************************//**
-Moves a block to the start of the LRU list. */
-void
-buf_LRU_make_block_young(
-/*=====================*/
-	buf_page_t*	bpage)	/*!< in: control block */
+/** Move a block to the start of the LRU list. */
+void buf_page_make_young(buf_page_t *bpage)
 {
-	ut_ad(mutex_own(&buf_pool.mutex));
+  ut_ad(bpage->in_file());
 
-	if (bpage->old) {
-		buf_pool.stat.n_pages_made_young++;
-	}
+  mutex_enter(&buf_pool.mutex);
 
-	buf_LRU_remove_block(bpage);
-	buf_LRU_add_block(bpage, false);
+  if (UNIV_UNLIKELY(bpage->old))
+    buf_pool.stat.n_pages_made_young++;
+
+  buf_LRU_remove_block(bpage);
+  buf_LRU_add_block(bpage, false);
+
+  mutex_exit(&buf_pool.mutex);
 }
 
 /** Try to free a block. If bpage is a descriptor of a compressed-only

storage/innobase/include/buf0buf.h

@@ -363,15 +363,8 @@ buf_page_release_latch(
 	buf_block_t*	block,		/*!< in: buffer block */
 	ulint		rw_latch);	/*!< in: RW_S_LATCH, RW_X_LATCH,
 					RW_NO_LATCH */
-/********************************************************************//**
-Moves a page to the start of the buffer pool LRU list. This high-level
-function can be used to prevent an important page from slipping out of
-the buffer pool. */
-void
-buf_page_make_young(
-/*================*/
-	buf_page_t*	bpage);	/*!< in: buffer block of a file page */
-
+/** Move a block to the start of the LRU list. */
+void buf_page_make_young(buf_page_t *bpage);
 /** Mark the page status as FREED for the given tablespace id and
 page number. If the page is not in buffer pool then ignore it.
 @param[in]	page_id	page_id
@@ -654,10 +647,6 @@ void buf_stats_get_pool_info(buf_pool_info_t *pool_info);
 /** Refresh the statistics used to print per-second averages. */
 void buf_refresh_io_stats();
 
-/** Check that there currently are no I/O operations pending.
-@return number of pending i/o */
-ulint buf_pool_check_no_pending_io();
-
 /** Invalidate all pages in the buffer pool.
 All pages must be in a replaceable state (not modified or latched). */
 void buf_pool_invalidate();
@@ -1441,7 +1430,7 @@ struct buf_pool_stat_t{
 				pages that are evicted without
 				being accessed */
 	ulint	n_pages_made_young; /*!< number of pages made young, in
-				calls to buf_LRU_make_block_young() */
+				buf_page_make_young() */
 	ulint	n_pages_not_made_young; /*!< number of pages not made
 				young because the first access
 				was not long enough ago, in
@@ -1943,9 +1932,8 @@ class buf_pool_t
 	UT_LIST_BASE_NODE_T(buf_page_t) flush_list;
 					/*!< base node of the modified block
 					list */
-	ibool		init_flush[3];
-					/*!< this is TRUE when a flush of the
-					given type is being initialized */
+	/** set if a flush of the type is being initialized */
+	Atomic_relaxed<bool> init_flush[3];
 	/** Number of pending writes of a flush type.
 	The sum of these is approximately the sum of BUF_IO_WRITE blocks. */
 	Atomic_counter<ulint> n_flush[3];
@@ -1978,13 +1966,13 @@ class buf_pool_t
 					to read this for heuristic
 					purposes without holding any
 					mutex or latch */
-	ibool		try_LRU_scan;	/*!< Set to FALSE when an LRU
+	bool		try_LRU_scan;	/*!< Cleared when an LRU
 					scan for free block fails. This
 					flag is used to avoid repeated
 					scans of LRU list when we know
 					that there is no free block
 					available in the scan depth for
-					eviction. Set to TRUE whenever
+					eviction. Set whenever
 					we flush a batch from the
 					buffer pool. Protected by the
 					buf_pool.mutex */
@@ -2063,6 +2051,21 @@ class buf_pool_t
   buf_page_t watch[innodb_purge_threads_MAX + 1];
   /** Reserve a buffer. */
   buf_tmp_buffer_t *io_buf_reserve() { return io_buf.reserve(); }
+
+  /** @return whether any I/O is pending */
+  bool any_io_pending() const
+  {
+    return n_pend_reads ||
+      n_flush[IORequest::LRU] || n_flush[IORequest::FLUSH_LIST] ||
+      n_flush[IORequest::SINGLE_PAGE];
+  }
+  /** @return total amount of pending I/O */
+  ulint io_pending() const
+  {
+    return n_pend_reads +
+      n_flush[IORequest::LRU] + n_flush[IORequest::FLUSH_LIST] +
+      n_flush[IORequest::SINGLE_PAGE];
+  }
 private:
   /** Temporary memory for page_compressed and encrypted I/O */
   struct io_buf_t

storage/innobase/include/buf0lru.h

@@ -133,10 +133,6 @@ buf_unzip_LRU_add_block(
 	buf_block_t*	block,	/*!< in: control block */
 	ibool		old);	/*!< in: TRUE if should be put to the end
 				of the list, else put to the start */
-/******************************************************************//**
-Moves a block to the start of the LRU list. */
-void
-buf_LRU_make_block_young(buf_page_t* bpage);
 
 /** Update buf_pool.LRU_old_ratio.
 @param[in]	old_pct		Reserve this percentage of

storage/innobase/log/log0log.cc

@@ -1663,7 +1663,7 @@ void logs_empty_and_mark_files_at_shutdown()
 
 	if (!buf_pool.is_initialised()) {
 		ut_ad(!srv_was_started);
-	} else if (ulint pending_io = buf_pool_check_no_pending_io()) {
+	} else if (ulint pending_io = buf_pool.io_pending()) {
 		if (srv_print_verbose_log && count > 600) {
 			ib::info() << "Waiting for " << pending_io << " buffer"
 				" page I/Os to complete";

storage/innobase/srv/srv0start.cc

@@ -269,7 +269,7 @@ static dberr_t create_log_file(lsn_t lsn, std::string& logfile0)
 	}
 
 	DBUG_PRINT("ib_log", ("After innodb_log_abort_6"));
-	ut_ad(!buf_pool_check_no_pending_io());
+	DBUG_ASSERT(!buf_pool.any_io_pending());
 
 	DBUG_EXECUTE_IF("innodb_log_abort_7", return DB_ERROR;);
 	DBUG_PRINT("ib_log", ("After innodb_log_abort_7"));
@@ -979,14 +979,13 @@ static lsn_t srv_prepare_to_delete_redo_log_file(bool old_exists)
 	DBUG_ENTER("srv_prepare_to_delete_redo_log_file");
 
 	lsn_t	flushed_lsn;
-	ulint	pending_io = 0;
 	ulint	count = 0;
 
 	if (log_sys.log.subformat != 2) {
 		srv_log_file_size = 0;
 	}
 
-	do {
+	for (;;) {
 		/* Clean the buffer pool. */
 		buf_flush_sync();
 
@@ -1043,9 +1042,7 @@ static lsn_t srv_prepare_to_delete_redo_log_file(bool old_exists)
 
 		/* Check if the buffer pools are clean.  If not
 		retry till it is clean. */
-		pending_io = buf_pool_check_no_pending_io();
-
-		if (pending_io > 0) {
+		if (ulint pending_io = buf_pool.io_pending()) {
 			count++;
 			/* Print a message every 60 seconds if we
 			are waiting to clean the buffer pools */
@@ -1055,10 +1052,13 @@ static lsn_t srv_prepare_to_delete_redo_log_file(bool old_exists)
 					<< "page I/Os to complete";
 				count = 0;
 			}
+
+			os_thread_sleep(100000);
+			continue;
 		}
-		os_thread_sleep(100000);
 
-	} while (buf_pool_check_no_pending_io());
+		break;
+	}
 
 	DBUG_RETURN(flushed_lsn);
 }
@@ -1677,7 +1677,7 @@ dberr_t srv_start(bool create_new_db)
 			ut_ad(recv_no_log_write);
 			buf_flush_sync();
 			err = fil_write_flushed_lsn(log_get_lsn());
-			ut_ad(!buf_pool_check_no_pending_io());
+			DBUG_ASSERT(!buf_pool.any_io_pending());
 			log_sys.log.close_file();
 			if (err == DB_SUCCESS) {
 				bool trunc = srv_operation
@@ -1721,7 +1721,7 @@ dberr_t srv_start(bool create_new_db)
 			threads until creating a log checkpoint at the
 			end of create_log_file(). */
 			ut_d(recv_no_log_write = true);
-			ut_ad(!buf_pool_check_no_pending_io());
+			DBUG_ASSERT(!buf_pool.any_io_pending());
 
 			DBUG_EXECUTE_IF("innodb_log_abort_3",
 					return(srv_init_abort(DB_ERROR)););