Merge 10.1 into 10.2

14 files changed, 165 insertions, 295 deletions

diff --git a/storage/innobase/buf/buf0flu.cc b/storage/innobase/buf/buf0flu.cc
index beffc3f8e8e..2c660c76c33 100644
--- a/storage/innobase/buf/buf0flu.cc
+++ b/storage/innobase/buf/buf0flu.cc
@@ -70,7 +70,7 @@ is set to TRUE by the page_cleaner thread when it is spawned and is set
 back to FALSE at shutdown by the page_cleaner as well. Therefore no
 need to protect it by a mutex. It is only ever read by the thread
 doing the shutdown */
-bool buf_page_cleaner_is_active = false;
+bool buf_page_cleaner_is_active;
 
 /** Factor for scan length to determine n_pages for intended oldest LSN
 progress */
@@ -3160,8 +3160,6 @@ DECLARE_THREAD(buf_flush_page_cleaner_coordinator)(
 	}
 #endif /* UNIV_LINUX */
 
-	buf_page_cleaner_is_active = true;
-
 	while (!srv_read_only_mode
 	       && srv_shutdown_state == SRV_SHUTDOWN_NONE
 	       && recv_sys->heap != NULL) {
@@ -3489,8 +3487,6 @@ thread_exit:
 	buf_page_cleaner_is_active = false;
 
 	my_thread_end();
-
-	my_thread_end();
 	/* We count the number of threads in os_thread_exit(). A created
 	thread should always use that to exit and not use return() to exit. */
 	os_thread_exit();
diff --git a/storage/innobase/dict/dict0stats_bg.cc b/storage/innobase/dict/dict0stats_bg.cc
index 85d9ea01081..f2ef18b116d 100644
--- a/storage/innobase/dict/dict0stats_bg.cc
+++ b/storage/innobase/dict/dict0stats_bg.cc
@@ -38,8 +38,6 @@ Created Apr 25, 2012 Vasil Dimov
 /** Minimum time interval between stats recalc for a given table */
 #define MIN_RECALC_INTERVAL	10 /* seconds */
 
-#define SHUTTING_DOWN()		(srv_shutdown_state != SRV_SHUTDOWN_NONE)
-
 /** Event to wake up dict_stats_thread on dict_stats_recalc_pool_add()
 or shutdown. Not protected by any mutex. */
 os_event_t			dict_stats_event;
@@ -270,7 +268,6 @@ Initialize global variables needed for the operation of dict_stats_thread()
 Must be called before dict_stats_thread() is started. */
 void
 dict_stats_thread_init()
-/*====================*/
 {
 	ut_a(!srv_read_only_mode);
 
@@ -315,15 +312,9 @@ dict_stats_thread_deinit()
 
 	mutex_free(&recalc_pool_mutex);
 
-#ifdef UNIV_DEBUG
-	os_event_destroy(dict_stats_disabled_event);
-	dict_stats_disabled_event = NULL;
-#endif /* UNIV_DEBUG */
-
+	ut_d(os_event_destroy(dict_stats_disabled_event));
 	os_event_destroy(dict_stats_event);
 	os_event_destroy(dict_stats_shutdown_event);
-	dict_stats_event = NULL;
-	dict_stats_shutdown_event = NULL;
 	dict_stats_start_shutdown = false;
 }
 
@@ -492,7 +483,7 @@ DECLARE_THREAD(dict_stats_thread)(void*)
 	OS_THREAD_DUMMY_RETURN;
 }
 
