Merge 10.1 into 10.2

14 files changed, 182 insertions, 266 deletions

diff --git a/storage/xtradb/buf/buf0buf.cc b/storage/xtradb/buf/buf0buf.cc
index 290712e52af..01bec11d2ed 100644
--- a/storage/xtradb/buf/buf0buf.cc
+++ b/storage/xtradb/buf/buf0buf.cc
@@ -4639,6 +4639,7 @@ buf_page_check_corrupt(buf_page_t* bpage, fil_space_t* space)
 		!bpage->encrypted &&
 		fil_space_verify_crypt_checksum(dst_frame, zip_size,
 			space, bpage->offset));
+
 	if (!still_encrypted) {
 		/* If traditional checksums match, we assume that page is
 		not anymore encrypted. */
diff --git a/storage/xtradb/buf/buf0flu.cc b/storage/xtradb/buf/buf0flu.cc
index 64b236d5688..84eea3bc692 100644
--- a/storage/xtradb/buf/buf0flu.cc
+++ b/storage/xtradb/buf/buf0flu.cc
@@ -62,10 +62,10 @@ 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 */
-UNIV_INTERN ibool buf_page_cleaner_is_active = FALSE;
+UNIV_INTERN bool buf_page_cleaner_is_active;
 
 /** Flag indicating if the lru_manager is in active state. */
-UNIV_INTERN bool buf_lru_manager_is_active = false;
+UNIV_INTERN bool buf_lru_manager_is_active;
 
 #ifdef UNIV_PFS_THREAD
 UNIV_INTERN mysql_pfs_key_t buf_page_cleaner_thread_key;
@@ -2799,8 +2799,6 @@ DECLARE_THREAD(buf_flush_page_cleaner_thread)(
 		os_thread_pf(os_thread_get_curr_id()));
 #endif /* UNIV_DEBUG_THREAD_CREATION */
 
-	buf_page_cleaner_is_active = TRUE;
-
 	while (srv_shutdown_state == SRV_SHUTDOWN_NONE) {
 
 		ulint	page_cleaner_sleep_time;
@@ -2909,7 +2907,7 @@ DECLARE_THREAD(buf_flush_page_cleaner_thread)(
 	/* We have lived our life. Time to die. */
 
 thread_exit:
-	buf_page_cleaner_is_active = FALSE;
+	buf_page_cleaner_is_active = false;
 
 	my_thread_end();
 	/* We count the number of threads in os_thread_exit(). A created
@@ -2950,8 +2948,6 @@ DECLARE_THREAD(buf_flush_lru_manager_thread)(
 		os_thread_pf(os_thread_get_curr_id()));
 #endif /* UNIV_DEBUG_THREAD_CREATION */
 
-	buf_lru_manager_is_active = true;
-
 	/* On server shutdown, the LRU manager thread runs through cleanup
 	phase to provide free pages for the master and purge threads.  */
 	while (srv_shutdown_state == SRV_SHUTDOWN_NONE
diff --git a/storage/xtradb/dict/dict0stats_bg.cc b/storage/xtradb/dict/dict0stats_bg.cc
index ec5efe0dd7c..ba6fd115551 100644
--- a/storage/xtradb/dict/dict0stats_bg.cc
+++ b/storage/xtradb/dict/dict0stats_bg.cc
@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 2012, 2017, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2017, MariaDB Corporation. All Rights Reserved.
+Copyright (c) 2017, 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
@@ -39,12 +39,18 @@ 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. */
 UNIV_INTERN os_event_t		dict_stats_event;
 
+/** Variable to initiate shutdown the dict stats thread. Note we don't
+use 'srv_shutdown_state' because we want to shutdown dict stats thread
+before purge thread. */
+static bool			dict_stats_start_shutdown;
+
+/** Event to wait for shutdown of the dict stats thread */
+static os_event_t		dict_stats_shutdown_event;
+
 /** This mutex protects the "recalc_pool" variable. */
 static ib_mutex_t		recalc_pool_mutex;
 static ib_mutex_t		defrag_pool_mutex;
@@ -341,11 +347,11 @@ Must be called before dict_stats_thread() is started. */
 UNIV_INTERN
 void
 dict_stats_thread_init()
-/*====================*/
 {
 	ut_a(!srv_read_only_mode);
 
 	dict_stats_event = os_event_create();
+	dict_stats_shutdown_event = os_event_create();
 
 	/* The recalc_pool_mutex is acquired from:
 	1) the background stats gathering thread before any other latch
@@ -390,6 +396,9 @@ dict_stats_thread_deinit()
 
 	os_event_free(dict_stats_event);
 	dict_stats_event = NULL;
+	os_event_free(dict_stats_shutdown_event);
+	dict_stats_shutdown_event = NULL;
+	dict_stats_start_shutdown = false;
 }
 
 /*****************************************************************//**
@@ -533,7 +542,7 @@ DECLARE_THREAD(dict_stats_thread)(void*)
 	my_thread_init();
 	ut_a(!srv_read_only_mode);
 
-	while (!SHUTTING_DOWN()) {
+	while (!dict_stats_start_shutdown) {
 
 		/* Wake up periodically even if not signaled. This is
 		because we may lose an event - if the below call to
@@ -543,7 +552,7 @@ DECLARE_THREAD(dict_stats_thread)(void*)
 		os_event_wait_time(
 			dict_stats_event, MIN_RECALC_INTERVAL * 1000000);
 
-		if (SHUTTING_DOWN()) {
+		if (dict_stats_start_shutdown) {
 			break;
 		}
 
@@ -557,6 +566,7 @@ DECLARE_THREAD(dict_stats_thread)(void*)
 
 	srv_dict_stats_thread_active = false;
 
+	os_event_set(dict_stats_shutdown_event);
 	my_thread_end();
 	/* We count the number of threads in os_thread_exit(). A created
 	thread should always use that to exit instead of return(). */
@@ -564,3 +574,12 @@ DECLARE_THREAD(dict_stats_thread)(void*)
 
 	OS_THREAD_DUMMY_RETURN;
 }
+
+/** Shut down the dict_stats_thread. */
+void
+dict_stats_shutdown()
+{
+	dict_stats_start_shutdown = true;
+	os_event_set(dict_stats_event);
+	os_event_wait(dict_stats_shutdown_event);
+}
diff --git a/storage/xtradb/fil/fil0crypt.cc b/storage/xtradb/fil/fil0crypt.cc
index 46ed7d37e42..e73d600d2ca 100644
--- a/storage/xtradb/fil/fil0crypt.cc
+++ b/storage/xtradb/fil/fil0crypt.cc
@@ -1115,6 +1115,36 @@ 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;
+	}
+
+	mtr_t	mtr;
+	mtr_start(&mtr);
+	ulint zip_size = fsp_flags_get_zip_size(space->flags);
+	ulint offset = fsp_header_get_crypt_offset(zip_size);
+	if (buf_block_t* block = buf_page_get(space->id, zip_size, 0,
+					      RW_S_LATCH, &mtr)) {
+		mutex_enter(&fil_system->mutex);
+		if (!space->crypt_data) {
+			space->crypt_data = fil_space_read_crypt_data(
+				space->id, block->frame, offset);
+		}
+		mutex_exit(&fil_system->mutex);
+	}
+
+	mtr_commit(&mtr);
+}
+
 /***********************************************************************
 Start encrypting a space
 @param[in,out]		space		Tablespace
@@ -1125,6 +1155,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;
@@ -1191,8 +1222,6 @@ fil_crypt_start_encrypting_space(
 		byte* frame = buf_block_get_frame(block);
 		crypt_data->type = CRYPT_SCHEME_1;
 		crypt_data->write_page0(frame, &mtr);
-
-
 		mtr_commit(&mtr);
 
 		/* record lsn of update */
@@ -1620,6 +1649,8 @@ fil_crypt_find_space_to_rotate(
 	}
 
 	while (!state->should_shutdown() && state->space) {
+		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
@@ -2314,8 +2345,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. */
@@ -2545,10 +2578,10 @@ fil_space_crypt_get_status(
 	memset(status, 0, sizeof(*status));
 
 	ut_ad(space->n_pending_ops > 0);
-	fil_space_crypt_t* crypt_data = space->crypt_data;
+	fil_crypt_read_crypt_data(const_cast<fil_space_t*>(space));
 	status->space = space->id;
 
-	if (crypt_data != NULL) {
+	if (fil_space_crypt_t* crypt_data = space->crypt_data) {
 		mutex_enter(&crypt_data->mutex);
 		status->scheme = crypt_data->type;
 		status->keyserver_requests = crypt_data->keyserver_requests;
diff --git a/storage/xtradb/fil/fil0fil.cc b/storage/xtradb/fil/fil0fil.cc
index d979c05c9a6..12048bc479f 100644
--- a/storage/xtradb/fil/fil0fil.cc
+++ b/storage/xtradb/fil/fil0fil.cc
@@ -661,12 +661,10 @@ fil_node_open_file(
 
 		/* Try to read crypt_data from page 0 if it is not yet
 		read. */
-		if (!node->space->page_0_crypt_read) {
-			ulint offset = fsp_header_get_crypt_offset(
-				fsp_flags_get_zip_size(flags));
-			ut_ad(node->space->crypt_data == NULL);
+		if (!node->space->crypt_data) {
+			const ulint offset = fsp_header_get_crypt_offset(
+					fsp_flags_get_zip_size(flags));
 			node->space->crypt_data = fil_space_read_crypt_data(space_id, page, offset);
-			node->space->page_0_crypt_read = true;
 		}
 
 		ut_free(buf2);
@@ -1600,22 +1598,6 @@ 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;
-	}
-
-#ifdef UNIV_DEBUG
-	ib_logf(IB_LOG_LEVEL_INFO,
-		"Created tablespace for space %lu name %s key_id %u encryption %d.",
-		space->id,
-		space->name,
-		space->crypt_data ? space->crypt_data->key_id : 0,
-		space->crypt_data ? space->crypt_data->encryption : 0);
-#endif
-
 	rw_lock_create(fil_space_latch_key, &space->latch, SYNC_FSP);
 
 	HASH_INSERT(fil_space_t, hash, fil_system->spaces, id, space);
@@ -2463,8 +2445,8 @@ fil_read_first_page(
 		/* Possible encryption crypt data is also stored only to first page
 		of the first datafile. */
 
-		ulint offset = fsp_header_get_crypt_offset(
-			fsp_flags_get_zip_size(*flags));
+		const ulint offset = fsp_header_get_crypt_offset(
+					fsp_flags_get_zip_size(*flags));
 
 		cdata = fil_space_read_crypt_data(*space_id, page, offset);
 
@@ -4211,6 +4193,7 @@ fsp_flags_try_adjust(ulint space_id, ulint flags)
 					 flags, MLOG_4BYTES, &mtr);
 		}
 	}
+
 	mtr_commit(&mtr);
 }
 
@@ -4631,7 +4614,17 @@ cleanup_and_exit:
 
 	mem_free(def.filepath);
 
-	if (err == DB_SUCCESS && !srv_read_only_mode) {
+	/* We need to check fsp flags when no errors has happened and
+	server was not started on read only mode and tablespace validation
+	was requested or flags contain other table options except
+	low order bits to FSP_FLAGS_POS_PAGE_SSIZE position.
+	Note that flag comparison is pessimistic. Adjust is required
+	only when flags contain buggy MariaDB 10.1.0 -
+	MariaDB 10.1.20 flags.  */
+	if (err == DB_SUCCESS
+	    && !srv_read_only_mode
+	    && (validate
+		|| flags >= (1U << FSP_FLAGS_POS_PAGE_SSIZE))) {
 		fsp_flags_try_adjust(id, flags & ~FSP_FLAGS_MEM_MASK);
 	}
 
diff --git a/storage/xtradb/fsp/fsp0fsp.cc b/storage/xtradb/fsp/fsp0fsp.cc
index bd87b88f58d..df8c6ffe222 100644
--- a/storage/xtradb/fsp/fsp0fsp.cc
+++ b/storage/xtradb/fsp/fsp0fsp.cc
@@ -2740,7 +2740,6 @@ fsp_reserve_free_extents(
 	ibool		success;
 	ulint		n_pages_added;
 	size_t		total_reserved = 0;
-	ulint		rounds = 0;
 
 	ut_ad(mtr);
 	*n_reserved = n_ext;
@@ -2819,17 +2818,7 @@ try_to_extend:
 	success = fsp_try_extend_data_file(&n_pages_added, space,
 					   space_header, mtr);
 	if (success && n_pages_added > 0) {
-
-		rounds++;
 		total_reserved += n_pages_added;
-
-		if (rounds > 50) {
-			ib_logf(IB_LOG_LEVEL_INFO,
-				"Space id %lu trying to reserve %lu extents actually reserved %lu "
-				" reserve %lu free %lu size %lu rounds %lu total_reserved %llu",
-				space, n_ext, n_pages_added, reserve, n_free, size, rounds, (ullint) total_reserved);
-		}
-
 		goto try_again;
 	}
 
@@ -4150,20 +4139,8 @@ ulint
 fsp_header_get_crypt_offset(
 	const ulint   zip_size)
 {
-	ulint pageno = 0;
-	/* compute first page_no that will have xdes stored on page != 0*/
-	for (ulint i = 0;
-	     (pageno = xdes_calc_descriptor_page(zip_size, i)) == 0; )
-		i++;
-
-	/* use pageno prior to this...i.e last page on page 0 */
-	ut_ad(pageno > 0);
-	pageno--;
-
-	ulint iv_offset = XDES_ARR_OFFSET +
-		XDES_SIZE * (1 + xdes_calc_descriptor_index(zip_size, pageno));
-
-	return FSP_HEADER_OFFSET + iv_offset;
+	return (FSP_HEADER_OFFSET + (XDES_ARR_OFFSET + XDES_SIZE *
+			(zip_size ? zip_size : UNIV_PAGE_SIZE) / FSP_EXTENT_SIZE));
 }
 
 /**********************************************************************//**
diff --git a/storage/xtradb/handler/ha_innodb.cc b/storage/xtradb/handler/ha_innodb.cc
index 6b404b68dc9..264bb31db64 100644
--- a/storage/xtradb/handler/ha_innodb.cc
+++ b/storage/xtradb/handler/ha_innodb.cc
@@ -1615,14 +1615,11 @@ innobase_drop_database(
 					the path is used as the database name:
 					for example, in 'mysql/data/test' the
 					database name is 'test' */
-/*******************************************************************//**
-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);
 
 #if NOT_USED
 /*****************************************************************//**
@@ -4498,21 +4495,13 @@ error:
 	DBUG_RETURN(TRUE);
 }
 
-/*******************************************************************//**
-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 (innodb_inited) {
 
@@ -4529,9 +4518,7 @@ innobase_end(
 		innodb_inited = 0;
 		hash_table_free(innobase_open_tables);
 		innobase_open_tables = NULL;
-		if (innobase_shutdown_for_mysql() != DB_SUCCESS) {
-			err = 1;
-		}
+		innodb_shutdown();
 		srv_free_paths_and_sizes();
 		my_free(internal_innobase_data_file_path);
 		mysql_mutex_destroy(&innobase_share_mutex);
@@ -4540,7 +4527,7 @@ innobase_end(
 		mysql_mutex_destroy(&pending_checkpoint_mutex);
 	}
 
-	DBUG_RETURN(err);
+	DBUG_RETURN(0);
 }
 
 /****************************************************************//**
@@ -18823,6 +18810,10 @@ innodb_sched_priority_cleaner_update(
 	const void*			save)	/*!< in: immediate result
 						from check function */
 {
+	if (srv_read_only_mode) {
+		return;
+	}
+
 	ulint	priority = *static_cast<const ulint *>(save);
 	ulint	actual_priority;
 	ulint	nice = 0;
@@ -18849,10 +18840,6 @@ innodb_sched_priority_cleaner_update(
 	}
 
 	/* Set the priority for the page cleaner thread */
-	if (srv_read_only_mode) {
-
-		return;
-	}
 
 	ut_ad(buf_page_cleaner_is_active);
 	nice = os_thread_get_priority(srv_cleaner_tid);
diff --git a/storage/xtradb/include/buf0flu.h b/storage/xtradb/include/buf0flu.h
index af50a5498ef..6089baf81e8 100644
--- a/storage/xtradb/include/buf0flu.h
+++ b/storage/xtradb/include/buf0flu.h
@@ -34,7 +34,7 @@ Created 11/5/1995 Heikki Tuuri
 #include "buf0types.h"
 
 /** Flag indicating if the page_cleaner is in active state. */
-extern ibool buf_page_cleaner_is_active;
+extern bool buf_page_cleaner_is_active;
 
 /** Flag indicating if the lru_manager is in active state. */
 extern bool buf_lru_manager_is_active;
diff --git a/storage/xtradb/include/dict0stats_bg.h b/storage/xtradb/include/dict0stats_bg.h
index d5f0870718d..8f3385eb22b 100644
--- a/storage/xtradb/include/dict0stats_bg.h
+++ b/storage/xtradb/include/dict0stats_bg.h
@@ -1,7 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 2012, 2016, Oracle and/or its affiliates. All Rights Reserved.
-Copyright (c) 2017, MariaDB Corporation. All Rights Reserved.
+Copyright (c) 2017, 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
@@ -144,6 +144,10 @@ DECLARE_THREAD(dict_stats_thread)(
 	void*	arg);	/*!< in: a dummy parameter
 			required by os_thread_create */
 
+/** Shut down the dict_stats_thread. */
+void
+dict_stats_shutdown();
+
 # ifndef UNIV_NONINL
 #  include "dict0stats_bg.ic"
 # endif
diff --git a/storage/xtradb/include/fil0fil.h b/storage/xtradb/include/fil0fil.h
index b861225f562..a09833c3a73 100644
--- a/storage/xtradb/include/fil0fil.h
+++ b/storage/xtradb/include/fil0fil.h
@@ -350,9 +350,6 @@ struct fil_space_t {
 				compression failure */
 	fil_space_crypt_t* crypt_data;
 				/*!< tablespace crypt data or NULL */
-	bool		page_0_crypt_read;
-				/*!< tablespace crypt data has been
-				read */
 	ulint		file_block_size;
 				/*!< file system block size */
 
diff --git a/storage/xtradb/include/srv0start.h b/storage/xtradb/include/srv0start.h
index a60776a4665..b055a9d834f 100644
--- a/storage/xtradb/include/srv0start.h
+++ b/storage/xtradb/include/srv0start.h
@@ -1,6 +1,7 @@
 /*****************************************************************************
 
 Copyright (c) 1995, 2017, Oracle and/or its affiliates. All Rights Reserved.
+Copyright (c) 2017, 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
@@ -75,22 +76,12 @@ are not found and the user wants.
 @return	DB_SUCCESS or error code */
 UNIV_INTERN
 dberr_t
-innobase_start_or_create_for_mysql(void);
-/*====================================*/
-/****************************************************************//**
-Shuts down the Innobase database.
-@return	DB_SUCCESS or error code */
-UNIV_INTERN
-dberr_t
-innobase_shutdown_for_mysql(void);
+innobase_start_or_create_for_mysql();
 
-/********************************************************************
-Signal all per-table background threads to shutdown, and wait for them to do
-so. */
+/** Shut down InnoDB. */
 UNIV_INTERN
 void
-srv_shutdown_table_bg_threads(void);
-/*=============================*/
+innodb_shutdown();
 
 /*************************************************************//**
 Copy the file path component of the physical file to parameter. It will
@@ -158,6 +149,9 @@ enum srv_shutdown_state {
 	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_state	srv_shutdown_state;
diff --git a/storage/xtradb/srv/srv0srv.cc b/storage/xtradb/srv/srv0srv.cc
index 88ca7fc263b..cc5d1320142 100644
--- a/storage/xtradb/srv/srv0srv.cc
+++ b/storage/xtradb/srv/srv0srv.cc
@@ -3209,31 +3209,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(
-/*==============*/
-	ulint		n_purged)	/*!< in: pages purged in last batch */
+srv_purge_should_exit(ulint n_purged)
 {
-	switch (srv_shutdown_state) {
-	case SRV_SHUTDOWN_NONE:
-		/* Normal operation. */
-		break;
+	ut_ad(srv_shutdown_state == SRV_SHUTDOWN_NONE
+	      || srv_shutdown_state == SRV_SHUTDOWN_CLEANUP);
 
-	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);
-
-	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);
 }
 
 /*********************************************************************//**
@@ -3508,7 +3506,7 @@ srv_purge_coordinator_suspend(
 		}
 
 		rw_lock_x_unlock(&purge_sys->latch);
-	} while (stop);
+	} while (stop && srv_undo_sources);
 
 	srv_resume_thread(slot, 0, false);
 }
@@ -3562,6 +3560,7 @@ DECLARE_THREAD(srv_purge_coordinator_thread)(
 		purge didn't purge any records then wait for activity. */
 
 		if (srv_shutdown_state == SRV_SHUTDOWN_NONE
+		    && srv_undo_sources
 		    && (purge_sys->state == PURGE_STATE_STOP
 			|| n_total_purged == 0)) {
 
@@ -3582,36 +3581,8 @@ DECLARE_THREAD(srv_purge_coordinator_thread)(
 			srv_n_purge_threads, &n_total_purged);
 
 		srv_inc_activity_count();
-
 	} while (!srv_purge_should_exit(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(n_total_purged));
-
-	ulint	n_pages_purged = ULINT_MAX;
-
-	/* Ensure that all records are purged if it is not a fast shutdown.
-	This covers the case where a record can be added after we exit the
-	loop above. */
-	while (srv_fast_shutdown == 0 && n_pages_purged > 0) {
-		n_pages_purged = trx_purge(1, srv_purge_batch_size, false);
-	}
-
-	/* 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 */
-	const	uint temp_batch_size = 20;
-
-	n_pages_purged = trx_purge(1, srv_purge_batch_size <= temp_batch_size
-				      ? srv_purge_batch_size : temp_batch_size,
-				   true);
-	ut_a(n_pages_purged == 0 || srv_fast_shutdown != 0);
-
 	/* The task queue should always be empty, independent of fast
 	shutdown state. */
 	ut_a(srv_get_task_queue_length() == 0);
diff --git a/storage/xtradb/srv/srv0start.cc b/storage/xtradb/srv/srv0start.cc
index bca59c148ca..fd129c3e55f 100644
--- a/storage/xtradb/srv/srv0start.cc
+++ b/storage/xtradb/srv/srv0start.cc
@@ -132,7 +132,11 @@ UNIV_INTERN ibool	srv_is_being_started = FALSE;
 /** TRUE if the server was successfully started */
 UNIV_INTERN ibool	srv_was_started = FALSE;
 /** TRUE if innobase_start_or_create_for_mysql() has been called */
-static ibool		srv_start_has_been_called = FALSE;
+static ibool		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;
@@ -1684,8 +1688,7 @@ are not found and the user wants.
 @return	DB_SUCCESS or error code */
 UNIV_INTERN
 dberr_t
-innobase_start_or_create_for_mysql(void)
-/*====================================*/
+innobase_start_or_create_for_mysql()
 {
 	bool		create_new_db;
 	lsn_t		flushed_lsn;
@@ -2831,8 +2834,8 @@ files_checked:
 			}
 		}
 
-		srv_startup_is_before_trx_rollback_phase = FALSE;
 		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
@@ -2964,6 +2967,16 @@ files_checked:
 			srv_master_thread,
 			NULL, thread_ids + (1 + SRV_MAX_N_IO_THREADS));
 		thread_started[1 + SRV_MAX_N_IO_THREADS] = true;
+
+		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
@@ -3008,12 +3021,16 @@ files_checked:
 		}
 
 
-		buf_flush_page_cleaner_thread_handle = os_thread_create(buf_flush_page_cleaner_thread, NULL, NULL);
+		buf_page_cleaner_is_active = true;
+		buf_flush_page_cleaner_thread_handle = os_thread_create(
+			buf_flush_page_cleaner_thread, NULL, NULL);
 		buf_flush_page_cleaner_thread_started = true;
-	}
 
-	buf_flush_lru_manager_thread_handle = os_thread_create(buf_flush_lru_manager_thread, NULL, NULL);
-	buf_flush_lru_manager_thread_started = true;
+		buf_lru_manager_is_active = true;
+		buf_flush_lru_manager_thread_handle = os_thread_create(
+			buf_flush_lru_manager_thread, NULL, NULL);
+		buf_flush_lru_manager_thread_started = true;
+	}
 
 	if (!srv_file_per_table && srv_pass_corrupt_table) {
 		fprintf(stderr, "InnoDB: Warning:"
@@ -3061,10 +3078,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 {
@@ -3074,15 +3091,6 @@ files_checked:
 		}
 #endif /* WITH_WSREP */
 
-		/* 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();
-
 		/* Create thread(s) that handles key rotation */
 		fil_system_enter();
 		fil_crypt_threads_init();
@@ -3132,13 +3140,10 @@ srv_fts_close(void)
 }
 #endif
 
-/****************************************************************//**
-Shuts down the InnoDB database.
-@return	DB_SUCCESS or error code */
+/** Shut down InnoDB. */
 UNIV_INTERN
-dberr_t
-innobase_shutdown_for_mysql(void)
-/*=============================*/
+void
+innodb_shutdown()
 {
 	ulint	i;
 
@@ -3148,15 +3153,20 @@ innobase_shutdown_for_mysql(void)
 				"Shutting down an improperly started, "
 				"or created database!");
 		}
-
-		return(DB_SUCCESS);
 	}
 
-	if (!srv_read_only_mode) {
+	if (srv_undo_sources) {
+		ut_ad(!srv_read_only_mode);
 		/* Shutdown the FTS optimize sub system. */
 		fts_optimize_start_shutdown();
 
 		fts_optimize_end();
+		dict_stats_shutdown();
+		while (row_get_background_drop_list_len_low()) {
+			srv_wake_master_thread();
+			os_thread_yield();
+		}
+		srv_undo_sources = false;
 	}
 
 	/* 1. Flush the buffer pool to disk, write the current lsn to
@@ -3371,88 +3381,9 @@ innobase_shutdown_for_mysql(void)
 	srv_start_has_been_called = FALSE;
 	/* reset io_tid_i, in case current process does second innodb start (xtrabackup might do that).*/
 	io_tid_i = 0;
-	return(DB_SUCCESS);
 }
 #endif /* !UNIV_HOTBACKUP */
 
-
-/********************************************************************
-Signal all per-table background threads to shutdown, and wait for them to do
-so. */
-UNIV_INTERN
-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;
-	}
-}
-
 /*****************************************************************//**
 Get the meta-data filename from the table name. */
 UNIV_INTERN
diff --git a/storage/xtradb/trx/trx0purge.cc b/storage/xtradb/trx/trx0purge.cc
index 90f22e7ef82..df4a3217820 100644
--- a/storage/xtradb/trx/trx0purge.cc
+++ b/storage/xtradb/trx/trx0purge.cc
@@ -247,6 +247,19 @@ trx_purge_add_update_undo_to_history(
 			hist_size + undo->size, MLOG_4BYTES, mtr);
 	}
 
+	/* 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,
 		       undo_header + TRX_UNDO_HISTORY_NODE, mtr);