-/** Shutdown the dict stats thread. */
+/** Shut down the dict_stats_thread. */
 void
 dict_stats_shutdown()
 {
diff --git a/storage/innobase/fil/fil0crypt.cc b/storage/innobase/fil/fil0crypt.cc
index 5cfacc5bcd8..7917cbb528b 100644
--- a/storage/innobase/fil/fil0crypt.cc
+++ b/storage/innobase/fil/fil0crypt.cc
@@ -922,6 +922,34 @@ fil_crypt_needs_rotation(
 	return false;
 }
 
+/** Read page 0 and possible crypt data from there.
+@param[in,out]	space		Tablespace */
+static inline
+void
+fil_crypt_read_crypt_data(fil_space_t* space)
+{
+	if (space->crypt_data || space->size) {
+		/* The encryption metadata has already been read, or
+		the tablespace is not encrypted and the file has been
+		opened already. */
+		return;
+	}
+
+	const page_size_t page_size(space->flags);
+	mtr_t	mtr;
+	mtr.start();
+	if (buf_block_t* block = buf_page_get(page_id_t(space->id, 0),
+					      page_size, RW_S_LATCH, &mtr)) {
+		mutex_enter(&fil_system->mutex);
+		if (!space->crypt_data) {
+			space->crypt_data = fil_space_read_crypt_data(
+				page_size, block->frame);
+		}
+		mutex_exit(&fil_system->mutex);
+	}
+	mtr.commit();
+}
+
 /***********************************************************************
 Start encrypting a space
 @param[in,out]		space		Tablespace
@@ -932,6 +960,7 @@ fil_crypt_start_encrypting_space(
 	fil_space_t*	space)
 {
 	bool recheck = false;
+
 	mutex_enter(&fil_crypt_threads_mutex);
 
 	fil_space_crypt_t *crypt_data = space->crypt_data;
@@ -1428,6 +1457,13 @@ fil_crypt_find_space_to_rotate(
 	}
 
 	while (!state->should_shutdown() && state->space) {
+		/* If there is no crypt data and we have not yet read
+		page 0 for this tablespace, we need to read it before
+		we can continue. */
+		if (!state->space->crypt_data) {
+			fil_crypt_read_crypt_data(state->space);
+		}
+
 		if (fil_crypt_space_needs_rotation(state, key_state, recheck)) {
 			ut_ad(key_state->key_id);
 			/* init state->min_key_version_found before
@@ -2129,8 +2165,10 @@ DECLARE_THREAD(fil_crypt_thread)(
 			while (!thr.should_shutdown() &&
 			       fil_crypt_find_page_to_rotate(&new_state, &thr)) {
 
-				/* rotate a (set) of pages */
-				fil_crypt_rotate_pages(&new_state, &thr);
+				if (!thr.space->is_stopping()) {
+					/* rotate a (set) of pages */
+					fil_crypt_rotate_pages(&new_state, &thr);
+				}
 
 				/* If space is marked as stopping, release
 				space and stop rotation. */
@@ -2357,6 +2395,14 @@ fil_space_crypt_get_status(
 	memset(status, 0, sizeof(*status));
 
 	ut_ad(space->n_pending_ops > 0);
+
+	/* If there is no crypt data and we have not yet read
+	page 0 for this tablespace, we need to read it before
+	we can continue. */
+	if (!space->crypt_data) {
+		fil_crypt_read_crypt_data(const_cast<fil_space_t*>(space));
+	}
+
 	fil_space_crypt_t* crypt_data = space->crypt_data;
 	status->space = space->id;
 
diff --git a/storage/innobase/fil/fil0fil.cc b/storage/innobase/fil/fil0fil.cc
index 72358e7628b..2cbc863102c 100644
--- a/storage/innobase/fil/fil0fil.cc
+++ b/storage/innobase/fil/fil0fil.cc
@@ -655,12 +655,8 @@ retry:
 			FSP_HEADER_OFFSET + FSP_FREE + page);
 
 		/* Try to read crypt_data from page 0 if it is not yet
-		read. FIXME: Remove page_0_crypt_read, and simply ensure in
-		fil_space_t object creation that node->size==0 if and only
-		if the crypt_data is not known and must be read. */
-		if (!space->page_0_crypt_read) {
-			space->page_0_crypt_read = true;
-			ut_ad(space->crypt_data == NULL);
+		read. */
+		if (!space->crypt_data) {
 			space->crypt_data = fil_space_read_crypt_data(
 				page_size_t(space->flags), page);
 		}
@@ -1591,7 +1587,6 @@ Error messages are issued to the server log.
 @param[in]	flags		tablespace flags
 @param[in]	purpose		tablespace purpose
 @param[in,out]	crypt_data	encryption information
-@param[in]	create_table	whether this is CREATE TABLE
 @param[in]	mode		encryption mode
 @return pointer to created tablespace, to be filled in with fil_node_create()
 @retval NULL on failure (such as when the same tablespace exists) */
@@ -1602,7 +1597,6 @@ fil_space_create(
 	ulint			flags,
 	fil_type_t		purpose,
 	fil_space_crypt_t*	crypt_data,
-	bool			create_table,
 	fil_encryption_t	mode)
 {
 	fil_space_t*	space;
@@ -1667,16 +1661,8 @@ fil_space_create(
 	space->magic_n = FIL_SPACE_MAGIC_N;
 	space->crypt_data = crypt_data;
 
-	/* In create table we write page 0 so we have already
-	"read" it and for system tablespaces we have read
-	crypt data at startup. */
-	if (create_table || crypt_data != NULL) {
-		space->page_0_crypt_read = true;
-	}
-
 	DBUG_LOG("tablespace",
-		 "Tablespace for space " << id << " name " << name
-		 << (create_table ? " created" : " opened"));
+		 "Created metadata for " << id << " name " << name);
 	if (crypt_data) {
 		DBUG_LOG("crypt",
 			 "Tablespace " << id << " name " << name
@@ -4004,7 +3990,7 @@ fil_ibd_create(
 	}
 
 	space = fil_space_create(name, space_id, flags, FIL_TYPE_TABLESPACE,
-				 crypt_data, true, mode);
+				 crypt_data, mode);
 
 	fil_node_t* node = NULL;
 
@@ -4372,7 +4358,7 @@ skip_validate:
 			space_name, id, flags, purpose,
 			df_remote.is_open() ? df_remote.get_crypt_info() :
 			df_dict.is_open() ? df_dict.get_crypt_info() :
-			df_default.get_crypt_info(), false);
+			df_default.get_crypt_info());
 
 		/* We do not measure the size of the file, that is why
 		we pass the 0 below */
@@ -4692,7 +4678,7 @@ fil_ibd_load(
 
 	space = fil_space_create(
 		file.name(), space_id, flags, FIL_TYPE_TABLESPACE,
-		file.get_crypt_info(), false);
+		file.get_crypt_info());
 
 	if (space == NULL) {
 		return(FIL_LOAD_INVALID);
diff --git a/storage/innobase/fsp/fsp0fsp.cc b/storage/innobase/fsp/fsp0fsp.cc
index 467fbbd557d..bf86d83a8c7 100644
--- a/storage/innobase/fsp/fsp0fsp.cc
+++ b/storage/innobase/fsp/fsp0fsp.cc
@@ -874,14 +874,12 @@ fsp_try_extend_data_file_with_pages(
 @param[in,out]	space	tablespace
 @param[in,out]	header	tablespace header
 @param[in,out]	mtr	mini-transaction
-@return whether the tablespace was extended */
-static UNIV_COLD MY_ATTRIBUTE((nonnull))
+@return	number of pages added
+@retval	0 if the tablespace was not extended */
+UNIV_COLD MY_ATTRIBUTE((nonnull))
+static
 ulint
-fsp_try_extend_data_file(
-	fil_space_t*	space,
-	fsp_header_t*	header,
-	mtr_t*		mtr,
-	ulint*		n_pages_added)
+fsp_try_extend_data_file(fil_space_t* space, fsp_header_t* header, mtr_t* mtr)
 {
 	ulint	size;		/* current number of pages in the datafile */
 	ulint	size_increase;	/* number of pages to extend this file */
@@ -904,7 +902,7 @@ fsp_try_extend_data_file(
 				<< " innodb_data_file_path.";
 			srv_sys_space.set_tablespace_full_status(true);
 		}
-		return(false);
+		return(0);
 	} else if (fsp_is_system_temporary(space->id)
 		   && !srv_tmp_space.can_auto_extend_last_file()) {
 
@@ -918,7 +916,7 @@ fsp_try_extend_data_file(
 				<< " innodb_temp_data_file_path.";
 			srv_tmp_space.set_tablespace_full_status(true);
 		}
-		return(false);
+		return(0);
 	}
 
 	size = mach_read_from_4(header + FSP_SIZE);
@@ -941,7 +939,7 @@ fsp_try_extend_data_file(
 			/* Let us first extend the file to extent_size */
 			if (!fsp_try_extend_data_file_with_pages(
 				    space, extent_pages - 1, header, mtr)) {
-				return(false);
+				return(0);
 			}
 
 			size = extent_pages;
@@ -951,16 +949,13 @@ fsp_try_extend_data_file(
 	}
 
 	if (size_increase == 0) {
-
-		return(false);
+		return(0);
 	}
 
 	if (!fil_space_extend(space, size + size_increase)) {
-		return(false);
+		return(0);
 	}
 
-	*n_pages_added = size_increase;
-
 	/* We ignore any fragments of a full megabyte when storing the size
 	to the space header */
 
@@ -970,7 +965,7 @@ fsp_try_extend_data_file(
 	mlog_write_ulint(
 		header + FSP_SIZE, space->size_in_header, MLOG_4BYTES, mtr);
 
-	return(true);
+	return(size_increase);
 }
 
 /** Calculate the number of pages to extend a datafile.
@@ -1062,8 +1057,7 @@ fsp_fill_free_list(
 		}
 
 		if (!skip_resize) {
-			ulint n_pages = 0;
-			fsp_try_extend_data_file(space, header, mtr, &n_pages);
+			fsp_try_extend_data_file(space, header, mtr);
 			size = space->size_in_header;
 		}
 	}
@@ -2784,8 +2778,6 @@ fsp_reserve_free_extents(
 	ulint		n_free_up;
 	ulint		reserve;
 	size_t		total_reserved = 0;
-	ulint		rounds = 0;
-	ulint		n_pages_added = 0;
 
 	ut_ad(mtr);
 	*n_reserved = n_ext;
@@ -2866,23 +2858,8 @@ try_again:
 		return(true);
 	}
 try_to_extend:
-	n_pages_added = 0;
-
-	if (fsp_try_extend_data_file(space, space_header, mtr, &n_pages_added)) {
-
-		rounds++;
-		total_reserved += n_pages_added;
-
-		if (rounds > 10) {
-			ib::info() << "Space id: "
-				   << space << " trying to reserve: "
-				   << n_ext << " extents actually reserved: "
-				   << n_pages_added << " reserve: "
-				   << reserve << " free: " << n_free
-				   << " size: " << size
-				   << " rounds: " << rounds
-				   << " total_reserved: " << total_reserved << ".";
-		}
+	if (ulint n = fsp_try_extend_data_file(space, space_header, mtr)) {
+		total_reserved += n;
 		goto try_again;
 	}
 
diff --git a/storage/innobase/fsp/fsp0space.cc b/storage/innobase/fsp/fsp0space.cc
index 76269a749f9..7ca81898f70 100644
--- a/storage/innobase/fsp/fsp0space.cc
+++ b/storage/innobase/fsp/fsp0space.cc
@@ -132,8 +132,7 @@ Tablespace::open_or_create(bool is_temp)
 				m_name, m_space_id, FSP_FLAGS_PAGE_SSIZE(),
 				is_temp
 				? FIL_TYPE_TEMPORARY : FIL_TYPE_TABLESPACE,
-				it->m_crypt_info,
-				false);
+				it->m_crypt_info);
 		}
 
 		ut_a(fil_validate());
diff --git a/storage/innobase/fsp/fsp0sysspace.cc b/storage/innobase/fsp/fsp0sysspace.cc
index fa0203586fb..e4bb11c9a22 100644
--- a/storage/innobase/fsp/fsp0sysspace.cc
+++ b/storage/innobase/fsp/fsp0sysspace.cc
@@ -930,8 +930,8 @@ SysTablespace::open_or_create(
 
 			space = fil_space_create(
 				name(), space_id(), flags(), is_temp
-				? FIL_TYPE_TEMPORARY : FIL_TYPE_TABLESPACE, m_crypt_info,
-				false);
+				? FIL_TYPE_TEMPORARY : FIL_TYPE_TABLESPACE,
+				m_crypt_info);
 		}
 
 		ut_a(fil_validate());
diff --git a/storage/innobase/handler/ha_innodb.cc b/storage/innobase/handler/ha_innodb.cc
index c162c1f9f3f..23edecda245 100644
--- a/storage/innobase/handler/ha_innodb.cc
+++ b/storage/innobase/handler/ha_innodb.cc
@@ -349,12 +349,12 @@ thd_destructor_proxy(void *)
 		while (trx_sys_any_active_transactions()) {
 			os_thread_sleep(1000);
 		}
-
-		/* Some background threads might generate undo pages that will
-		need to be purged, so they have to be shut down before purge
-		threads if slow shutdown is requested.  */
-		srv_shutdown_bg_undo_sources();
 	}
+
+	/* Some background threads might generate undo pages that will
+	need to be purged, so they have to be shut down before purge
+	threads if slow shutdown is requested.  */
+	srv_shutdown_bg_undo_sources();
 	srv_purge_wakeup();
 
 	destroy_thd(thd);
@@ -1448,14 +1448,11 @@ innobase_drop_database(
 	handlerton*	hton,
 	char*		path);
 
-/*******************************************************************//**
-Closes an InnoDB database. */
+/** Shut down the InnoDB storage engine.
+@return	0 */
 static
 int
-innobase_end(
-/*=========*/
-	handlerton*		hton,	/* in: InnoDB handlerton */
-	ha_panic_function	type);
+innobase_end(handlerton*, ha_panic_function);
 
 /*****************************************************************//**
 Creates an InnoDB transaction struct for the thd if it does not yet have one.
@@ -3749,10 +3746,7 @@ innobase_space_shutdown()
 	srv_tmp_space.shutdown();
 
 #ifdef WITH_INNODB_DISALLOW_WRITES
-	if (srv_allow_writes_event) {
-		os_event_destroy(srv_allow_writes_event);
-		srv_allow_writes_event = NULL;
-	}
+	os_event_destroy(srv_allow_writes_event);
 #endif /* WITH_INNODB_DISALLOW_WRITES */
 
 	DBUG_VOID_RETURN;
@@ -4575,21 +4569,13 @@ error:
 	DBUG_RETURN(1);
 }
 
-/*******************************************************************//**
-Closes an InnoDB database.
-@return TRUE if error */
+/** Shut down the InnoDB storage engine.
+@return	0 */
 static
 int
-innobase_end(
-/*=========*/
-	handlerton*		hton,	/*!< in/out: InnoDB handlerton */
-	ha_panic_function	type MY_ATTRIBUTE((unused)))
-					/*!< in: ha_panic() parameter */
+innobase_end(handlerton*, ha_panic_function)
 {
-	int	err= 0;
-
 	DBUG_ENTER("innobase_end");
-	DBUG_ASSERT(hton == innodb_hton_ptr);
 
 	if (srv_was_started) {
 		THD *thd= current_thd;
@@ -4622,7 +4608,7 @@ innobase_end(
 		mysql_mutex_destroy(&pending_checkpoint_mutex);
 	}
 
-	DBUG_RETURN(err);
+	DBUG_RETURN(0);
 }
 
 /*****************************************************************//**
diff --git a/storage/innobase/include/dict0stats_bg.h b/storage/innobase/include/dict0stats_bg.h
index b0570d21ecb..f99cbeed780 100644
--- a/storage/innobase/include/dict0stats_bg.h
+++ b/storage/innobase/include/dict0stats_bg.h
@@ -140,7 +140,7 @@ DECLARE_THREAD(dict_stats_thread)(
 	void*	arg);	/*!< in: a dummy parameter
 			required by os_thread_create */
 
-/** Shutdown the dict stats thread. */
+/** Shut down the dict_stats_thread. */
 void
 dict_stats_shutdown();
 
diff --git a/storage/innobase/include/fil0fil.h b/storage/innobase/include/fil0fil.h
index fddf4f813dd..7d9327c598b 100644
--- a/storage/innobase/include/fil0fil.h
+++ b/storage/innobase/include/fil0fil.h
@@ -180,9 +180,6 @@ struct fil_space_t {
 	/** MariaDB encryption data */
 	fil_space_crypt_t* crypt_data;
 
-	/** tablespace crypt data has been read */
-	bool		page_0_crypt_read;
-
 	/** True if we have already printed compression failure */
 	bool		printed_compression_failure;
 
@@ -586,7 +583,6 @@ Error messages are issued to the server log.
 @param[in]	flags		tablespace flags
 @param[in]	purpose		tablespace purpose
 @param[in,out]	crypt_data	encryption information
-@param[in]	create_table	whether this is CREATE TABLE
 @param[in]	mode		encryption mode
 @return pointer to created tablespace, to be filled in with fil_node_create()
 @retval NULL on failure (such as when the same tablespace exists) */
@@ -597,7 +593,6 @@ fil_space_create(
 	ulint			flags,
 	fil_type_t		purpose,
 	fil_space_crypt_t*	crypt_data,
-	bool			create_table,
 	fil_encryption_t	mode = FIL_ENCRYPTION_DEFAULT)
 	MY_ATTRIBUTE((warn_unused_result));
 
diff --git a/storage/innobase/include/srv0start.h b/storage/innobase/include/srv0start.h
index 04ee6b42867..d7d988f5282 100644
--- a/storage/innobase/include/srv0start.h
+++ b/storage/innobase/include/srv0start.h
@@ -43,8 +43,8 @@ Starts Innobase and creates a new database if database files
 are not found and the user wants.
 @return DB_SUCCESS or error code */
 dberr_t
-innobase_start_or_create_for_mysql(void);
-/*====================================*/
+innobase_start_or_create_for_mysql();
+
 /** Shut down InnoDB. */
 void
 innodb_shutdown();
@@ -52,7 +52,7 @@ innodb_shutdown();
 /****************************************************************//**
 Shuts down background threads that can generate undo pages. */
 void
-srv_shutdown_bg_undo_sources(void);
+srv_shutdown_bg_undo_sources();
 
 /*************************************************************//**
 Copy the file path component of the physical file to parameter. It will
@@ -138,6 +138,9 @@ enum srv_shutdown_t {
 	SRV_SHUTDOWN_EXIT_THREADS/*!< Exit all threads */
 };
 
+/** Whether any undo log records can be generated */
+extern bool srv_undo_sources;
+
 /** At a shutdown this value climbs from SRV_SHUTDOWN_NONE to
 SRV_SHUTDOWN_CLEANUP and then to SRV_SHUTDOWN_LAST_PHASE, and so on */
 extern	enum srv_shutdown_t	srv_shutdown_state;
diff --git a/storage/innobase/srv/srv0srv.cc b/storage/innobase/srv/srv0srv.cc
index 56aa5626450..d7b7711e402 100644
--- a/storage/innobase/srv/srv0srv.cc
+++ b/storage/innobase/srv/srv0srv.cc
@@ -195,9 +195,6 @@ my_bool	srv_use_mtflush;
 my_bool	srv_master_thread_disabled_debug;
 /** Event used to inform that master thread is disabled. */
 static os_event_t	srv_master_thread_disabled_event;
-/** Debug variable to find if any background threads are adding
-to purge during slow shutdown. */
-extern bool		trx_commit_disallowed;
 #endif /* UNIV_DEBUG */
 
 /*------------------------- LOG FILES ------------------------ */
@@ -2558,35 +2555,29 @@ suspend_thread:
 	goto loop;
 }
 
-/**
-Check if purge should stop.
-@return true if it should shutdown. */
+/** Check if purge should stop.
+@param[in]	n_purged	pages purged in the last batch
+@return whether purge should exit */
 static
 bool
-srv_purge_should_exit(
-	MYSQL_THD	thd,
-	ulint		n_purged)	/*!< in: pages purged in last batch */
+srv_purge_should_exit(ulint n_purged)
 {
-	switch (srv_shutdown_state) {
-	case SRV_SHUTDOWN_NONE:
-		if ((!srv_was_started || srv_running)
-		    && !thd_kill_level(thd)) {
-			/* Normal operation. */
-			break;
-		}
-		/* close_connections() was called */
-		/* fall through */
-	case SRV_SHUTDOWN_CLEANUP:
-	case SRV_SHUTDOWN_EXIT_THREADS:
-		/* Exit unless slow shutdown requested or all done. */
-		return(srv_fast_shutdown != 0 || n_purged == 0);
+	ut_ad(srv_shutdown_state == SRV_SHUTDOWN_NONE
+	      || srv_shutdown_state == SRV_SHUTDOWN_CLEANUP);
 
-	case SRV_SHUTDOWN_LAST_PHASE:
-	case SRV_SHUTDOWN_FLUSH_PHASE:
-		ut_error;
+	if (srv_undo_sources) {
+		return(false);
 	}
-
-	return(false);
+	if (srv_fast_shutdown) {
+		return(true);
+	}
+	/* Slow shutdown was requested. */
+	if (n_purged) {
+		/* The previous round still did some work. */
+		return(false);
+	}
+	/* Exit if there are no active transactions to roll back. */
+	return(trx_sys_any_active_transactions() == 0);
 }
 
 /*********************************************************************//**
@@ -2699,16 +2690,12 @@ DECLARE_THREAD(srv_worker_thread)(
 	OS_THREAD_DUMMY_RETURN;	/* Not reached, avoid compiler warning */
 }
 
-/*********************************************************************//**
-Do the actual purge operation.
+/** Do the actual purge operation.
+@param[in,out]	n_total_purged	total number of purged pages
 @return length of history list before the last purge batch. */
 static
 ulint
-srv_do_purge(
-/*=========*/
-	MYSQL_THD	thd,
-	ulint		n_threads,	/*!< in: number of threads to use */
-	ulint*		n_total_purged)	/*!< in/out: total pages purged */
+srv_do_purge(ulint* n_total_purged)
 {
 	ulint		n_pages_purged;
 
@@ -2716,6 +2703,7 @@ srv_do_purge(
 	static ulint	n_use_threads = 0;
 	static ulint	rseg_history_len = 0;
 	ulint		old_activity_count = srv_get_activity_count();
+	const ulint	n_threads = srv_n_purge_threads;
 
 	ut_a(n_threads > 0);
 	ut_ad(!srv_read_only_mode);
@@ -2777,7 +2765,7 @@ srv_do_purge(
 
 		*n_total_purged += n_pages_purged;
 
-	} while (!srv_purge_should_exit(thd, n_pages_purged)
+	} while (!srv_purge_should_exit(n_pages_purged)
 		 && n_pages_purged > 0
 		 && purge_sys->state == PURGE_STATE_RUN);
 
@@ -2790,7 +2778,6 @@ static
 void
 srv_purge_coordinator_suspend(
 /*==========================*/
-	MYSQL_THD	thd,
 	srv_slot_t*	slot,			/*!< in/out: Purge coordinator
 						thread slot */
 	ulint		rseg_history_len)	/*!< in: history list length
@@ -2852,7 +2839,7 @@ srv_purge_coordinator_suspend(
 		}
 
 		rw_lock_x_unlock(&purge_sys->latch);
-	} while (stop && !thd_kill_level(thd));
+	} while (stop && srv_undo_sources);
 
 	srv_resume_thread(slot, 0, false);
 }
@@ -2902,51 +2889,23 @@ DECLARE_THREAD(srv_purge_coordinator_thread)(
 		purge didn't purge any records then wait for activity. */
 
 		if (srv_shutdown_state == SRV_SHUTDOWN_NONE
-		    && !thd_kill_level(thd)
+		    && srv_undo_sources
 		    && (purge_sys->state == PURGE_STATE_STOP
 			|| n_total_purged == 0)) {
 
-			srv_purge_coordinator_suspend(thd, slot, rseg_history_len);
+			srv_purge_coordinator_suspend(slot, rseg_history_len);
 		}
 
 		ut_ad(!slot->suspended);
 
-		if (srv_purge_should_exit(thd, n_total_purged)) {
+		if (srv_purge_should_exit(n_total_purged)) {
 			break;
 		}
 
 		n_total_purged = 0;
 
-		rseg_history_len = srv_do_purge(
-			thd, srv_n_purge_threads, &n_total_purged);
-
-	} while (!srv_purge_should_exit(thd, n_total_purged));
-
-	/* Ensure that we don't jump out of the loop unless the
-	exit condition is satisfied. */
-
-	ut_a(srv_purge_should_exit(thd, n_total_purged));
-
-	/* Ensure that all records are purged on slow shutdown. */
-	while (srv_fast_shutdown == 0
-	       && trx_purge(1, srv_purge_batch_size, false));
-
-#ifdef UNIV_DEBUG
-	if (srv_fast_shutdown == 0) {
-		trx_commit_disallowed = true;
-	}
-#endif /* UNIV_DEBUG */
-
-	/* This trx_purge is called to remove any undo records (added by
-	background threads) after completion of the above loop. When
-	srv_fast_shutdown != 0, a large batch size can cause significant
-	delay in shutdown ,so reducing the batch size to magic number 20
-	(which was default in 5.5), which we hope will be sufficient to
-	remove all the undo records */
-
-	if (trx_purge(1, std::min(srv_purge_batch_size, 20UL), true)) {
-		ut_a(srv_fast_shutdown);
-	}
+		rseg_history_len = srv_do_purge(&n_total_purged);
+	} while (!srv_purge_should_exit(n_total_purged));
 
 	/* The task queue should always be empty, independent of fast
 	shutdown state. */
diff --git a/storage/innobase/srv/srv0start.cc b/storage/innobase/srv/srv0start.cc
index 99843719008..0398e36eaf8 100644
--- a/storage/innobase/srv/srv0start.cc
+++ b/storage/innobase/srv/srv0start.cc
@@ -139,7 +139,11 @@ bool	srv_sys_tablespaces_open;
 /** TRUE if the server was successfully started */
 bool	srv_was_started;
 /** TRUE if innobase_start_or_create_for_mysql() has been called */
-static bool	srv_start_has_been_called;
+static bool		srv_start_has_been_called;
+
+/** Whether any undo log records can be generated */
+UNIV_INTERN bool	srv_undo_sources;
+
 #ifdef UNIV_DEBUG
 /** InnoDB system tablespace to set during recovery */
 UNIV_INTERN uint	srv_sys_space_size_debug;
@@ -451,8 +455,7 @@ create_log_files(
 
 	fil_space_t*	log_space = fil_space_create(
 		"innodb_redo_log", SRV_LOG_SPACE_FIRST_ID, 0, FIL_TYPE_LOG,
-		NULL, /* innodb_encrypt_log works at a different level */
-		true /* this is create */);
+		NULL/* innodb_encrypt_log works at a different level */);
 
 	ut_a(fil_validate());
 	ut_a(log_space != NULL);
@@ -707,7 +710,7 @@ srv_undo_tablespace_open(
 
 		space = fil_space_create(
 			undo_name, space_id, FSP_FLAGS_PAGE_SSIZE(),
-			FIL_TYPE_TABLESPACE, NULL, true);
+			FIL_TYPE_TABLESPACE, NULL);
 
 		ut_a(fil_validate());
 		ut_a(space);
@@ -1451,8 +1454,7 @@ Starts InnoDB and creates a new database if database files
 are not found and the user wants.
 @return DB_SUCCESS or error code */
 dberr_t
-innobase_start_or_create_for_mysql(void)
-/*====================================*/
+innobase_start_or_create_for_mysql()
 {
 	bool		create_new_db = false;
 	lsn_t		flushed_lsn;
@@ -1854,6 +1856,7 @@ innobase_start_or_create_for_mysql(void)
 	if (!srv_read_only_mode) {
 		buf_flush_page_cleaner_init();
 
+		buf_page_cleaner_is_active = true;
 		os_thread_create(buf_flush_page_cleaner_coordinator,
 				 NULL, NULL);
 
@@ -1864,11 +1867,6 @@ innobase_start_or_create_for_mysql(void)
 					 NULL, NULL);
 		}
 
-		/* Make sure page cleaner is active. */
-		while (!buf_page_cleaner_is_active) {
-			os_thread_sleep(10000);
-		}
-
 		srv_start_state_set(SRV_START_STATE_IO);
 	}
 
@@ -2070,8 +2068,7 @@ innobase_start_or_create_for_mysql(void)
 			"innodb_redo_log",
 			SRV_LOG_SPACE_FIRST_ID, 0,
 			FIL_TYPE_LOG,
-			NULL /* no encryption yet */,
-			true /* create */);
+			NULL /* no encryption yet */);
 
 		ut_a(fil_validate());
 		ut_a(log_space);
@@ -2499,6 +2496,7 @@ files_checked:
 		}
 
 		recv_recovery_rollback_active();
+		srv_startup_is_before_trx_rollback_phase = FALSE;
 
 		/* It is possible that file_format tag has never
 		been set. In this case we initialize it to minimum
@@ -2611,6 +2609,16 @@ files_checked:
 			NULL, thread_ids + (1 + SRV_MAX_N_IO_THREADS));
 		thread_started[1 + SRV_MAX_N_IO_THREADS] = true;
 		srv_start_state_set(SRV_START_STATE_MASTER);
+
+		srv_undo_sources = true;
+		/* Create the dict stats gathering thread */
+		srv_dict_stats_thread_active = true;
+		dict_stats_thread_handle = os_thread_create(
+			dict_stats_thread, NULL, NULL);
+		dict_stats_thread_started = true;
+
+		/* Create the thread that will optimize the FTS sub-system. */
+		fts_optimize_init();
 	}
 
 	if (!srv_read_only_mode
@@ -2691,10 +2699,10 @@ files_checked:
 		if (!wsrep_recovery) {
 #endif /* WITH_WSREP */
 		/* Create the buffer pool dump/load thread */
+		srv_buf_dump_thread_active = true;
 		buf_dump_thread_handle=
 			os_thread_create(buf_dump_thread, NULL, NULL);
 
-		srv_buf_dump_thread_active = true;
 		buf_dump_thread_started = true;
 #ifdef WITH_WSREP
 		} else {
@@ -2719,15 +2727,6 @@ files_checked:
 		*/
 		log_make_checkpoint_at(LSN_MAX, TRUE);
 
-		/* Create the dict stats gathering thread */
-		dict_stats_thread_handle = os_thread_create(
-			dict_stats_thread, NULL, NULL);
-		srv_dict_stats_thread_active = true;
-		dict_stats_thread_started = true;
-
-		/* Create the thread that will optimize the FTS sub-system. */
-		fts_optimize_init();
-
 		/* Init data for datafile scrub threads */
 		btr_scrub_init();
 
@@ -2779,13 +2778,17 @@ srv_fts_close(void)
 /****************************************************************//**
 Shuts down background threads that can generate undo pages. */
 void
-srv_shutdown_bg_undo_sources(void)
-/*===========================*/
+srv_shutdown_bg_undo_sources()
 {
-	if (srv_start_state_is_set(SRV_START_STATE_STAT)) {
+	if (srv_undo_sources) {
 		ut_ad(!srv_read_only_mode);
 		fts_optimize_shutdown();
 		dict_stats_shutdown();
+		while (row_get_background_drop_list_len_low()) {
+			srv_wake_master_thread();
+			os_thread_yield();
+		}
+		srv_undo_sources = false;
 	}
 }
 
@@ -2795,9 +2798,7 @@ innodb_shutdown()
 {
 	ut_ad(!srv_running);
 
-	if (srv_fast_shutdown) {
-		srv_shutdown_bg_undo_sources();
-	}
+	srv_shutdown_bg_undo_sources();
 
 	/* 1. Flush the buffer pool to disk, write the current lsn to
 	the tablespace header(s), and copy all log data to archive.
@@ -2943,85 +2944,6 @@ innodb_shutdown()
 	srv_start_has_been_called = false;
 }
 
-#if 0 // TODO: Enable this in WL#6608
-/********************************************************************
-Signal all per-table background threads to shutdown, and wait for them to do
-so. */
-static
-void
-srv_shutdown_table_bg_threads(void)
-/*===============================*/
-{
-	dict_table_t*	table;
-	dict_table_t*	first;
-	dict_table_t*	last = NULL;
-
-	mutex_enter(&dict_sys->mutex);
-
-	/* Signal all threads that they should stop. */
-	table = UT_LIST_GET_FIRST(dict_sys->table_LRU);
-	first = table;
-	while (table) {
-		dict_table_t*	next;
-		fts_t*		fts = table->fts;
-
-		if (fts != NULL) {
-			fts_start_shutdown(table, fts);
-		}
-
-		next = UT_LIST_GET_NEXT(table_LRU, table);
-
-		if (!next) {
-			last = table;
-		}
-
-		table = next;
-	}
-
-	/* We must release dict_sys->mutex here; if we hold on to it in the
-	loop below, we will deadlock if any of the background threads try to
-	acquire it (for example, the FTS thread by calling que_eval_sql).
-
-	Releasing it here and going through dict_sys->table_LRU without
-	holding it is safe because:
-
-	 a) MySQL only starts the shutdown procedure after all client
-	 threads have been disconnected and no new ones are accepted, so no
-	 new tables are added or old ones dropped.
-
-	 b) Despite its name, the list is not LRU, and the order stays
-	 fixed.
-
-	To safeguard against the above assumptions ever changing, we store
-	the first and last items in the list above, and then check that
-	they've stayed the same below. */
-
-	mutex_exit(&dict_sys->mutex);
-
-	/* Wait for the threads of each table to stop. This is not inside
-	the above loop, because by signaling all the threads first we can
-	overlap their shutting down delays. */
-	table = UT_LIST_GET_FIRST(dict_sys->table_LRU);
-	ut_a(first == table);
-	while (table) {
-		dict_table_t*	next;
-		fts_t*		fts = table->fts;
-
-		if (fts != NULL) {
-			fts_shutdown(table, fts);
-		}
-
-		next = UT_LIST_GET_NEXT(table_LRU, table);
-
-		if (table == last) {
-			ut_a(!next);
-		}
-
-		table = next;
-	}
-}
-#endif
-
 /** Get the meta-data filename from the table name for a
 single-table tablespace.
 @param[in]	table		table object
diff --git a/storage/innobase/trx/trx0purge.cc b/storage/innobase/trx/trx0purge.cc
index 28c41c7c935..e317a38815b 100644
--- a/storage/innobase/trx/trx0purge.cc
+++ b/storage/innobase/trx/trx0purge.cc
@@ -61,7 +61,6 @@ trx_undo_rec_t	trx_purge_dummy_rec;
 
 #ifdef UNIV_DEBUG
 my_bool		srv_purge_view_update_only_debug;
-bool		trx_commit_disallowed = false;
 #endif /* UNIV_DEBUG */
 
 /** Sentinel value */
@@ -282,7 +281,18 @@ trx_purge_add_update_undo_to_history(
 			hist_size + undo->size, MLOG_4BYTES, mtr);
 	}
 
-	ut_ad(!trx_commit_disallowed);
+	/* Before any transaction-generating background threads or the
+	purge have been started, recv_recovery_rollback_active() can
+	start transactions in row_merge_drop_temp_indexes() and
+	fts_drop_orphaned_tables(), and roll back recovered transactions.
+	After the purge thread has been given permission to exit,
+	in fast shutdown, we may roll back transactions (trx->undo_no==0)
+	in THD::cleanup() invoked from unlink_thd(). */
+	ut_ad(srv_undo_sources
+	      || ((srv_startup_is_before_trx_rollback_phase
+		   || trx_rollback_or_clean_is_active)
+		  && purge_sys->state == PURGE_STATE_INIT)
+	      || (trx->undo_no == 0 && srv_fast_shutdown));
 
 	/* Add the log as the first in the history list */
 	flst_add_first(rseg_header + TRX_RSEG_HISTORY,